feat(federation): peer-lifecycle presence hooks (snapshot on activate, mark-offline on deactivate)
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.
This commit is contained in:
Jannis Braun
@@ -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);
|
||||
|
||||
@@ -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<typeof drizzle<typeof schema>>;
|
||||
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');
|
||||
});
|
||||
});
|
||||
@@ -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<string>();
|
||||
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<void> {
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user