diff --git a/packages/server/src/routes/federation.dmReconcile.test.ts b/packages/server/src/routes/federation.dmReconcile.test.ts new file mode 100644 index 00000000..103a8bce --- /dev/null +++ b/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>; +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' }; +}); + +// 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); + }); +}); diff --git a/packages/server/src/routes/federation.ts b/packages/server/src/routes/federation.ts index d2f85862..382ae82f 100644 --- a/packages/server/src/routes/federation.ts +++ b/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, + 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