feat(federation): startup sweep reconciles drifted 1-on-1 DM federatedIds — heals pre-fix re-attaches (reattach-dm-reconcile spec §3.3)

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<typeof drizzle<typeof schema>>;
+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<typeof import('../utils/federationAuth.js')>();
+  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);
+  });
+});

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

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.