feat(federation): reconcileDmChannelFederatedId — re-key/merge 1-on-1 DM channels on identity change (reattach-dm-reconcile spec §3.1)

packages/server/src/routes/federation.dmReconcile.test.ts

@@ -0,0 +1,151 @@
+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' };
+});
+
+// federation.ts also imports connectionManager/ws — stub minimal surface so
+// the route module loads at test time. The function under test doesn't touch any of these.
+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('reconcileDmChannelFederatedId', () => {
+  it('noop when the stored id already matches the members', async () => {
+    seedUser('a', 'a', null); seedUser('b', 'b-home', 'orbit.test');
+    const fed = computeFederatedId('a', 'b-home');
+    seedChannel('ch1', fed, ['a', 'b']);
+    const { reconcileDmChannelFederatedId } = await import('./federation.js');
+    const r = reconcileDmChannelFederatedId(sqlite, 'ch1');
+    expect(r.action).toBe('noop');
+    expect(testDb.select().from(schema.dmChannels).get()!.federatedId).toBe(fed);
+  });
+
+  it('re-keys in place when a member home id changed and no target exists', async () => {
+    // member b now has NEW home id 'b-new'; channel still carries the OLD-pair id.
+    seedUser('a', 'a', null); seedUser('b', 'b-new', 'orbit.test');
+    const oldFed = computeFederatedId('a', 'b-old');
+    seedChannel('ch1', oldFed, ['a', 'b']);
+    const { reconcileDmChannelFederatedId } = await import('./federation.js');
+    const r = reconcileDmChannelFederatedId(sqlite, 'ch1');
+    expect(r.action).toBe('rekeyed');
+    expect(testDb.select().from(schema.dmChannels).get()!.federatedId).toBe(computeFederatedId('a', 'b-new'));
+  });
+
+  it('merges into the target when one already carries the new id', async () => {
+    seedUser('a', 'a', null); seedUser('b', 'b-new', 'orbit.test');
+    const oldFed = computeFederatedId('a', 'b-old');
+    const newFed = computeFederatedId('a', 'b-new');
+    seedChannel('chOld', oldFed, ['a', 'b']); seedMsg('m1', 'chOld', 'a', 100); seedMsg('m2', 'chOld', 'b', 110);
+    seedChannel('chNew', newFed, ['a', 'b']); seedMsg('m3', 'chNew', 'a', 120);
+    const { reconcileDmChannelFederatedId } = await import('./federation.js');
+    const r = reconcileDmChannelFederatedId(sqlite, 'chOld');
+    expect(r.action).toBe('merged');
+    expect(r.targetChannelId).toBe('chNew');
+    // old channel gone, all 3 messages now on chNew, ordered.
+    expect(testDb.select().from(schema.dmChannels).all().map(c => c.id)).toEqual(['chNew']);
+    const msgs = testDb.select().from(schema.dmMessages).all().filter(m => m.dmChannelId === 'chNew').sort((x, y) => x.createdAt - y.createdAt);
+    expect(msgs.map(m => m.id)).toEqual(['m1', 'm2', 'm3']);
+    // members deduped, read_states intact.
+    expect(testDb.select().from(schema.dmMembers).all().filter(m => m.dmChannelId === 'chNew').map(m => m.userId).sort()).toEqual(['a', 'b']);
+    expect(r.affectedUserIds.sort()).toEqual(['a', 'b']);
+  });
+
+  it('skips group DMs (UUID federatedId / >2 members)', async () => {
+    seedUser('a', 'a', null); seedUser('b', 'b', null); seedUser('c', 'c', null);
+    seedChannel('g1', 'c361f0db-d856-2b62-44f5-ed9eba92a67d', ['a', 'b', 'c']);
+    const { reconcileDmChannelFederatedId } = await import('./federation.js');
+    expect(reconcileDmChannelFederatedId(sqlite, 'g1').action).toBe('noop');
+  });
+
+  it('skips a channel with an unresolvable member set (not exactly 2)', async () => {
+    seedUser('a', 'a', null);
+    seedChannel('ch1', computeFederatedId('a', 'b'), ['a']);
+    const { reconcileDmChannelFederatedId } = await import('./federation.js');
+    expect(reconcileDmChannelFederatedId(sqlite, 'ch1').action).toBe('noop');
+  });
+
+  it('dedupes read_states on merge (composite PK user_id+channel_id)', async () => {
+    seedUser('a', 'a', null); seedUser('b', 'b-new', 'orbit.test');
+    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', 120);
+    testDb.insert(schema.readStates).values([
+      { userId: 'a', channelId: 'chOld', lastReadMessageId: 'm1', updatedAt: 1 },
+      { userId: 'a', channelId: 'chNew', lastReadMessageId: 'm2', updatedAt: 2 },
+    ]).run();
+    const { reconcileDmChannelFederatedId } = await import('./federation.js');
+    reconcileDmChannelFederatedId(sqlite, 'chOld');
+    const rs = testDb.select().from(schema.readStates).all();
+    expect(rs.filter(r => r.channelId === 'chOld')).toHaveLength(0);
+    expect(rs.filter(r => r.channelId === 'chNew' && r.userId === 'a')).toHaveLength(1);
+  });
+});

