diff --git a/packages/server/src/routes/federation.dmReconcileSweep.test.ts b/packages/server/src/routes/federation.dmReconcileSweep.test.ts new file mode 100644 index 00000000..21a65f28 --- /dev/null +++ b/packages/server/src/routes/federation.dmReconcileSweep.test.ts @@ -0,0 +1,113 @@ +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 { computeFederatedId } from '../utils/federationOutbox.js'; + +const __dirname = path.dirname(fileURLToPath(import.meta.url)); +type TestDb = ReturnType>; +let sqlite: Database.Database; +let testDb: TestDb; + +vi.mock('../db/index.js', () => ({ + getDb: () => testDb, + getRawDb: () => sqlite, + schema, +})); + +let _sf = 1; +vi.mock('../utils/snowflake.js', () => ({ + generateSnowflake: () => String(_sf++), + setWorkerId: vi.fn(), +})); + +vi.mock('../utils/federationAuth.js', async (importActual) => { + const actual = await importActual(); + return { ...actual, getOurOrigin: () => 'https://home.test' }; +}); + +vi.mock('../ws/handler.js', () => ({ + connectionManager: { + sendToUser: vi.fn(), + 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); + } + } +} + +function seedUser(id: string, homeUserId: string | null, homeInstance: string | null): void { + testDb.insert(schema.users).values({ + id, username: `${id}@x`, passwordHash: 'h', + homeUserId, homeInstance, createdAt: 1, + }).run(); +} +function seedChannel(id: string, fedId: string | null, members: string[]): void { + testDb.insert(schema.dmChannels).values({ id, federatedId: fedId, createdAt: 1 }).run(); + for (const u of members) testDb.insert(schema.dmMembers).values({ dmChannelId: id, userId: u, closed: 0 }).run(); +} +function seedMsg(id: string, chId: string, userId: string, ts: number): void { + testDb.insert(schema.dmMessages).values({ id, dmChannelId: chId, userId, content: 'x', createdAt: ts }).run(); +} + +beforeEach(() => { + sqlite = new Database(':memory:'); + testDb = drizzle(sqlite, { schema }); + applyMigrations(sqlite); + _sf = 1; +}); + +describe('reconcileDriftedDmFederatedIds', () => { + it('heals a drifted 1-on-1 channel and leaves correct ones untouched', async () => { + seedUser('a', 'a', null); seedUser('b', 'b-new', 'orbit.test'); seedUser('c', 'c', null); + // drifted: stored under old pairing, member b now has b-new; target under new pairing exists. + const oldFed = computeFederatedId('a', 'b-old'); const newFed = computeFederatedId('a', 'b-new'); + seedChannel('chOld', oldFed, ['a', 'b']); seedMsg('m1', 'chOld', 'a', 100); + seedChannel('chNew', newFed, ['a', 'b']); seedMsg('m2', 'chNew', 'a', 200); + // correct channel untouched + const okFed = computeFederatedId('a', 'c'); seedChannel('chOk', okFed, ['a', 'c']); + + const { reconcileDriftedDmFederatedIds } = await import('./federation.js'); + reconcileDriftedDmFederatedIds(); + + expect(testDb.select().from(schema.dmChannels).all().map(c => c.id).sort()).toEqual(['chNew', 'chOk']); + expect(testDb.select().from(schema.dmMessages).all().filter(m => m.dmChannelId === 'chNew').map(m => m.id).sort()).toEqual(['m1', 'm2']); + }); + + it('is idempotent — second run is a noop', async () => { + seedUser('a', 'a', null); seedUser('b', 'b', null); + seedChannel('ch1', computeFederatedId('a', 'b'), ['a', 'b']); + const { reconcileDriftedDmFederatedIds } = await import('./federation.js'); + reconcileDriftedDmFederatedIds(); + const before = testDb.select().from(schema.dmChannels).all(); + reconcileDriftedDmFederatedIds(); + expect(testDb.select().from(schema.dmChannels).all()).toEqual(before); + }); + + it('does not touch group DMs', async () => { + seedUser('a', 'a', null); seedUser('b', 'b', null); seedUser('c', 'c', null); + seedChannel('g1', 'c361f0db-d856-2b62-44f5-ed9eba92a67d', ['a', 'b', 'c']); + const { reconcileDriftedDmFederatedIds } = await import('./federation.js'); + reconcileDriftedDmFederatedIds(); + expect(testDb.select().from(schema.dmChannels).all().some(c => c.id === 'g1')).toBe(true); + }); +}); diff --git a/packages/server/src/routes/federation.ts b/packages/server/src/routes/federation.ts index c85f43c4..40a38ae8 100644 --- a/packages/server/src/routes/federation.ts +++ b/packages/server/src/routes/federation.ts @@ -7099,6 +7099,39 @@ export function reconcileDmChannelFederatedId( return { action: 'merged', channelId, targetChannelId: targetId, affectedUserIds: affected }; } +/** + * Startup sweep: reconcile any 1-on-1 DM channel whose stored federatedId has + * drifted from its members' current home identities (reattach-dm-reconcile + * spec §3.3). Heals accounts re-attached before inline reconciliation shipped + * (e.g. the live split-conversation duplicate). Idempotent; a noop on a clean DB. + */ +export function reconcileDriftedDmFederatedIds(): void { + const rawDb = getRawDb(); + const candidates = rawDb.prepare(` + SELECT c.id FROM dm_channels c + WHERE c.deleted_at IS NULL + AND c.federated_id IS NOT NULL + AND (SELECT count(*) FROM dm_members m WHERE m.dm_channel_id = c.id) = 2 + `).all() as Array<{ id: string }>; + if (candidates.length === 0) return; + + let rekeyed = 0; + let merged = 0; + rawDb.transaction(() => { + for (const c of candidates) { + // A prior merge in this loop may have deleted this id — reconcile returns + // noop for a missing/mutated channel, so this is safe. + const r = reconcileDmChannelFederatedId(rawDb, c.id); + if (r.action === 'rekeyed') rekeyed++; + else if (r.action === 'merged') merged++; + } + })(); + + if (rekeyed > 0 || merged > 0) { + console.log(`[federation] DM federatedId reconciliation: rekeyed ${rekeyed}, merged ${merged}`); + } +} + /** * Remove dead-incarnation artifacts produced by pre-fix initial syncs * (dead-incarnation spec §3.4): DM channels with no native member, and diff --git a/packages/server/src/utils/federationWorker.ts b/packages/server/src/utils/federationWorker.ts index 3f396532..58cc40ac 100644 --- a/packages/server/src/utils/federationWorker.ts +++ b/packages/server/src/utils/federationWorker.ts @@ -13,7 +13,7 @@ import { generateThumbnail } from './thumbnail.js'; import type { FederationRelayRequest, FederationRelayResponse, FederationRelayEvent } from '@backspace/shared'; import { startupBootstrapSync, onPeerDeactivated } from './federationPeerActivation.js'; import { probePeerReachable, recoverOrDetectReset, detectResetOnNeedsAttentionPeers, detectResetForPeer } from './federationRecovery.js'; -import { backfillReplicatedProfileAssets, sweepDeadIncarnationArtifacts } from '../routes/federation.js'; +import { backfillReplicatedProfileAssets, sweepDeadIncarnationArtifacts, reconcileDriftedDmFederatedIds } from '../routes/federation.js'; import { invokePermanentFailureCallback } from './federationRollback.js'; import { refreshPeerEpochs, getInstanceId } from './federationEpoch.js'; import fs from 'node:fs'; @@ -1312,6 +1312,14 @@ export function startFederationWorkers(): void { console.error('[federation-worker] Dead-incarnation sweep error:', err); } + // Heal any 1-on-1 DM channels whose federatedId drifted from their members' + // current identities (reattach-dm-reconcile spec §3.3). + try { + reconcileDriftedDmFederatedIds(); + } catch (err) { + console.error('[federation-worker] DM federatedId reconciliation error:', err); + } + // Backfill any replicated user avatars/banners still stored as absolute URLs // (legacy data from before file replication, or rows whose home was offline // on a previous attempt). Best-effort and idempotent — safe to re-run.