Something went wrong. Try again.
ATProto-based AI Assistant Chat interface.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268import { nsid } from '@scn-chat/lexicons'import { sql } from 'kysely'import { describe, expect, it } from 'vitest'import { recordCid } from '../../src/storage/records.ts'import { SyncEngine } from '../../src/sync/engine.ts'import { ALICE, collect, spacesHarness, userMessage } from '../helpers/spaces.ts'
const stateOf = async (h: Awaited<ReturnType<typeof spacesHarness>>, space: string) => h.db.selectFrom('sync_state').selectAll().where('space_uri', '=', space).executeTakeFirst()
/** A direct client following the index convention: write the message, then bump the entry. */async function writeAsClient( h: Awaited<ReturnType<typeof spacesHarness>>, skey: string, rkey: string, record: object,) { await h.external.createRecord(h.chats.conversationUri(skey), nsid.message, rkey, record as never) const ref = await h.external.getRecord(h.chats.settingsUri, nsid.conversationRef, skey) await h.external.putRecord(h.chats.settingsUri, nsid.conversationRef, skey, { ...(ref?.value ?? { $type: nsid.conversationRef, conversation: h.chats.conversationUri(skey) }), updatedAt: new Date().toISOString(), })}
describe('SyncEngine', () => { it('does not sync a user without access', async () => { let allowed = false const h = await spacesHarness({ hasAccess: async () => allowed }) const { skey } = await h.chats.createConversation() await h.external.createRecord( h.chats.conversationUri(skey), nsid.message, '3mmmmmmmmmmm1', userMessage('hi') as never, ) const changed = collect(h.events, 'message:changed') await h.engine.syncConversation(ALICE, skey) expect(changed).toEqual([]) allowed = true await h.engine.syncConversation(ALICE, skey) expect(changed).toHaveLength(1) })
it('turns a direct write noticed through the index into a live message event', async () => { const h = await spacesHarness() const { skey } = await h.chats.createConversation() await h.engine.syncIndex(ALICE) const changed = collect(h.events, 'message:changed') await writeAsClient(h, skey, '3mmmmmmmmmmm1', userMessage('hi', { generation: {} })) await h.engine.syncIndex(ALICE) expect(changed).toMatchObject([ { did: ALICE, skey, rkey: '3mmmmmmmmmmm1', kind: 'create', live: true }, ]) })
it('emits backfill events on the first sync of a conversation', async () => { const h = await spacesHarness() const { skey } = await h.chats.createConversation() await h.external.createRecord( h.chats.conversationUri(skey), nsid.message, '3mmmmmmmmmmm1', userMessage('old') as never, ) const changed = collect(h.events, 'message:changed') await h.engine.syncConversation(ALICE, skey) expect(changed).toMatchObject([{ rkey: '3mmmmmmmmmmm1', live: false }]) })
it('tells open tabs when another client deletes a message', async () => { const h = await spacesHarness() const { skey, uri } = await h.chats.createConversation() await h.external.createRecord(uri, nsid.message, '3mmmmmmmmmmm1', userMessage('hi') as never) await h.engine.syncConversation(ALICE, skey) const changed = collect(h.events, 'conversation:changed') await h.external.deleteRecord(uri, nsid.message, '3mmmmmmmmmmm1') await h.engine.syncConversation(ALICE, skey) expect(changed).toEqual([{ did: ALICE, skey }]) })
it('does not copy the title of an info record that fails validation into the index', async () => { const h = await spacesHarness() const { skey, uri } = await h.chats.createConversation() await h.engine.syncConversation(ALICE, skey) const value = { $type: nsid.info, title: 'Bad', createdAt: 5 } const raw = JSON.stringify(value) const cid = await recordCid(value) await sql`update local_record set value_json = ${raw}, cid = ${cid} where space_uri = ${uri}`.execute( h.db, ) await sql`insert into local_op (space_uri, collection, rkey, cid, value_json, created_at) values (${uri}, ${nsid.info}, 'self', ${cid}, ${raw}, 'now')`.execute( h.db, ) await h.engine.syncConversation(ALICE, skey) const { conversations } = await h.chats.listConversations() expect(conversations[0]?.title).toBeNull() })
it('emits no message events for the server own writes', async () => { const h = await spacesHarness() const { skey } = await h.chats.createConversation() await h.engine.syncConversation(ALICE, skey) const changed = collect(h.events, 'message:changed') await h.chats.createMessage(skey, '3mmmmmmmmmmm1', userMessage('mine') as never) await h.engine.syncConversation(ALICE, skey) expect(changed).toEqual([]) })
it('syncs only conversations inside the backfill window on a first index sync, and starts at the head', async () => { const h = await spacesHarness({ backfillWindowMs: 60 * 60_000 }) const recent = await h.chats.createConversation() const old = await h.chats.createConversation() await h.external.putRecord(h.chats.settingsUri, nsid.conversationRef, old.skey, { $type: nsid.conversationRef, conversation: old.uri, updatedAt: new Date(Date.now() - 2 * 60 * 60_000).toISOString(), }) await h.engine.syncIndex(ALICE) expect(await stateOf(h, recent.uri)).toBeDefined() expect(await stateOf(h, old.uri)).toBeUndefined() expect((await stateOf(h, h.chats.settingsUri))?.last_rev).toBe( await h.external.headRev(h.chats.settingsUri), ) })
it('reports a synced message that fails validation instead of treating it as a message', async () => { const h = await spacesHarness() const { skey, uri } = await h.chats.createConversation() await h.engine.syncConversation(ALICE, skey) const invalid = collect(h.events, 'message:invalid') const changed = collect(h.events, 'message:changed') const value = { $type: nsid.message, role: 'user', createdAt: new Date().toISOString() } const raw = JSON.stringify(value) const cid = await recordCid(value) await sql`insert into local_record values (${uri}, ${nsid.message}, 'bad', ${raw}, ${cid}, 'now')`.execute( h.db, ) await sql`insert into local_op (space_uri, collection, rkey, cid, value_json, created_at) values (${uri}, ${nsid.message}, 'bad', ${cid}, ${raw}, 'now')`.execute( h.db, ) await h.engine.syncConversation(ALICE, skey) expect(invalid).toMatchObject([{ rkey: 'bad', error: expect.stringMatching(/content/) }]) expect(changed).toEqual([]) })
it('drops a conversation whose index entry was deleted', async () => { const h = await spacesHarness() const { skey, uri } = await h.chats.createConversation() await h.engine.syncIndex(ALICE) const deleted = collect(h.events, 'conversation:deleted') await h.external.deleteRecord(h.chats.settingsUri, nsid.conversationRef, skey) await h.engine.syncIndex(ALICE) expect(deleted).toMatchObject([{ skey }]) expect(await stateOf(h, uri)).toBeUndefined() })
it('copies an externally changed info title into the index entry', async () => { const h = await spacesHarness() const { skey, uri } = await h.chats.createConversation() await h.engine.syncConversation(ALICE, skey) await h.external.putRecord(uri, nsid.info, 'self', { $type: nsid.info, createdAt: new Date().toISOString(), title: 'Renamed elsewhere', }) await h.engine.syncConversation(ALICE, skey) expect((await h.chats.listConversations()).conversations[0]?.title).toBe('Renamed elsewhere') })
it('adds a missing index entry for a conversation created by another client, and syncs it', async () => { const h = await spacesHarness() const uri = await h.external.createSpace(nsid.conversation, '3cccccccccccc') await h.external.putRecord(uri, nsid.info, 'self', { $type: nsid.info, createdAt: new Date().toISOString(), title: 'From elsewhere', }) await h.external.createRecord(uri, nsid.message, '3mmmmmmmmmmm1', userMessage('hello') as never) const changed = collect(h.events, 'message:changed') await h.engine.discover(ALICE) expect((await h.chats.listConversations()).conversations).toMatchObject([ { skey: '3cccccccccccc', title: 'From elsewhere' }, ]) expect(changed).toMatchObject([{ rkey: '3mmmmmmmmmmm1', live: false }]) })
it('keeps discovering past a conversation with an invalid info record', async () => { const h = await spacesHarness() const bad = await h.external.createSpace(nsid.conversation, '3bbbbbbbbbbbb') await h.external.putRecord(bad, nsid.info, 'self', { $type: nsid.info, createdAt: new Date().toISOString(), }) await sql`update local_record set value_json = ${JSON.stringify({ $type: nsid.info, title: 5 })} where space_uri = ${bad}`.execute( h.db, ) const good = await h.external.createSpace(nsid.conversation, '3cccccccccccc') await h.external.putRecord(good, nsid.info, 'self', { $type: nsid.info, createdAt: new Date().toISOString(), }) await h.engine.discover(ALICE) expect((await h.chats.listConversations()).conversations.map((c) => c.skey)).toContain( '3cccccccccccc', ) })
it('does nothing for a local account', async () => { const h = await spacesHarness({ storageMode: 'local' }) const changed = collect(h.events, 'message:changed') await h.engine.syncIndex(ALICE) expect(changed).toEqual([]) expect(await stateOf(h, h.chats.settingsUri)).toBeUndefined() })
it('stores the registration expiry and renews registrations that expire within an hour', async () => { const h = await spacesHarness() await h.engine.registerIndex(ALICE) expect((await stateOf(h, h.chats.settingsUri))?.registered_until).toBeTruthy() await h.db .updateTable('sync_state') .set({ registered_until: new Date(Date.now() + 60_000).toISOString() }) .execute() await h.engine.renewRegistrations() const renewed = Date.parse((await stateOf(h, h.chats.settingsUri))?.registered_until ?? '') expect(renewed).toBeGreaterThan(Date.now() + 60 * 60_000) })})
describe('SyncEngine registration on a loopback URL', () => { it('skips registering for notifications a PDS could never deliver', async () => { const h = await spacesHarness() const engine = new SyncEngine({ ...(h.engine as unknown as { deps: ConstructorParameters<typeof SyncEngine>[0] }).deps, publicUrl: 'http://127.0.0.1:5173', }) await engine.registerIndex(ALICE) expect(await stateOf(h, h.chats.settingsUri)).toBeUndefined() })})
describe('SyncEngine gap detection', () => { const signedHarness = spacesHarness
it('keeps the cursor when the signed commit matches the index', async () => { const h = await signedHarness() await h.chats.createConversation() await h.engine.syncIndex(ALICE) const before = (await stateOf(h, h.chats.settingsUri))?.last_rev await h.engine.safetyNet(ALICE) expect((await stateOf(h, h.chats.settingsUri))?.last_rev).toBe(before) })
it('resets the cursor when the signed commit does not match the index', async () => { const h = await signedHarness() const { skey } = await h.chats.createConversation() await h.engine.syncIndex(ALICE) await h.db .deleteFrom('sync_state') .where('space_uri', '=', h.chats.conversationUri(skey)) .execute() h.pds.tamperCommits = true await h.engine.safetyNet(ALICE) expect(await stateOf(h, h.chats.conversationUri(skey))).toBeDefined() })})