feat(federation): queue S2S presence_update on auth/disconnect/status/activity changes

New FederationPresenceUpdatePayload + queuePresenceRelay() helper. Five WS
sites now project the native user's status (and optional activities) to all
active peers via the outbox: WS auth-success, finalizeDisconnect,
manual presence_update, activity_update, showActivity-toggle clear.

Outbox-only (no mutation-log entry) — presence is ephemeral; the upcoming
peer-activation hook re-emits a fresh snapshot so peers recovering from
unreachable converge without history replay. No-op for replicated users.
This commit is contained in:
Jannis Braun
2026-05-05 16:01:28 +02:00
parent bdbc90ebd2
commit 613424e1c7
6 changed files with 230 additions and 1 deletions
+11
View File
@@ -447,6 +447,17 @@ export async function userRoutes(app: FastifyInstance): Promise<void> {
connectionManager.sendToSpace(spaceId, clearPayload, request.userId); connectionManager.sendToSpace(spaceId, clearPayload, request.userId);
} }
connectionManager.sendToUser(request.userId, clearPayload); connectionManager.sendToUser(request.userId, clearPayload);
// S2S: project the cleared-activities snapshot to all active peers.
void import('../utils/federationPresence.js').then(({ queuePresenceRelay }) => {
try {
queuePresenceRelay(
request.userId,
(connectionManager.getUserStatus(request.userId) ?? 'online') as 'online' | 'idle' | 'dnd' | 'offline',
[],
);
} catch (e) { console.warn('[users] queuePresenceRelay(showActivity-clear) failed', e); }
});
} }
} }
@@ -0,0 +1,120 @@
import { describe, it, expect, beforeEach, vi } from 'vitest';
import Database from 'better-sqlite3';
import { drizzle } from 'drizzle-orm/better-sqlite3';
import fs from 'node:fs';
import path from 'node:path';
import { fileURLToPath } from 'node:url';
import * as schema from '../db/schema.js';
const __dirname = path.dirname(fileURLToPath(import.meta.url));
type TestDb = ReturnType<typeof drizzle<typeof schema>>;
let sqlite: Database.Database;
let testDb: TestDb;
const queueCalls: Array<{
entityId: string;
contextId: string;
eventType: string;
payload: string;
targetPeerOrigins: string[] | undefined;
contextType: string;
}> = [];
const mutationLogCalls: Array<{ entityId: string; eventType: string }> = [];
vi.mock('../db/index.js', () => ({
getDb: () => testDb,
getRawDb: () => sqlite,
schema,
}));
vi.mock('./federationAuth.js', () => ({
getOurOrigin: () => 'https://nova.ddns.net',
}));
vi.mock('./federationOutbox.js', () => ({
isFederationRelayEnabled: () => true,
queueOutboxEvent: vi.fn((entityId, contextId, eventType, payload, targetPeerOrigins, contextType) => {
queueCalls.push({ entityId, contextId, eventType, payload, targetPeerOrigins, contextType });
}),
appendMutationLog: vi.fn((entityId, _ctxId, eventType) => {
mutationLogCalls.push({ entityId, eventType });
}),
}));
function applyMigrations(db: Database.Database): void {
const migrationsDir = path.resolve(__dirname, '../../drizzle');
const files = fs.readdirSync(migrationsDir).filter(f => f.endsWith('.sql')).sort();
for (const f of files) {
const sql = fs.readFileSync(path.join(migrationsDir, f), 'utf8');
for (const stmt of sql.split(/-->\s*statement-breakpoint/)) {
const clean = stmt.trim();
if (clean) db.exec(clean);
}
}
}
beforeEach(() => {
sqlite = new Database(':memory:');
testDb = drizzle(sqlite, { schema });
applyMigrations(sqlite);
queueCalls.length = 0;
mutationLogCalls.length = 0;
// Native local user
testDb.insert(schema.users).values({
id: 'native-1',
username: 'youruser',
passwordHash: 'x',
status: 'online',
isAdmin: 0,
homeUserId: 'native-1',
createdAt: Date.now(),
}).run();
});
describe('queuePresenceRelay', () => {
it('queues an outbox event with status + activities for a native user', async () => {
const { queuePresenceRelay } = await import('./federationPresence.js');
queuePresenceRelay('native-1', 'online', [{ type: 'playing', name: 'Test' }]);
expect(queueCalls.length).toBe(1);
const call = queueCalls[0]!;
expect(call.eventType).toBe('presence_update');
expect(call.contextType).toBe('profile');
expect(call.targetPeerOrigins).toBeUndefined(); // broadcast to all active peers
const event = JSON.parse(call.payload);
expect(event.eventType).toBe('presence_update');
expect(event.presenceUpdate.status).toBe('online');
expect(event.presenceUpdate.activities).toEqual([{ type: 'playing', name: 'Test' }]);
// appendMutationLog NOT called — presence is outbox-only
expect(mutationLogCalls).toEqual([]);
});
it('omits activities field when none are passed', async () => {
const { queuePresenceRelay } = await import('./federationPresence.js');
queuePresenceRelay('native-1', 'offline', []);
const event = JSON.parse(queueCalls[0]!.payload);
expect(event.presenceUpdate.activities).toBeUndefined();
});
it('is a no-op for replicated users (homeInstance set)', async () => {
testDb.insert(schema.users).values({
id: 'stub-1',
username: 'pbtest3@orbit.ddns.net',
passwordHash: '!federation-replicated',
status: 'online',
isAdmin: 0,
homeInstance: 'orbit.ddns.net',
homeUserId: 'remote-1',
createdAt: Date.now(),
}).run();
const { queuePresenceRelay } = await import('./federationPresence.js');
queuePresenceRelay('stub-1', 'online', []);
expect(queueCalls).toEqual([]);
});
it('is a no-op for unknown user IDs', async () => {
const { queuePresenceRelay } = await import('./federationPresence.js');
queuePresenceRelay('does-not-exist', 'online', []);
expect(queueCalls).toEqual([]);
});
});
@@ -0,0 +1,61 @@
import { eq } from 'drizzle-orm';
import type { Activity, FederationRelayEvent, FederationPresenceUpdatePayload } from '@backspace/shared';
import { getDb, schema } from '../db/index.js';
import { getOurOrigin } from './federationAuth.js';
import { isFederationRelayEnabled, queueOutboxEvent } from './federationOutbox.js';
export type PresenceStatus = 'online' | 'idle' | 'dnd' | 'offline';
/**
* Queue a presence_update event for the given native user. Broadcast to all
* active peers (mirrors profile_update). Outbox-only — presence is ephemeral;
* stale replays from a mutation log are wrong, so we never call
* appendMutationLog. The peer-activation hook re-emits a fresh snapshot for
* peer-related online natives, so a peer recovering from unreachable converges
* without history replay.
*
* No-op for replicated users (their home instance owns presence projection).
*/
export function queuePresenceRelay(
userId: string,
status: PresenceStatus,
activities: Activity[],
): void {
if (!isFederationRelayEnabled()) return;
const db = getDb();
const user = db.select().from(schema.users).where(eq(schema.users.id, userId)).get();
if (!user) return;
if (user.homeInstance) return; // replicated — not our authority
const ts = Date.now();
const payload: FederationPresenceUpdatePayload = {
homeUserId: user.id,
homeInstance: getOurOrigin(),
status,
ts,
...(activities.length > 0 ? { activities } : {}),
};
const event: FederationRelayEvent = {
eventType: 'presence_update',
contextType: 'profile',
messageId: `presence:${user.id}:${ts}`,
encryptionVersion: 0,
timestamp: ts,
presenceUpdate: payload,
};
// entityId = userId so the outbox coalesces rapid status flaps into the latest.
// contextId = userId, contextType = 'profile' (reuses existing routing).
// targetPeerOrigins = undefined → broadcast to all active peers.
// NO appendMutationLog — presence must not be replayed from history.
queueOutboxEvent(
user.id,
user.id,
'presence_update',
JSON.stringify(event),
undefined,
'profile',
);
}
+10
View File
@@ -511,6 +511,11 @@ function handlePresenceUpdate(event: Record<string, unknown>, userId: string): v
// Also send to self (other tabs) // Also send to self (other tabs)
connectionManager.sendToUser(userId, payload); connectionManager.sendToUser(userId, payload);
// S2S: project to all active peers
void import('../utils/federationPresence.js').then(({ queuePresenceRelay }) => {
try { queuePresenceRelay(userId, status as 'online' | 'idle' | 'dnd', activities); } catch (e) { console.warn('[ws] queuePresenceRelay(manual) failed', e); }
});
} }
function handleActivityUpdate(event: Record<string, unknown>, userId: string): void { function handleActivityUpdate(event: Record<string, unknown>, userId: string): void {
@@ -535,6 +540,11 @@ function handleActivityUpdate(event: Record<string, unknown>, userId: string): v
connectionManager.sendToSpace(spaceId, payload, userId); connectionManager.sendToSpace(spaceId, payload, userId);
} }
connectionManager.sendToUser(userId, payload); connectionManager.sendToUser(userId, payload);
// S2S: project to all active peers (activities + current status).
void import('../utils/federationPresence.js').then(({ queuePresenceRelay }) => {
try { queuePresenceRelay(userId, status as 'online' | 'idle' | 'dnd' | 'offline', activities); } catch (e) { console.warn('[ws] queuePresenceRelay(activity) failed', e); }
});
} }
// ─── Voice Handlers (Unified Room API) ───────────────────────────────────── // ─── Voice Handlers (Unified Room API) ─────────────────────────────────────
+12
View File
@@ -307,6 +307,12 @@ class ConnectionManager {
}); });
} }
// S2S: project offline to all active peers (mirrors profile_update fanout).
// Imported lazily to avoid circular import (federationPresence → db → ws/handler).
void import('../utils/federationPresence.js').then(({ queuePresenceRelay }) => {
try { queuePresenceRelay(userId, 'offline', []); } catch (e) { console.warn('[ws] queuePresenceRelay(offline) failed', e); }
});
// Clean up userSpaces (re-populated on next connect via setUserSpaces) // Clean up userSpaces (re-populated on next connect via setUserSpaces)
this.userSpaces.delete(userId); this.userSpaces.delete(userId);
@@ -1672,6 +1678,12 @@ export async function registerWebSocket(app: FastifyInstance): Promise<void> {
status: 'online', status: 'online',
}, userId); }, userId);
} }
// S2S: project online to all active peers (mirrors profile_update fanout).
const _uid = userId;
void import('../utils/federationPresence.js').then(({ queuePresenceRelay }) => {
try { queuePresenceRelay(_uid, 'online', []); } catch (e) { console.warn('[ws] queuePresenceRelay(online) failed', e); }
});
} catch { } catch {
ws.send(JSON.stringify({ type: 'error', message: 'Invalid token' })); ws.send(JSON.stringify({ type: 'error', message: 'Invalid token' }));
ws.close(); ws.close();
+16 -1
View File
@@ -882,7 +882,7 @@ export interface FederationRelayEvent {
| 'friend_add' | 'friend_remove' | 'file_rejected' | 'friend_add' | 'friend_remove' | 'file_rejected'
| 'dm_call_start' | 'dm_call_accept' | 'dm_call_reject' | 'dm_call_end' | 'dm_call_start' | 'dm_call_accept' | 'dm_call_reject' | 'dm_call_end'
| 'dm_typing_start' | 'dm_typing_stop' | 'dm_typing_start' | 'dm_typing_stop'
| 'profile_update' | 'profile_update' | 'presence_update'
| 'read_state_update' | 'read_state_update'
| 'dm_close' | 'dm_reopen'; | 'dm_close' | 'dm_reopen';
contextType?: 'dm' | 'friend' | 'profile'; contextType?: 'dm' | 'friend' | 'profile';
@@ -922,6 +922,7 @@ export interface FederationRelayEvent {
username: string; username: string;
}; };
profileUpdate?: FederationProfileUpdatePayload; profileUpdate?: FederationProfileUpdatePayload;
presenceUpdate?: FederationPresenceUpdatePayload;
readState?: { readState?: {
user: { homeUserId: string; homeInstance: string }; user: { homeUserId: string; homeInstance: string };
messageRef: { sourceInstance: string; sourceMessageId: string }; messageRef: { sourceInstance: string; sourceMessageId: string };
@@ -986,6 +987,20 @@ export interface FederationProfileUpdatePayload {
bio: string | null; bio: string | null;
} }
/**
* Presence projection from a home instance to peers. Carries the user's current
* online status and (optionally) rich activities. Outbox-only on the wire — never
* written to federation_mutation_log; presence is ephemeral and stale replays on
* peer activation are wrong (the activation hook re-emits a fresh snapshot).
*/
export interface FederationPresenceUpdatePayload {
homeUserId: string;
homeInstance: string;
status: 'online' | 'idle' | 'dnd' | 'offline';
activities?: Activity[];
ts: number; // emitter clock; receiver may use for last-write-wins
}
export interface FederationFriendshipPayload { export interface FederationFriendshipPayload {
from: FederationRelayParticipant; from: FederationRelayParticipant;
to: FederationRelayParticipant; to: FederationRelayParticipant;