From 1cb151d3a9f2c91b329392d66724a961196671de Mon Sep 17 00:00:00 2001 From: Jannis Braun <151788261+TheZwiss@users.noreply.github.com> Date: Tue, 5 May 2026 16:12:22 +0200 Subject: [PATCH] feat(federation): peer-lifecycle presence hooks (snapshot on activate, mark-offline on deactivate) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit onPeerActivated re-emits a presence_update for every relationship-related online native to the activating peer (relationship = friend with peer-stub / DM-mate with peer-stub / replicatedInstances opt-in for peer origin). Scope bounded by relationship count, not native count — flap recovery cost stays proportional to actual interaction surface. onPeerDeactivated flips every replicated stub from that peer to offline and broadcasts a local presence_update so connected friends/DM-mates/space-co-members see them disappear immediately, instead of seeing stale online until next signal. Re-snapshot on every activation (incl. health-check unreachable→active flap) is load-bearing for correctness — markPeerStubsOffline ran on the prior deactivation and presence is not in the mutation log. --- .../src/utils/federationPeerActivation.ts | 20 +++ .../federationPresence.peerLifecycle.test.ts | 165 +++++++++++++++++ .../server/src/utils/federationPresence.ts | 169 +++++++++++++++++- 3 files changed, 352 insertions(+), 2 deletions(-) create mode 100644 packages/server/src/utils/federationPresence.peerLifecycle.test.ts diff --git a/packages/server/src/utils/federationPeerActivation.ts b/packages/server/src/utils/federationPeerActivation.ts index 55e5cd64..da131d12 100644 --- a/packages/server/src/utils/federationPeerActivation.ts +++ b/packages/server/src/utils/federationPeerActivation.ts @@ -64,6 +64,15 @@ export async function onPeerActivated( await backfillStubUsernamesForPeer(peerRow.origin).catch((e) => { console.warn(`[onPeerActivated] backfillStubUsernamesForPeer(${peerRow.origin}) failed`, e); }); + + // Re-emit a fresh presence snapshot to the activating peer so its stubs + // of our online natives reflect current reality. Necessary because + // presence is outbox-only (no mutation-log replay), and any prior + // markPeerStubsOffline ran on our side too. + const { snapshotPresenceForPeer } = await import('./federationPresence.js'); + try { snapshotPresenceForPeer(peerRow.origin); } catch (e) { + console.warn(`[onPeerActivated] snapshotPresenceForPeer(${peerRow.origin}) failed`, e); + } } const { connectionManager } = await import('../ws/handler.js'); @@ -385,6 +394,17 @@ export async function onPeerDeactivated( ); } + // Mark every stub whose home is this peer as offline locally, and + // broadcast a presence_update WS event to friends/DM-mates/space-co-members + // so connected users see them go offline immediately, instead of seeing + // stale 'online' until the peer recovers. + try { + const { markPeerStubsOffline } = await import('./federationPresence.js'); + await markPeerStubsOffline(peer.origin); + } catch (e) { + console.warn(`[onPeerDeactivated] markPeerStubsOffline(${peer.origin}) failed`, e); + } + connectionManager.sendToAdmins({ type: 'federation_peers_changed' as const }); } catch (err) { console.error(`[federation] onPeerDeactivated(${peerId}, ${reason}) failed:`, err); diff --git a/packages/server/src/utils/federationPresence.peerLifecycle.test.ts b/packages/server/src/utils/federationPresence.peerLifecycle.test.ts new file mode 100644 index 00000000..f4dc974a --- /dev/null +++ b/packages/server/src/utils/federationPresence.peerLifecycle.test.ts @@ -0,0 +1,165 @@ +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'; +import { eq } from 'drizzle-orm'; + +const __dirname = path.dirname(fileURLToPath(import.meta.url)); +type TestDb = ReturnType>; +let sqlite: Database.Database; +let testDb: TestDb; + +const queueCalls: Array<{ entityId: string; eventType: string; targets: string[] | undefined; payload: string }> = []; +const sentToUser: Array<{ userId: string; payload: any }> = []; + +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, _ctxId, eventType, payload, targets) => { + queueCalls.push({ entityId, eventType, targets, payload }); + }), + appendMutationLog: vi.fn(), +})); + +vi.mock('../ws/handler.js', () => ({ + connectionManager: { + sendToUser: vi.fn((uid: string, p: any) => sentToUser.push({ userId: uid, payload: p })), + sendToSpace: vi.fn(), + sendToDmMembers: vi.fn(), + sendToAdmins: vi.fn(), + getAllOnlineUserIds: () => [], + evictFederatedCallsForHost: vi.fn(), + federatedCalls: new Map(), + isUserOnline: vi.fn(), + lateBindFederatedCall: vi.fn(), + }, +})); + +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; + sentToUser.length = 0; + // Stub from orbit (peer being activated) + testDb.insert(schema.users).values({ + id: 'stub-pbtest3', username: 'pbtest3@orbit.ddns.net', passwordHash: '!fr', + status: 'online', isAdmin: 0, homeInstance: 'orbit.ddns.net', + homeUserId: 'remote-pbtest3', createdAt: Date.now(), + }).run(); + // Online native FRIENDED with the stub — should be snapshotted + testDb.insert(schema.users).values({ + id: 'native-friend', username: 'youruser', passwordHash: 'x', + status: 'online', isAdmin: 0, homeUserId: 'native-friend', createdAt: Date.now(), + }).run(); + testDb.insert(schema.friends).values({ + userId: 'native-friend', friendId: 'stub-pbtest3', createdAt: Date.now(), + }).run(); + // Online native sharing a DM with the stub — should be snapshotted + testDb.insert(schema.users).values({ + id: 'native-dm', username: 'dmuser', passwordHash: 'x', + status: 'online', isAdmin: 0, homeUserId: 'native-dm', createdAt: Date.now(), + }).run(); + testDb.insert(schema.dmChannels).values({ + id: 'dm-1', ownerId: null, federatedId: null, createdAt: Date.now(), + }).run(); + testDb.insert(schema.dmMembers).values([ + { dmChannelId: 'dm-1', userId: 'native-dm', closed: 0 }, + { dmChannelId: 'dm-1', userId: 'stub-pbtest3', closed: 0 }, + ]).run(); + // Online native with replicatedInstances opt-in for orbit — should be snapshotted + testDb.insert(schema.users).values({ + id: 'native-optin', username: 'optin', passwordHash: 'x', + status: 'online', isAdmin: 0, homeUserId: 'native-optin', + replicatedInstances: JSON.stringify([{ origin: 'https://orbit.ddns.net', domain: 'orbit.ddns.net' }]), + createdAt: Date.now(), + }).run(); + // Online native with NO relationship to orbit — must NOT be snapshotted + testDb.insert(schema.users).values({ + id: 'native-unrelated', username: 'unrelated', passwordHash: 'x', + status: 'online', isAdmin: 0, homeUserId: 'native-unrelated', createdAt: Date.now(), + }).run(); + // Offline native that IS a friend of the stub — must NOT be snapshotted (offline) + testDb.insert(schema.users).values({ + id: 'native-offline-friend', username: 'sleepyfriend', passwordHash: 'x', + status: 'offline', isAdmin: 0, homeUserId: 'native-offline-friend', createdAt: Date.now(), + }).run(); + testDb.insert(schema.friends).values({ + userId: 'native-offline-friend', friendId: 'stub-pbtest3', createdAt: Date.now(), + }).run(); +}); + +describe('snapshotPresenceForPeer — scope', () => { + it('snapshots online natives that are friended with a peer stub', async () => { + const { snapshotPresenceForPeer } = await import('./federationPresence.js'); + snapshotPresenceForPeer('https://orbit.ddns.net'); + const friendCall = queueCalls.find((c) => c.entityId === 'native-friend'); + expect(friendCall).toBeDefined(); + expect(friendCall!.targets).toEqual(['https://orbit.ddns.net']); + }); + + it('snapshots online natives that share a DM with a peer stub', async () => { + const { snapshotPresenceForPeer } = await import('./federationPresence.js'); + snapshotPresenceForPeer('https://orbit.ddns.net'); + expect(queueCalls.find((c) => c.entityId === 'native-dm')).toBeDefined(); + }); + + it('snapshots online natives that opted into client-federation (replicatedInstances)', async () => { + const { snapshotPresenceForPeer } = await import('./federationPresence.js'); + snapshotPresenceForPeer('https://orbit.ddns.net'); + expect(queueCalls.find((c) => c.entityId === 'native-optin')).toBeDefined(); + }); + + it('does NOT snapshot online natives with no relationship to the peer', async () => { + const { snapshotPresenceForPeer } = await import('./federationPresence.js'); + snapshotPresenceForPeer('https://orbit.ddns.net'); + expect(queueCalls.find((c) => c.entityId === 'native-unrelated')).toBeUndefined(); + }); + + it('does NOT snapshot offline natives even when they are related to the peer', async () => { + const { snapshotPresenceForPeer } = await import('./federationPresence.js'); + snapshotPresenceForPeer('https://orbit.ddns.net'); + expect(queueCalls.find((c) => c.entityId === 'native-offline-friend')).toBeUndefined(); + }); +}); + +describe('markPeerStubsOffline', () => { + it('flips all stubs from the deactivated peer to offline and broadcasts', async () => { + const { markPeerStubsOffline } = await import('./federationPresence.js'); + await markPeerStubsOffline('https://orbit.ddns.net'); + + const stub = testDb.select().from(schema.users).where(eq(schema.users.id, 'stub-pbtest3')).get(); + expect(stub!.status).toBe('offline'); + + const friendBroadcast = sentToUser.find( + (c) => c.userId === 'native-friend' && c.payload.userId === 'stub-pbtest3', + ); + expect(friendBroadcast).toBeDefined(); + expect(friendBroadcast!.payload.status).toBe('offline'); + expect(friendBroadcast!.payload.type).toBe('presence_update'); + }); +}); diff --git a/packages/server/src/utils/federationPresence.ts b/packages/server/src/utils/federationPresence.ts index 83e3b2d9..9cf67f61 100644 --- a/packages/server/src/utils/federationPresence.ts +++ b/packages/server/src/utils/federationPresence.ts @@ -1,8 +1,10 @@ -import { eq } from 'drizzle-orm'; -import type { Activity, FederationRelayEvent, FederationPresenceUpdatePayload } from '@backspace/shared'; +import { and, eq, inArray, isNull, or } from 'drizzle-orm'; +import type { Activity, FederationRelayEvent, FederationPresenceUpdatePayload, ReplicatedInstance } from '@backspace/shared'; import { getDb, schema } from '../db/index.js'; import { getOurOrigin } from './federationAuth.js'; import { isFederationRelayEnabled, queueOutboxEvent } from './federationOutbox.js'; +import { collectProfileBroadcastTargetIds } from './userDeletion.js'; +import { extractDomain } from '../routes/federation.js'; export type PresenceStatus = 'online' | 'idle' | 'dnd' | 'offline'; @@ -59,3 +61,166 @@ export function queuePresenceRelay( 'profile', ); } + +/** + * On peer activation, send a fresh presence snapshot to the newly-active peer + * for every online local native user that has an S2S relationship with that + * peer (so the peer's stubs reflect current reality — presence is outbox-only, + * no mutation-log replay can do this). + * + * Scope is bounded by relationship count, not native count. A native qualifies + * if ANY of: + * - friend with at least one stub whose home_instance = peer domain + * - DM-member (closed=0) with at least one stub whose home_instance = peer domain + * - replicatedInstances JSON includes peerOrigin (explicit client-federation opt-in) + * + * Re-runs on every activation (including health-check unreachable→active flaps), + * because markPeerStubsOffline ran on deactivation — peers and our stubs both + * need a fresh handshake on recovery, not a stale-window-skip. + */ +export function snapshotPresenceForPeer(peerOrigin: string): void { + if (!isFederationRelayEnabled()) return; + + const db = getDb(); + const peerDomain = extractDomain(peerOrigin); + + // 1. All stubs from this peer that exist locally. + const peerStubIdList = db + .select({ id: schema.users.id }) + .from(schema.users) + .where(and( + eq(schema.users.homeInstance, peerDomain), + eq(schema.users.isDeleted, 0), + )) + .all() + .map((s) => s.id); + const peerStubIds = new Set(peerStubIdList); // O(1) membership tests in the friend loop + + // 2. Build the set of native IDs related to those stubs (friends + DM co-members). + const relatedNativeIds = new Set(); + if (peerStubIdList.length > 0) { + const friendRows = db.select().from(schema.friends) + .where(or( + inArray(schema.friends.userId, peerStubIdList), + inArray(schema.friends.friendId, peerStubIdList), + )) + .all(); + for (const f of friendRows) { + const stubSide = peerStubIds.has(f.userId) ? f.userId : f.friendId; + const otherSide = stubSide === f.userId ? f.friendId : f.userId; + relatedNativeIds.add(otherSide); + } + + // Stub's DM memberships — filter closed=0 so DMs the stub left don't pull + // their old co-members into snapshot scope. + const stubDmIds = db.select({ dmChannelId: schema.dmMembers.dmChannelId }) + .from(schema.dmMembers) + .where(and( + inArray(schema.dmMembers.userId, peerStubIdList), + eq(schema.dmMembers.closed, 0), + )) + .all() + .map((d) => d.dmChannelId); + if (stubDmIds.length > 0) { + // Co-members — filter closed=0 so a native who closed the DM locally + // doesn't receive snapshots for its lingering stub. + const dmCoMembers = db.select({ userId: schema.dmMembers.userId }) + .from(schema.dmMembers) + .where(and( + inArray(schema.dmMembers.dmChannelId, stubDmIds), + eq(schema.dmMembers.closed, 0), + )) + .all(); + for (const m of dmCoMembers) relatedNativeIds.add(m.userId); + } + } + + // 3. Add explicit client-federation opt-ins via replicatedInstances JSON. + // (We can't index a JSON string in SQLite, so scan natives once and parse.) + // SCALING NOTE: this full-natives scan is fine pre-launch and remains cheap + // for instances under ~10k users. If population grows past that, replace with + // a `user_replicated_instance_index` table populated when replicatedInstances + // is written, indexed on (origin) for direct EXISTS lookup. + const allNatives = db.select().from(schema.users) + .where(and( + isNull(schema.users.homeInstance), + eq(schema.users.isDeleted, 0), + )) + .all(); + for (const u of allNatives) { + if (!u.replicatedInstances) continue; + try { + const list = JSON.parse(u.replicatedInstances) as ReplicatedInstance[]; + if (list.some((ri) => ri.origin === peerOrigin)) relatedNativeIds.add(u.id); + } catch { /* malformed JSON — skip */ } + } + + // 4. Filter to online natives in the related set; emit one outbox event each. + for (const u of allNatives) { + if (!relatedNativeIds.has(u.id)) continue; + if (!u.status || u.status === 'offline') continue; + if (u.homeInstance) continue; // belt-and-braces: must be native + + const ts = Date.now(); + const event: FederationRelayEvent = { + eventType: 'presence_update', + contextType: 'profile', + messageId: `presence:${u.id}:${ts}:snap`, + encryptionVersion: 0, + timestamp: ts, + presenceUpdate: { + homeUserId: u.id, + homeInstance: getOurOrigin(), + status: u.status as 'online' | 'idle' | 'dnd', + ts, + }, + }; + queueOutboxEvent(u.id, u.id, 'presence_update', JSON.stringify(event), [peerOrigin], 'profile'); + } +} + +/** + * On peer deactivation (status flipping out of 'active'), flip all replicated + * stubs whose home is that peer to status='offline' and broadcast a local + * presence_update so connected friends/DM-mates/space-co-members see them go + * offline immediately. + * + * Imported lazily by onPeerDeactivated to avoid an import cycle through + * ws/handler.js (connectionManager). + * + * Accepts an origin string; derives the bare domain via extractDomain so the + * caller doesn't need to handle that detail. + */ +export async function markPeerStubsOffline(peerOrigin: string): Promise { + const peerDomain = extractDomain(peerOrigin); + const db = getDb(); + const stubs = db + .select({ id: schema.users.id }) + .from(schema.users) + .where(and( + eq(schema.users.homeInstance, peerDomain), + eq(schema.users.isDeleted, 0), + )) + .all(); + + if (stubs.length === 0) return; + + // Lazy import keeps this module pure for tests that don't need ws/handler. + const { connectionManager } = await import('../ws/handler.js'); + + for (const stub of stubs) { + db.update(schema.users) + .set({ status: 'offline' }) + .where(eq(schema.users.id, stub.id)) + .run(); + + const targets = collectProfileBroadcastTargetIds(stub.id); + const payload = { + type: 'presence_update' as const, + userId: stub.id, + status: 'offline' as const, + activities: [] as Activity[], + }; + for (const uid of targets) connectionManager.sendToUser(uid, payload); + } +}