feat: add ConnectionManager in-memory activity state and ready payload
Add Rich Presence infrastructure to ConnectionManager: - Four new Maps: userActivities, userShowActivity, userStatuses, lastActivityUpdate - Public accessors: set/get/clear activities, showActivity, status, rate limit check - Cleanup in finalizeDisconnect() and forceDisconnectUser() - Cache showActivity and status from DB at auth time in buildReadyPayload() - Assemble userActivities snapshot in ready payload with customStatus fallback - Include activities: [] in offline presence_update broadcast
This commit is contained in:
@@ -18,6 +18,7 @@ import type {
|
|||||||
SpaceLayoutItem,
|
SpaceLayoutItem,
|
||||||
ReadState,
|
ReadState,
|
||||||
ActiveCallInfo,
|
ActiveCallInfo,
|
||||||
|
Activity,
|
||||||
} from '@backspace/shared';
|
} from '@backspace/shared';
|
||||||
import { sanitizeUser } from '../utils/sanitize.js';
|
import { sanitizeUser } from '../utils/sanitize.js';
|
||||||
|
|
||||||
@@ -92,6 +93,16 @@ class ConnectionManager {
|
|||||||
// Per-user WebSocket rate limiters (shared across all tabs/connections)
|
// Per-user WebSocket rate limiters (shared across all tabs/connections)
|
||||||
private userRateLimiters: Map<string, WsRateLimiter> = new Map();
|
private userRateLimiters: Map<string, WsRateLimiter> = new Map();
|
||||||
|
|
||||||
|
// ─── Rich Presence ──────────────────────────────────────────────────────
|
||||||
|
// userId → Activity[] (ephemeral, same lifecycle as voiceUserStates)
|
||||||
|
private userActivities: Map<string, Activity[]> = new Map();
|
||||||
|
// userId → boolean (cached from DB at auth time, updated via REST)
|
||||||
|
private userShowActivity: Map<string, boolean> = new Map();
|
||||||
|
// userId → status string (cached at auth, updated on presence_update)
|
||||||
|
private userStatuses: Map<string, string> = new Map();
|
||||||
|
// userId → timestamp of last activity_update (rate limiting)
|
||||||
|
private lastActivityUpdate: Map<string, number> = new Map();
|
||||||
|
|
||||||
addConnection(userId: string, ws: WebSocket): void {
|
addConnection(userId: string, ws: WebSocket): void {
|
||||||
if (!this.connections.has(userId)) {
|
if (!this.connections.has(userId)) {
|
||||||
this.connections.set(userId, new Set());
|
this.connections.set(userId, new Set());
|
||||||
@@ -193,6 +204,12 @@ class ConnectionManager {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Clear activity state
|
||||||
|
this.clearUserActivities(userId);
|
||||||
|
this.userShowActivity.delete(userId);
|
||||||
|
this.userStatuses.delete(userId);
|
||||||
|
this.lastActivityUpdate.delete(userId);
|
||||||
|
|
||||||
// Broadcast offline to all spaces
|
// Broadcast offline to all spaces
|
||||||
const userSpaces = this.getUserSpaces(userId);
|
const userSpaces = this.getUserSpaces(userId);
|
||||||
for (const spaceId of userSpaces) {
|
for (const spaceId of userSpaces) {
|
||||||
@@ -200,6 +217,7 @@ class ConnectionManager {
|
|||||||
type: 'presence_update',
|
type: 'presence_update',
|
||||||
userId: userId,
|
userId: userId,
|
||||||
status: 'offline',
|
status: 'offline',
|
||||||
|
activities: [] as Activity[],
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -228,6 +246,48 @@ class ConnectionManager {
|
|||||||
return limiter;
|
return limiter;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ─── Activity accessors ─────────────────────────────────────────────────
|
||||||
|
|
||||||
|
setUserActivities(userId: string, activities: Activity[]): void {
|
||||||
|
if (activities.length === 0) {
|
||||||
|
this.userActivities.delete(userId);
|
||||||
|
} else {
|
||||||
|
this.userActivities.set(userId, activities);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
getUserActivities(userId: string): Activity[] {
|
||||||
|
return this.userActivities.get(userId) ?? [];
|
||||||
|
}
|
||||||
|
|
||||||
|
clearUserActivities(userId: string): void {
|
||||||
|
this.userActivities.delete(userId);
|
||||||
|
}
|
||||||
|
|
||||||
|
setUserShowActivity(userId: string, show: boolean): void {
|
||||||
|
this.userShowActivity.set(userId, show);
|
||||||
|
}
|
||||||
|
|
||||||
|
getUserShowActivity(userId: string): boolean {
|
||||||
|
return this.userShowActivity.get(userId) ?? true;
|
||||||
|
}
|
||||||
|
|
||||||
|
setUserStatus(userId: string, status: string): void {
|
||||||
|
this.userStatuses.set(userId, status);
|
||||||
|
}
|
||||||
|
|
||||||
|
getUserStatus(userId: string): string {
|
||||||
|
return this.userStatuses.get(userId) ?? 'offline';
|
||||||
|
}
|
||||||
|
|
||||||
|
checkActivityRateLimit(userId: string): boolean {
|
||||||
|
const now = Date.now();
|
||||||
|
const last = this.lastActivityUpdate.get(userId) ?? 0;
|
||||||
|
if (now - last < 3000) return false;
|
||||||
|
this.lastActivityUpdate.set(userId, now);
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
|
||||||
setUserSpaces(userId: string, spaceIds: string[]): void {
|
setUserSpaces(userId: string, spaceIds: string[]): void {
|
||||||
this.userSpaces.set(userId, new Set(spaceIds));
|
this.userSpaces.set(userId, new Set(spaceIds));
|
||||||
}
|
}
|
||||||
@@ -594,6 +654,12 @@ class ConnectionManager {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Clear activity state
|
||||||
|
this.clearUserActivities(userId);
|
||||||
|
this.userShowActivity.delete(userId);
|
||||||
|
this.userStatuses.delete(userId);
|
||||||
|
this.lastActivityUpdate.delete(userId);
|
||||||
|
|
||||||
// Close all WebSocket connections
|
// Close all WebSocket connections
|
||||||
const connections = this.connections.get(userId);
|
const connections = this.connections.get(userId);
|
||||||
if (connections) {
|
if (connections) {
|
||||||
@@ -674,6 +740,7 @@ function buildReadyPayload(userId: string): {
|
|||||||
spaceVoiceStates: Record<string, { spaceMuted: boolean; spaceDeafened: boolean; permissionMuted: boolean }>;
|
spaceVoiceStates: Record<string, { spaceMuted: boolean; spaceDeafened: boolean; permissionMuted: boolean }>;
|
||||||
readStates: ReadState[];
|
readStates: ReadState[];
|
||||||
activeCalls: ActiveCallInfo[];
|
activeCalls: ActiveCallInfo[];
|
||||||
|
userActivities: Record<string, Activity[]>;
|
||||||
} {
|
} {
|
||||||
const db = getDb();
|
const db = getDb();
|
||||||
|
|
||||||
@@ -684,6 +751,10 @@ function buildReadyPayload(userId: string): {
|
|||||||
}
|
}
|
||||||
const user = sanitizeUser(userRow, true);
|
const user = sanitizeUser(userRow, true);
|
||||||
|
|
||||||
|
// Cache showActivity and status for Rich Presence
|
||||||
|
connectionManager.setUserShowActivity(userId, userRow.showActivity !== 0);
|
||||||
|
connectionManager.setUserStatus(userId, (userRow.status ?? 'offline') as string);
|
||||||
|
|
||||||
// Get user's space memberships
|
// Get user's space memberships
|
||||||
const memberships = db.select()
|
const memberships = db.select()
|
||||||
.from(schema.spaceMembers)
|
.from(schema.spaceMembers)
|
||||||
@@ -1073,7 +1144,35 @@ function buildReadyPayload(userId: string): {
|
|||||||
lastReadMessageId: rs.lastReadMessageId,
|
lastReadMessageId: rs.lastReadMessageId,
|
||||||
}));
|
}));
|
||||||
|
|
||||||
return { user, spaces, dmChannels, folders, spaceLayout, layoutUpdatedAt, voiceStates, voiceUserStates, spaceVoiceStates, readStates, activeCalls };
|
// Build user activities snapshot for all visible users
|
||||||
|
// Auto-inject customStatus as a 'custom' activity for users with no ephemeral activities
|
||||||
|
const userActivities: Record<string, Activity[]> = {};
|
||||||
|
const seenUserIds = new Set<string>();
|
||||||
|
|
||||||
|
function collectUserActivities(uid: string, customStatus: string | null) {
|
||||||
|
if (seenUserIds.has(uid)) return;
|
||||||
|
seenUserIds.add(uid);
|
||||||
|
let acts = connectionManager.getUserActivities(uid);
|
||||||
|
if (acts.length === 0 && customStatus) {
|
||||||
|
acts = [{ type: 'custom', name: customStatus }];
|
||||||
|
}
|
||||||
|
if (acts.length > 0) {
|
||||||
|
userActivities[uid] = acts;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
for (const space of spaces) {
|
||||||
|
for (const member of space.members) {
|
||||||
|
collectUserActivities(member.userId, member.user?.customStatus ?? null);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
for (const dm of dmChannels) {
|
||||||
|
for (const member of dm.members) {
|
||||||
|
collectUserActivities(member.id, member.customStatus ?? null);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return { user, spaces, dmChannels, folders, spaceLayout, layoutUpdatedAt, voiceStates, voiceUserStates, spaceVoiceStates, readStates, activeCalls, userActivities };
|
||||||
}
|
}
|
||||||
|
|
||||||
export async function registerWebSocket(app: FastifyInstance): Promise<void> {
|
export async function registerWebSocket(app: FastifyInstance): Promise<void> {
|
||||||
|
|||||||
Reference in New Issue
Block a user