diff --git a/ai-docs/ai-migration-v9-to-v10.md b/ai-docs/ai-migration-v9-to-v10.md index 2150f24e7a..6f400cbede 100644 --- a/ai-docs/ai-migration-v9-to-v10.md +++ b/ai-docs/ai-migration-v9-to-v10.md @@ -1567,6 +1567,149 @@ unread, and the call only ever 404'd. --- +# Part M — Optimistic edit, delete and unsent-message persistence + +Three v9 capabilities did not survive the move of the message lifecycle out of `` and into +`stream-chat`'s `MessageOperations` engine. All are restored, all inside the LLC. No API was removed +and there is nothing to migrate — but if you worked around any of them, the workaround is now +redundant and will double-apply. + +## M.1 Editing a message is optimistic again, and persisted (bug fix) + +An edit shows immediately and is written to the offline DB before the request is made, so it survives +a cold start and the hydration `` performs on mount. v9 did both (`Channel.tsx` wrote the +optimistic copy through `db.updateMessage`); v10 kept neither. + +`message_text_updated_at` is stamped optimistically too, so the "edited" indicator appears at once — +except when the message being edited is itself `failed`, which never had a server-confirmed text +update to advertise. + +**An edit is never rolled back.** Reverting would discard text the user typed. What changes is only +whether the failure is shown on the message: + +- **Offline support enabled and the request was queued for replay** — the message does **not** enter a + failed state. It is pending, not failed, and marking it failed lights up the retry affordance, which + re-*sends* the message rather than re-editing it. The promise still rejects, so any notification you + surface from a rejected `editMessage` is unaffected. +- **No offline DB, or a definitive rejection** (a server error that is not retryable, or a cancelled + request) — + the message keeps the edit and gains `status: 'failed'` plus `error`, as before. + +The predicate is the same one reactions already use: an offline DB is present **and** the error is +`isEphemeral`. + +## M.2 Deleting a message is optimistic, and reverted if it definitively fails (behavioral) + +`MessageOperations.delete` was the only operation that bypassed the optimistic lifecycle: it awaited +the request and then ingested the response. So a delete showed nothing until the server answered. + +Now: + +- A soft delete immediately marks the message `type: 'deleted'` with `deleted_at`, and + `deleted_for_me` for a `delete_for_me` delete. +- A **hard delete removes the message** instead of marking it, matching what the `message.deleted` WS + handler does for `hard_delete`. Previously the response message was ingested unconditionally, which + put a message the server had just destroyed back into the list. +- A delete that fails definitively is **reverted** — unlike an edit there is no user input to lose, and + a "Message deleted" placeholder on a message that still exists server-side only self-corrects on the + next query. A queued (offline) delete keeps the optimistic state. + + In practice the trigger is **offline support disabled plus no connectivity**, not a permission + rejection: the delete action is capability-gated in the UI (`deleteOwnMessage` / `deleteAnyMessage`), + so a user without permission is never offered it. If the revert is ever wrong — the server did delete + the message but the response was lost — the `message.deleted` event removes it again. +- Deleting a message nothing was displaying no longer inserts a phantom deleted row. + +The revert is guarded on object identity against the copy the optimistic step wrote, so a WS update +landing mid-request is never clobbered by the rollback. + +## M.3 Unsent messages survive closing the app (bug fix — data loss) + +Every send now writes the message to the offline DB **before** the request, pessimistically marked +`failed`, and overwrites it with `received` on success. A process death anywhere between composing and +the server's ack therefore leaves a message that hydrates as failed and retryable, instead of one that +silently disappears. This is v9's write-ahead (`Channel.tsx`), which v10 dropped when `sendMessage` +collapsed into `sendMessageWithLocalUpdate`. + +No schema change and no `dbVersion` bump: `status` already round-trips through the storable's +`extraData` blob. + +The retry payload does not need persisting alongside it. `MessageOperations.retry` reconstructs the +request from the message when its in-memory `failedSendCache` is cold, so a persisted failed message +stays retryable past that cache's 5-minute TTL and across restarts. + +`channel.reload()` also consults the DB, not just the in-memory window, when deciding which failed +messages a reconnect has to put back — a message evicted from the paginator, or one on a cold boot that +the paginator never held, was previously unrecoverable. v9 got this from a second, lagging copy of the +message list (the SDK's own React state); with a single reactive source of truth the persisted row is +that buffer. + +## M.4 An empty request response no longer overwrites local state (bug fix) + +`formatMessage(undefined)` does not throw — it returns `{ status: 'received', created_at: , +updated_at: }`. Because that `updated_at` is *now*, it beat the "is the server copy newer?" check +and was ingested as an id-less message. v9 guarded every apply with `if (response?.message)`; the guard +is back. Reachable from a custom `sendMessageRequest` / `updateMessageRequest` / `deleteMessageRequest` +that resolves without a `message`. + +## M.5 Editing or deleting a thread parent from inside a thread works (bug fix) + +The reply paginator's local filter is `{ cid, parent_id }`, and a parent message has no `parent_id` — +so routing a parent edit or delete through the open thread's instance (which is what the SDKs do while +a thread is on screen) handed it to a collection that could not hold it, and the operation was silently +dropped. Optimistic writes now fall back to the client-global message store when the paginator does not +accept the message, reaching it wherever it is held and fanning out to every collection that holds it — +the same addressing `applyReactionLocally` uses. The SDK additionally routes by membership rather than +by "is a thread open", mirroring `sendReaction`. + +## M.6 Additive on `AbstractOfflineDB` + +Both are **concrete** helpers composed from existing primitives, so a custom offline DB implementation +inherits them and has nothing new to implement: + +- `getFailedMessages({ cid })` — the channel's locally failed (unsent) messages, read back through + `getChannels`. +- `upsertMessageWithChannelGuard({ message })` — upserts one message, creating its channel row first + when the DB has never seen that channel, so the optimistic write cannot fail on the foreign key. + `updateMessage` cannot serve this: it is an UPDATE and no-ops when the row does not exist, which is + exactly the write-ahead case. + +## M.7 The offline channel guard actually guards now (bug fix) + +`channelExists` ran `SELECT EXISTS(SELECT 1 FROM channels WHERE cid = ?)` and returned +`rows.length > 0`. `SELECT EXISTS` always returns exactly one row — holding `0` or `1` — so the row +count carried no information and the helper reported `true` for every cid, present or not. It has done +so since offline support v2 shipped; the LLC's tests mock `channelExists`, so the logic was covered and +the SQL never was. + +Its only consumer is `AbstractOfflineDB.queriesWithChannelGuard`, used by ten WS-event handlers +(`message.new`, `message.deleted`, `message.updated`, `message.read`, `member.*`, `reaction.*`, …). Its +gate is `forceUpdate || !(await channelExists({ cid }))`, which collapsed to `forceUpdate` — so the +branch that recreates a missing channel row from the event never ran, and those writes died on the +`messages.cid → channels.cid` foreign key and were swallowed by the detached query runner. + +Reachable whenever an event arrives for a channel the DB has no row for: after `resetDB()` (which the +sync-failure path and a >30-day-stale sync both trigger) until the next channel-list query; in the +window between `queryChannels` resolving and its persistence completing (and `channel.query`'s +persistence is detached, widening it); and after being added to a channel mid-session. The message list +self-heals on the next query — reactions, read state, member changes and older messages in those +windows did not. + +**Both channel guards are also lazy now.** `queriesWithChannelGuard` and +`upsertMessageWithChannelGuard` attempt the write first and only probe for the channel row when it +fails, repairing and retrying once; a failure with the channel row present is rethrown without a retry. +Every statement involved is an upsert, so the retry is idempotent. The eager probe is kept for +`execute: false` callers (collecting queries for someone else's batch, so there is no failure to catch) +and for `forceUpdate`. + +That inversion matters because the probe is a native round-trip, not a cheap read. Measured through +op-sqlite on device: the probe costs ~0.5ms against ~8-12ms for the message upsert it protects, and +roughly two thirds of that is the JS↔native crossing rather than the query. Removing it from the happy +path takes it off every message received as well as every message written — so message writes are +cheaper than they were before this release, not more expensive. + +--- + ## 19. Verify - Typecheck the customer app; removed symbols surface as "Property does not diff --git a/package/src/__tests__/offline-support/optimistic-update.tsx b/package/src/__tests__/offline-support/optimistic-update.tsx index ecb54974fa..5fe0492110 100644 --- a/package/src/__tests__/offline-support/optimistic-update.tsx +++ b/package/src/__tests__/offline-support/optimistic-update.tsx @@ -86,6 +86,11 @@ const markConnectionUnhealthy = (client: StreamChat) => { (client.wsConnection as unknown as { isHealthy: boolean }).isHealthy = false; }; +/** The counterpart of {@link markConnectionUnhealthy}, for tests that go offline and then reconnect. */ +const markConnectionHealthy = (client: StreamChat) => { + (client.wsConnection as unknown as { isHealthy: boolean }).isHealthy = true; +}; + // React flushes passive effects child-first, so the test-callback effect below runs BEFORE `Channel`'s // own mount effects — verifiably: without this wait, `channel.configState.requestHandlers` at edit time // holds only the declaratively-registered `updateMessageRequest`, with no `sendMessageRequest`, because @@ -885,6 +890,13 @@ export const OptimisticUpdates = () => { { + // Same barrier every other "edit message" test uses. Without it the edit fires from + // this child mount effect BEFORE `Channel`'s own effect runs `channel.watch()`, whose + // seed then re-ingests the pre-edit copy from the mocked query response and overwrites + // the optimistic edit. Measured: the optimistic copy is correct at `editMessage` + // resolution and still correct a macrotask later, then the in-flight watch lands on + // top of it. That window is unreachable in production (see flushMountEffects). + await flushMountEffects(); // Go offline BEFORE editing so the default (no-handler) offline path runs. markConnectionUnhealthy(chatClient); try { @@ -914,6 +926,9 @@ export const OptimisticUpdates = () => { const dbMessage = dbMessages.find((row) => row.id === message.id); expect(updatedMessage?.text).toBe(editedText); + // Offline support is enabled and the edit was queued for replay, so this is "pending", + // not "failed" — the message must never enter a failed state on this path. + expect(updatedMessage?.status).not.toBe(MessageStatusTypes.FAILED); expect(dbMessage?.text).toBe(editedText); }, { timeout: 2500 }, @@ -928,6 +943,7 @@ export const OptimisticUpdates = () => { { + await flushMountEffects(); markConnectionUnhealthy(chatClient); try { await deleteMessage(message); @@ -959,6 +975,184 @@ export const OptimisticUpdates = () => { }); }); + describe('failed message persistence', () => { + it('persists a failed send so it survives a restart and reads back as retryable', async () => { + const localMessage = generateMessage({ + cid: channel.cid, + status: MessageStatusTypes.SENDING, + text: 'unsent across a restart', + user: chatClient.user as UserResponse, + user_id: chatClient.userID, + }); + + jest + .spyOn(channel.messageComposer, 'compose') + .mockResolvedValue({ localMessage, message: localMessage } as unknown as Awaited< + ReturnType + >); + + render( + + + { + await flushMountEffects(); + markConnectionUnhealthy(chatClient); + try { + await sendMessage(); + } catch (e) { + // do nothing + } + }} + context={MessageInputContext} + > + + + + , + ); + await waitFor(() => expect(screen.getByTestId('children')).toBeTruthy()); + + // The row itself is what makes a failed message survive a process death: v9 wrote it ahead of + // the request and v10 dropped that write, which is why closing the app lost unsent messages. + await waitFor(async () => { + const dbMessages = await BetterSqlite.selectFromTable<{ + extraData: string; + id: string; + text: string; + }>('messages'); + const dbMessage = dbMessages.find((row) => row.id === localMessage.id); + + expect(dbMessage).toBeTruthy(); + expect(dbMessage!.text).toBe(localMessage.text); + // `status` has no column of its own — it round-trips through the extraData blob. + expect(JSON.parse(dbMessage!.extraData).status).toBe(MessageStatusTypes.FAILED); + }); + + // And it has to come back through the DB's own read path, which is what a cold start hydrates + // from and what `Channel.reload` consults on reconnect. + const restored = await ( + chatClient.offlineDb as unknown as { + getFailedMessages: (o: { cid: string }) => Promise; + } + ).getFailedMessages({ cid: channel.cid }); + + expect(restored.map((message) => message.id)).toContain(localMessage.id); + expect(restored.find((message) => message.id === localMessage.id)?.text).toBe( + localMessage.text, + ); + }); + }); + + describe('channel guard cost', () => { + it('writes an optimistic message without probing for the channel row', async () => { + const localMessage = generateMessage({ + cid: channel.cid, + status: MessageStatusTypes.SENDING, + text: 'no guard probe please', + user: chatClient.user as UserResponse, + user_id: chatClient.userID, + }); + + jest + .spyOn(channel.messageComposer, 'compose') + .mockResolvedValue({ localMessage, message: localMessage } as unknown as Awaited< + ReturnType + >); + + let guardSpy: jest.SpyInstance | undefined; + + render( + + + { + await flushMountEffects(); + // Spied after mount, so the count covers only the send below — not the channel + // query that `Channel` performs while starting up. + guardSpy = jest.spyOn( + chatClient.offlineDb as unknown as { channelExists: () => Promise }, + 'channelExists', + ); + try { + await sendMessage(); + } catch (e) { + // do nothing + } + }} + context={MessageInputContext} + > + + + + , + ); + await waitFor(() => expect(screen.getByTestId('children')).toBeTruthy()); + + // The write has to actually have happened, or "no probe" would be trivially true. + await waitFor(async () => { + const dbMessages = await BetterSqlite.selectFromTable<{ id: string }>('messages'); + expect(dbMessages.some((row) => row.id === localMessage.id)).toBe(true); + }); + + // The guard is lazy: it attempts the write and only probes if that fails on the foreign key. + // A probe here means the eager version is back — a native round-trip per message write, for + // every message written AND every message received. + expect(guardSpy).not.toHaveBeenCalled(); + }); + }); + + describe('optimistic edit without offline support', () => { + it('keeps the optimistic edit AND surfaces the failure when there is no offline DB', async () => { + const message = channel.messagePaginator.headItems[0]; + const editedText = 'edited with no offline support'; + + chatClient.config.set({ + channel: { + requestHandlers: { + updateMessageRequest: (() => Promise.reject(new Error('validation'))) as never, + }, + }, + }); + + // No `enableOfflineSupport`, so `client.offlineDb` is never attached and there is no queue for + // the edit to fall back on. The failure is therefore definitive and must be shown on the + // message — the opposite of the offline-enabled case above, where it must NOT be. + render( + + + { + await flushMountEffects(); + try { + await editMessage({ + localMessage: { ...message, cid: channel.cid, text: editedText }, + options: {}, + }); + } catch (e) { + // do nothing + } + }} + context={MessageInputContext} + > + + + + , + ); + await waitFor(() => expect(screen.getByTestId('children')).toBeTruthy()); + + await waitFor(() => { + const updatedMessage = channel.messagePaginator.getItem(message.id); + + expect(chatClient.offlineDb).toBeUndefined(); + // The edit is never rolled back — reverting would throw away what the user typed. + expect(updatedMessage?.text).toBe(editedText); + expect(updatedMessage?.status).toBe(MessageStatusTypes.FAILED); + }); + }); + }); + describe('pending task execution', () => { it('pending task should be executed after connection is recovered', async () => { const message = channel.messagePaginator.headItems[0]; @@ -1123,13 +1317,33 @@ export const OptimisticUpdates = () => { jest .spyOn(channel, 'watch') .mockResolvedValue({} as Awaited>); + // Without this the reconnect below nukes the offline DB. `client.sync` is a POST, so it + // resolves with the `getOrCreateChannelApi` payload mocked in `beforeEach`, whose `events` is + // undefined; `OfflineDBSyncManager.sync` then throws reading `result.events.length` and its + // catch block calls `resetDB()` — taking the persisted failed message with it. Nothing to do + // with what these tests assert, so give sync an empty, well-formed reply. + jest + .spyOn(chatClient, 'sync') + .mockResolvedValue({ events: [] } as unknown as Awaited< + ReturnType + >); channel.messagePaginator.removeItem({ id: localMessage.id }); channel.messagePaginator.ingestItem(channel.state.formatMessage(serverMessage)); await getOfflineDb(chatClient).deletePendingTask({ id: pendingTask!.id }); await act(async () => { - await getOfflineDb(chatClient).syncManager.invokeSyncStatusListeners(true); + // The real reconnect signal. `invokeSyncStatusListeners(true)` on its own used to be enough + // because `Channel` subscribed to the offline DB's sync-status edge itself; on v10 that moved + // into the LLC's `ConnectionRecoveryManager`, which binds that subscription lazily from its + // `connection.changed` handler and only then reloads the active channels. Driving the edge + // directly therefore reached no subscriber at all — the assertions below never ran against a + // reload. `OfflineDBSyncManager` publishes the edge itself once it has replayed and synced. + markConnectionHealthy(chatClient); + dispatchConnectionChangedEvent(chatClient, true); + // Recovery is detached (`runDetached`), so yield once to let it start before the assertions + // below begin polling. + await flushMountEffects(); }); await waitFor(() => { @@ -1200,12 +1414,32 @@ export const OptimisticUpdates = () => { jest .spyOn(channel, 'watch') .mockResolvedValue({} as Awaited>); + // Without this the reconnect below nukes the offline DB. `client.sync` is a POST, so it + // resolves with the `getOrCreateChannelApi` payload mocked in `beforeEach`, whose `events` is + // undefined; `OfflineDBSyncManager.sync` then throws reading `result.events.length` and its + // catch block calls `resetDB()` — taking the persisted failed message with it. Nothing to do + // with what these tests assert, so give sync an empty, well-formed reply. + jest + .spyOn(chatClient, 'sync') + .mockResolvedValue({ events: [] } as unknown as Awaited< + ReturnType + >); channel.messagePaginator.removeItem({ id: localMessage.id }); await getOfflineDb(chatClient).deletePendingTask({ id: pendingTask!.id }); await act(async () => { - await getOfflineDb(chatClient).syncManager.invokeSyncStatusListeners(true); + // The real reconnect signal. `invokeSyncStatusListeners(true)` on its own used to be enough + // because `Channel` subscribed to the offline DB's sync-status edge itself; on v10 that moved + // into the LLC's `ConnectionRecoveryManager`, which binds that subscription lazily from its + // `connection.changed` handler and only then reloads the active channels. Driving the edge + // directly therefore reached no subscriber at all — the assertions below never ran against a + // reload. `OfflineDBSyncManager` publishes the edge itself once it has replayed and synced. + markConnectionHealthy(chatClient); + dispatchConnectionChangedEvent(chatClient, true); + // Recovery is detached (`runDetached`), so yield once to let it start before the assertions + // below begin polling. + await flushMountEffects(); }); await waitFor(() => { diff --git a/package/src/components/Channel/Channel.tsx b/package/src/components/Channel/Channel.tsx index db91ee5a86..3dbe25c812 100644 --- a/package/src/components/Channel/Channel.tsx +++ b/package/src/components/Channel/Channel.tsx @@ -852,12 +852,18 @@ const ChannelWithContext = (props: PropsWithChildren) = if (!channel) { throw new Error('Channel has not been initialized'); } - // The LLC handles the optimistic local update (ingest into the paginator), the network - // request (honoring any `updateMessageRequest` registered through - // `client.config.set({ channel: { requestHandlers } })`), the received/failed state transitions, - // and offline queueing. - // Thread edits route through the thread instance's own message operations. - await (threadInstance ?? channel).updateMessageWithLocalUpdate({ localMessage, options }); + // The LLC handles the optimistic local update, the network request (honoring any + // `updateMessageRequest` registered through `client.config.set({ channel: { requestHandlers } })`), + // the received/failed state transitions, offline queueing and the offline-DB write. + // + // Routed by MEMBERSHIP rather than "a thread is open", mirroring `useMessageOperations`' + // `sendReaction`: a reply loaded in the open thread is edited through the thread instance, and + // anything else — including the thread's own PARENT message, which the reply paginator cannot + // hold — through the channel. + const target = threadInstance?.messagePaginator.getItem(localMessage.id) + ? threadInstance + : channel; + await target.updateMessageWithLocalUpdate({ localMessage, options }); }, ); diff --git a/package/src/components/Message/hooks/useMessageOperations.ts b/package/src/components/Message/hooks/useMessageOperations.ts index f5fbe3027a..fe930e5371 100644 --- a/package/src/components/Message/hooks/useMessageOperations.ts +++ b/package/src/components/Message/hooks/useMessageOperations.ts @@ -126,9 +126,17 @@ export const useMessageOperations = (): MessageOperations => { options = { hard: true }; } - // The LLC performs the delete request (honoring any configState delete handler) and ingests - // the deleted message into the paginator. Thread deletes route through the thread instance. - await (threadInstance ?? channel).deleteMessageWithLocalUpdate({ + // The LLC owns the whole delete lifecycle: the optimistic `deleted` marking (or removal, for a + // hard delete), the request (honoring any configState delete handler), the offline-DB write, and + // the revert if the delete is definitively rejected. + // + // Routed by MEMBERSHIP, like `sendReaction` above: a reply loaded in the open thread goes through + // the thread instance, everything else — including the thread's own parent message, which the + // reply paginator cannot hold — through the channel. + const target = threadInstance?.messagePaginator.getItem(message.id) + ? threadInstance + : channel; + await target.deleteMessageWithLocalUpdate({ localMessage: message, options, }); diff --git a/package/src/mock-builders/DB/mock.ts b/package/src/mock-builders/DB/mock.ts index 8d03370c98..2d03d32331 100644 --- a/package/src/mock-builders/DB/mock.ts +++ b/package/src/mock-builders/DB/mock.ts @@ -32,11 +32,20 @@ export const sqliteMock = { rmSync(testDbName, { force: true }); }, execute: async (queryInput: string, params: unknown[]) => { - const query = queryInput.trim().toLowerCase(); + const query = queryInput.trim(); + // Lower-cased COPY, used only to classify the statement and to parse PRAGMA tokens. The query + // itself must be executed with its original casing: SQL keywords are case-insensitive, but + // string literals are not — and `selectMessagesForChannels` builds its result rows with + // `json_object('extraData', a.extraData, ...)`, whose keys are literals. Lower-casing the whole + // statement renamed every one of those keys (`extradata`, `createdAt` -> `createdat`, ...), so + // `mapStorableToMessage`'s destructuring silently produced `undefined` for every camelCase + // field — including the `extraData` blob that carries `status`. op-sqlite runs the SQL as + // written, so this only ever misled tests. + const classifier = query.toLowerCase(); const stmt = db.prepare(query); let result: unknown[] = []; - if (query.indexOf('select') === 0) { + if (classifier.indexOf('select') === 0) { const modifiedParams = params?.map((p) => (typeof p === 'boolean' ? Number(p) : p)) || []; result = await new Promise((resolve) => resolve(stmt.all(modifiedParams))); @@ -47,8 +56,8 @@ export const sqliteMock = { }; } - if (query.indexOf('pragma') === 0) { - const pragmaQueryTokens = query.split(' '); + if (classifier.indexOf('pragma') === 0) { + const pragmaQueryTokens = classifier.split(' '); if (pragmaQueryTokens[2] === '=') { db.pragma(`${pragmaQueryTokens[1]} = ${pragmaQueryTokens[3]}`); } else { diff --git a/package/src/store/__tests__/channelExists.test.ts b/package/src/store/__tests__/channelExists.test.ts new file mode 100644 index 0000000000..b9a85a8070 --- /dev/null +++ b/package/src/store/__tests__/channelExists.test.ts @@ -0,0 +1,44 @@ +import { generateChannelResponse } from '../../mock-builders/generator/channel'; +import { BetterSqlite } from '../../test-utils/BetterSqlite'; +import { channelExists } from '../apis/channelExists'; +import { upsertChannels } from '../apis/upsertChannels'; +import { SqliteClient } from '../SqliteClient'; + +/** + * Runs against real SQLite rather than a mocked DB on purpose: the bug this guards was in the SQL + * itself, so a mock that answers `true`/`false` on command would have kept passing forever. + */ +describe('channelExists', () => { + beforeEach(async () => { + await SqliteClient.initializeDatabase(); + await BetterSqlite.openDB(); + }); + + afterEach(() => { + BetterSqlite.dropAllTables(); + BetterSqlite.closeDB(); + }); + + it('reports false for a channel the database does not have', async () => { + expect(await channelExists({ cid: 'messaging:never-persisted' })).toBe(false); + }); + + it('reports true for a channel the database does have', async () => { + const channelResponse = generateChannelResponse({ members: [], messages: [] }); + await upsertChannels({ + channels: [channelResponse] as unknown as Parameters[0]['channels'], + }); + + expect(await channelExists({ cid: channelResponse.channel.cid as string })).toBe(true); + }); + + it('distinguishes one channel from another', async () => { + const persisted = generateChannelResponse({ members: [], messages: [] }); + await upsertChannels({ + channels: [persisted] as unknown as Parameters[0]['channels'], + }); + + expect(await channelExists({ cid: persisted.channel.cid as string })).toBe(true); + expect(await channelExists({ cid: 'messaging:some-other-channel' })).toBe(false); + }); +}); diff --git a/package/src/store/__tests__/insertReaction.test.ts b/package/src/store/__tests__/insertReaction.test.ts new file mode 100644 index 0000000000..1dd4640f62 --- /dev/null +++ b/package/src/store/__tests__/insertReaction.test.ts @@ -0,0 +1,107 @@ +import { generateChannelResponse } from '../../mock-builders/generator/channel'; +import { generateMessage } from '../../mock-builders/generator/message'; +import { generateReaction } from '../../mock-builders/generator/reaction'; +import { BetterSqlite } from '../../test-utils/BetterSqlite'; +import { insertReaction } from '../apis/insertReaction'; +import { updateReaction } from '../apis/updateReaction'; +import { upsertChannels } from '../apis/upsertChannels'; +import { upsertMessages } from '../apis/upsertMessages'; +import { SqliteClient } from '../SqliteClient'; + +/** + * Runs against real SQLite rather than a mocked DB on purpose: what this guards is a foreign key + * declared in the schema, and a mocked `executeSqlBatch` accepts any statement you hand it. + */ +describe('reaction writes when the message is not cached', () => { + const cid = 'messaging:reaction-guard'; + + const cacheAMessage = async (id: string) => { + const channelResponse = generateChannelResponse({ members: [], messages: [] }); + channelResponse.channel.cid = cid; + await upsertChannels({ + channels: [channelResponse] as unknown as Parameters[0]['channels'], + }); + const message = generateMessage({ cid, id }); + await upsertMessages({ + messages: [message] as unknown as Parameters[0]['messages'], + }); + return message; + }; + + const storedReactions = () => BetterSqlite.selectFromTable('reactions'); + + beforeEach(async () => { + await SqliteClient.initializeDatabase(); + await BetterSqlite.openDB(); + }); + + afterEach(() => { + BetterSqlite.dropAllTables(); + BetterSqlite.closeDB(); + }); + + it('inserts the reaction when the message is cached', async () => { + const message = await cacheAMessage('cached-message'); + const reaction = generateReaction({ message_id: message.id, type: 'love' }); + + await insertReaction({ message, reaction }); + + expect(await storedReactions()).toHaveLength(1); + }); + + // The failure this closes: a `/sync` replay carrying a reaction on a message outside the cached + // window aborted the whole batch with `FOREIGN KEY constraint failed`, so all 45 unrelated events + // in it were lost too. + it('skips the reaction, without throwing, when the message was never cached', async () => { + await cacheAMessage('some-other-message'); + const reaction = generateReaction({ message_id: 'never-persisted', type: 'love' }); + + await expect( + insertReaction({ + message: { id: 'never-persisted', reaction_groups: {} } as Parameters< + typeof insertReaction + >[0]['message'], + reaction, + }), + ).resolves.not.toThrow(); + + expect(await storedReactions()).toHaveLength(0); + }); + + it('does not abort the rest of the batch it shares', async () => { + const message = await cacheAMessage('cached-message'); + const orphanQueries = await insertReaction({ + execute: false, + message: { id: 'never-persisted', reaction_groups: {} } as Parameters< + typeof insertReaction + >[0]['message'], + reaction: generateReaction({ message_id: 'never-persisted', type: 'like' }), + }); + const validQueries = await insertReaction({ + execute: false, + message, + reaction: generateReaction({ message_id: message.id, type: 'love' }), + }); + + await SqliteClient.executeSqlBatch([...orphanQueries, ...validQueries]); + + // The orphan is dropped and the reaction that had a parent still lands. + expect(await storedReactions()).toHaveLength(1); + }); + + it('applies the same guard to updateReaction', async () => { + await cacheAMessage('some-other-message'); + const reaction = generateReaction({ message_id: 'never-persisted', type: 'love' }); + + await expect( + updateReaction({ + message: { id: 'never-persisted', reaction_groups: {} } as Parameters< + typeof updateReaction + >[0]['message'], + reaction, + }), + ).resolves.not.toThrow(); + + expect(await storedReactions()).toHaveLength(0); + }); +}); diff --git a/package/src/store/apis/channelExists.ts b/package/src/store/apis/channelExists.ts index 6c4a264b63..5b57015719 100644 --- a/package/src/store/apis/channelExists.ts +++ b/package/src/store/apis/channelExists.ts @@ -1,10 +1,18 @@ import { SqliteClient } from '../SqliteClient'; +/** + * Whether a channel row exists, which callers use to avoid writing a row whose `cid` foreign key + * would not resolve. + * + * Deliberately `SELECT 1 ... LIMIT 1` rather than `SELECT EXISTS(...)`: `EXISTS` always returns + * exactly one row (holding `0` or `1`), so the row COUNT carries no information and the previous + * implementation reported `true` for every cid, existing or not. Returning zero rows for a miss is + * what makes the answer readable without depending on the result column's name. + */ export const channelExists = async ({ cid }: { cid: string }) => { - const channels = await SqliteClient.executeSql( - 'SELECT EXISTS(SELECT 1 FROM channels WHERE cid = ?)', - [cid], - ); + const channels = await SqliteClient.executeSql('SELECT 1 FROM channels WHERE cid = ? LIMIT 1', [ + cid, + ]); SqliteClient.logger?.('info', 'channelExists', { cid, diff --git a/package/src/store/apis/insertReaction.ts b/package/src/store/apis/insertReaction.ts index da1c2006af..72f218e232 100644 --- a/package/src/store/apis/insertReaction.ts +++ b/package/src/store/apis/insertReaction.ts @@ -2,7 +2,7 @@ import type { LocalMessage, MessageResponse, ReactionResponse } from 'stream-cha import { mapReactionToStorable } from '../mappers/mapReactionToStorable'; import { createUpdateQuery } from '../sqlite-utils/createUpdateQuery'; -import { createUpsertQuery } from '../sqlite-utils/createUpsertQuery'; +import { createUpsertQueryIfParentExists } from '../sqlite-utils/createUpsertQueryIfParentExists'; import { SqliteClient } from '../SqliteClient'; import type { PreparedQueries } from '../types'; @@ -19,7 +19,17 @@ export const insertReaction = async ({ const storableReaction = mapReactionToStorable(reaction); - queries.push(createUpsertQuery('reactions', storableReaction)); + // Only a channel's cached window of messages is stored, so a reaction can arrive for a message + // this database has never held - an old one someone reacts to, or a `/sync` replay after a cold + // start. Writing it anyway violates the `reactions.messageId` foreign key and aborts the whole + // batch it travels in. + queries.push( + createUpsertQueryIfParentExists('reactions', storableReaction, { + column: 'id', + table: 'messages', + value: reaction.message_id, + }), + ); const stringifiedNewReactionGroups = JSON.stringify(message.reaction_groups); diff --git a/package/src/store/apis/updateReaction.ts b/package/src/store/apis/updateReaction.ts index ef919bb5c6..59ea3cc86d 100644 --- a/package/src/store/apis/updateReaction.ts +++ b/package/src/store/apis/updateReaction.ts @@ -6,6 +6,7 @@ import { mapUserToStorable } from '../mappers/mapUserToStorable'; import { createDeleteQuery } from '../sqlite-utils/createDeleteQuery'; import { createUpdateQuery } from '../sqlite-utils/createUpdateQuery'; import { createUpsertQuery } from '../sqlite-utils/createUpsertQuery'; +import { createUpsertQueryIfParentExists } from '../sqlite-utils/createUpsertQueryIfParentExists'; import { SqliteClient } from '../SqliteClient'; import type { PreparedQueries } from '../types'; @@ -34,7 +35,13 @@ export const updateReaction = async ({ userId: reaction.user_id, }), ); - queries.push(createUpsertQuery('reactions', storableReaction)); + queries.push( + createUpsertQueryIfParentExists('reactions', storableReaction, { + column: 'id', + table: 'messages', + value: reaction.message_id, + }), + ); let updatedReactionGroups: string | undefined; if (message.reaction_groups) { diff --git a/package/src/store/sqlite-utils/createUpsertQuery.ts b/package/src/store/sqlite-utils/createUpsertQuery.ts index 809c789298..b5d90418b1 100644 --- a/package/src/store/sqlite-utils/createUpsertQuery.ts +++ b/package/src/store/sqlite-utils/createUpsertQuery.ts @@ -2,18 +2,17 @@ import { Schema, tables } from '../schema'; import type { PreparedQueries, TableColumnNames, TableRow } from '../types'; /** - * Creates a simple upsert query for sqlite. + * The pieces every upsert statement is built from, shared with + * {@link import('./createUpsertQueryIfParentExists').createUpsertQueryIfParentExists} so the two + * forms cannot drift in how they filter columns or resolve conflict keys. * - * @param {string} table Table name - * @param {Object} row Table row to insert or update. - * @param {Array} conflictCheckKeys Custom list of columns to check conflicts for - https://www.sqlite.org/lang_UPSERT.html. By default conflicts are checked on primary keys. - * @returns {string} Final upsert query for sqlite + * @internal */ -export const createUpsertQuery = ( +export const upsertStatementParts = ( table: T, row: Partial>, conflictCheckKeys?: Array>, -): PreparedQueries => { +) => { const filteredRow: typeof row = {}; // In case of "DO UPDATE SET", we only want to update the properties which @@ -41,8 +40,35 @@ export const createUpsertQuery = ( ${conflictMatchersWithoutPK.join(',')}` : ''; + return { + columns: fields.join(','), + conflictConstraint, + questionMarks, + values: Object.values(filteredRow), + }; +}; + +/** + * Creates a simple upsert query for sqlite. + * + * @param {string} table Table name + * @param {Object} row Table row to insert or update. + * @param {Array} conflictCheckKeys Custom list of columns to check conflicts for - https://www.sqlite.org/lang_UPSERT.html. By default conflicts are checked on primary keys. + * @returns {string} Final upsert query for sqlite + */ +export const createUpsertQuery = ( + table: T, + row: Partial>, + conflictCheckKeys?: Array>, +): PreparedQueries => { + const { columns, conflictConstraint, questionMarks, values } = upsertStatementParts( + table, + row, + conflictCheckKeys, + ); + return [ - `INSERT INTO ${table} (${fields.join(',')}) VALUES (${questionMarks}) ${conflictConstraint}`, - Object.values(filteredRow), + `INSERT INTO ${table} (${columns}) VALUES (${questionMarks}) ${conflictConstraint}`, + values, ]; }; diff --git a/package/src/store/sqlite-utils/createUpsertQueryIfParentExists.ts b/package/src/store/sqlite-utils/createUpsertQueryIfParentExists.ts new file mode 100644 index 0000000000..f31440bc38 --- /dev/null +++ b/package/src/store/sqlite-utils/createUpsertQueryIfParentExists.ts @@ -0,0 +1,58 @@ +import { upsertStatementParts } from './createUpsertQuery'; + +import { Schema } from '../schema'; +import type { PreparedQueries, TableColumnNames, TableRow } from '../types'; + +/** + * The row this write depends on - the parent side of a foreign key declared in {@link Schema}. + */ +type ParentRow = { + column: string; + table: keyof Schema; + value: unknown; +}; + +/** + * An upsert that writes nothing at all when the row it references is absent. + * + * Deliberately NOT an option on `createUpsertQuery`: "insert or update" is a contract worth keeping + * exact, and this is a third thing - "insert or update, or do nothing" - so it says so in its name + * rather than hiding behind a flag. + * + * For the tables whose schema declares a foreign key. Only part of a channel's messages are ever + * cached, so a child row can genuinely arrive for a parent this database has never held - a reaction + * on an old message, say. SQLite rejects that child, and since these queries are executed as one + * batch, the one rejected statement aborts every unrelated write alongside it. + * + * Skipping is the honest outcome rather than the lesser evil: these tables mirror what is held + * locally, so a child with no parent has nothing to hang off, and it comes back on its own once the + * parent is cached (a message arrives carrying its own reactions). + * + * Expressed as `WHERE EXISTS` inside the statement rather than as a separate probe so it costs no + * extra round trip and cannot race with the write it guards - and the lookup it does is the same + * index lookup the foreign key already forces on every insert. SQLite requires that `WHERE` when an + * `INSERT ... SELECT` is followed by `ON CONFLICT`, so the clause doing the guarding is also what + * keeps the upsert unambiguous. + * + * @param table Table name. + * @param row Table row to insert or update. + * @param parent The row that must exist for anything to be written. + * @param conflictCheckKeys Custom list of columns to check conflicts for. Defaults to primary keys. + */ +export const createUpsertQueryIfParentExists = ( + table: T, + row: Partial>, + parent: ParentRow, + conflictCheckKeys?: Array>, +): PreparedQueries => { + const { columns, conflictConstraint, questionMarks, values } = upsertStatementParts( + table, + row, + conflictCheckKeys, + ); + + return [ + `INSERT INTO ${table} (${columns}) SELECT ${questionMarks} WHERE EXISTS (SELECT 1 FROM ${parent.table} WHERE ${parent.column} = ?) ${conflictConstraint}`, + [...values, parent.value], + ]; +};