packages/server/src/routes/federation.ts

@@ -6987,6 +6987,77 @@ export function processPresenceUpdateEvent(
 
 // ─── Dead-Incarnation Startup Sweep ─────────────────────────────────────────
 
+export interface DmReconcileResult {
+  action: 'noop' | 'rekeyed' | 'merged';
+  channelId: string;
+  targetChannelId: string;
+  affectedUserIds: string[];
+}
+
+/**
+ * Reconcile a single 1-on-1 DM channel's deterministic federatedId against its
+ * members' CURRENT home identities (reattach-dm-reconcile spec §3.1). A 1-on-1
+ * federatedId is f(sorted home user ids); when a participant's home_user_id
+ * changes (re-attach), the channel's stored id goes stale and new messages
+ * compute a different id → a split conversation. This re-keys the channel in
+ * place, or — when a channel already carries the correct id (idx_dm_federated is
+ * UNIQUE, so two rows can't share it) — merges this channel INTO that one and
+ * deletes it.
+ *
+ * Idempotent: a correctly-keyed channel is a noop. Group DMs (UUID federatedId
+ * or member count != 2) are skipped. Must be called inside a transaction.
+ */
+export function reconcileDmChannelFederatedId(
+  rawDb: ReturnType<typeof getRawDb>,
+  channelId: string,
+): DmReconcileResult {
+  const noop: DmReconcileResult = { action: 'noop', channelId, targetChannelId: channelId, affectedUserIds: [] };
+
+  const chan = rawDb.prepare(`SELECT id, federated_id FROM dm_channels WHERE id = ? AND deleted_at IS NULL`).get(channelId) as
+    { id: string; federated_id: string | null } | undefined;
+  if (!chan || !chan.federated_id) return noop;
+  // Only 1-on-1 shape (32 hex). Group DMs use a random UUID.
+  if (!/^[0-9a-f]{32}$/.test(chan.federated_id)) return noop;
+
+  const members = rawDb.prepare(`
+    SELECT u.id, u.home_user_id FROM dm_members m JOIN users u ON u.id = m.user_id
+    WHERE m.dm_channel_id = ?
+  `).all(channelId) as Array<{ id: string; home_user_id: string | null }>;
+  if (members.length !== 2) return noop;
+
+  const homeA = members[0]!.home_user_id || members[0]!.id;
+  const homeB = members[1]!.home_user_id || members[1]!.id;
+  const expected = computeFederatedId(homeA, homeB);
+  if (expected === chan.federated_id) return noop;
+
+  const target = rawDb.prepare(`SELECT id FROM dm_channels WHERE federated_id = ? AND deleted_at IS NULL AND id != ?`).get(expected, channelId) as
+    { id: string } | undefined;
+
+  if (!target) {
+    rawDb.prepare(`UPDATE dm_channels SET federated_id = ? WHERE id = ?`).run(expected, channelId);
+    return { action: 'rekeyed', channelId, targetChannelId: channelId, affectedUserIds: members.map(m => m.id) };
+  }
+
+  // Merge source (channelId) INTO target, then delete source.
+  const targetId = target.id;
+  const targetMemberIds = (rawDb.prepare(`SELECT user_id FROM dm_members WHERE dm_channel_id = ?`).all(targetId) as Array<{ user_id: string }>).map(r => r.user_id);
+  const affected = Array.from(new Set([...members.map(m => m.id), ...targetMemberIds]));
+
+  // Messages: globally-unique ids, straight move (attachments + dm_reactions
+  // reference dm_message_id and follow automatically).
+  rawDb.prepare(`UPDATE dm_messages SET dm_channel_id = ? WHERE dm_channel_id = ?`).run(targetId, channelId);
+  // Members: drop source rows already present on target (composite PK), repoint the rest.
+  rawDb.prepare(`DELETE FROM dm_members WHERE dm_channel_id = ? AND user_id IN (SELECT user_id FROM dm_members WHERE dm_channel_id = ?)`).run(channelId, targetId);
+  rawDb.prepare(`UPDATE dm_members SET dm_channel_id = ? WHERE dm_channel_id = ?`).run(targetId, channelId);
+  // read_states: keyed by channel_id; dedupe on (user_id, channel_id) then repoint.
+  rawDb.prepare(`DELETE FROM read_states WHERE channel_id = ? AND user_id IN (SELECT user_id FROM read_states WHERE channel_id = ?)`).run(channelId, targetId);
+  rawDb.prepare(`UPDATE read_states SET channel_id = ? WHERE channel_id = ?`).run(targetId, channelId);
+  // Remove the now-empty source channel.
+  rawDb.prepare(`DELETE FROM dm_channels WHERE id = ?`).run(channelId);
+
+  return { action: 'merged', channelId, targetChannelId: targetId, affectedUserIds: affected };
+}
+
 /**
  * Remove dead-incarnation artifacts produced by pre-fix initial syncs
  * (dead-incarnation spec §3.4): DM channels with no native member, and