diff --git a/.gitattributes b/.gitattributes index 2d6e25f1b09..74eb6ffa7f5 100644 --- a/.gitattributes +++ b/.gitattributes @@ -1,2 +1,3 @@ +*.patch text diff *.pbxproj -text pnpm-lock.yaml linguist-generated=true -diff diff --git a/CONTEXT.md b/CONTEXT.md index 1399e8e8b1d..25506a042fc 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -180,11 +180,13 @@ A **Message Action** is the active mode on a Message in the Room view. The three ## Server & Connection -| Term | Definition | Aliases to avoid | -| ------------------ | ------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------- | -| **Server** | A Rocket.Chat server instance the app connects to, with version, settings, and enterprise modules | Workspace (used by web but not consistently in mobile), instance | -| **Server History** | List of previously connected Servers for quick reconnection | Recent servers | -| **Meteor Connect** | The WebSocket connection to the Server's DDP (Distributed Data Protocol) endpoint | Socket, connection | +| Term | Definition | Aliases to avoid | +| --------------------- | ------------------------------------------------------------------------------------------------------------------ | ---------------------------------------------------------------- | +| **Server** | A Rocket.Chat server instance the app connects to, with version, settings, and enterprise modules | Workspace (used by web but not consistently in mobile), instance | +| **Server History** | List of previously connected Servers for quick reconnection | Recent servers | +| **Meteor Connect** | The WebSocket connection to the Server's DDP (Distributed Data Protocol) endpoint | Socket, connection | +| **Zombie Connection** | A live socket (`connected`) with no authenticated user and zero live subscriptions | Stranded session, dead connection | +| **Resubscribe** | The SDK re-sending its preserved `subscriptions` registry on every DDP login via `subscribeAll()`, reusing sub ids | Re-bind, registry replay | ## Navigation & Layout diff --git a/app/lib/database/__tests__/advanceSyncCursor.test.ts b/app/lib/database/__tests__/advanceSyncCursor.test.ts new file mode 100644 index 00000000000..50d98fe1d30 --- /dev/null +++ b/app/lib/database/__tests__/advanceSyncCursor.test.ts @@ -0,0 +1,173 @@ +import type { IMessage, TSubscriptionModel } from '../../../definitions'; +import type { TAppDatabase } from '../interfaces'; +import { advanceSyncCursor, maxUpdatedAt } from '../../methods/helpers/advanceSyncCursor'; +import log from '../../methods/helpers/log'; +import { + closeLokiTestDatabase, + createLokiTestDatabase, + resetLokiTestDatabase, + seedSubscription, + withWriterQueueDiagnosticCleared +} from './lokiTestDatabase'; + +// The real `advanceSyncCursor` reads `database.active` (via itself and via the real +// `getSubscriptionByRoomId` service). Point both at the live LokiJS database so the +// forward-only cursor and the re-read-inside-write guard run against the real WMDB +// writer lock — the behavior the `database.active` object-mock cannot prove. +let mockActiveDatabase: TAppDatabase; +jest.mock('../../database', () => ({ + __esModule: true, + default: { + get active() { + return mockActiveDatabase; + } + } +})); + +jest.mock('../../methods/helpers/log', () => ({ + __esModule: true, + default: jest.fn() +})); + +const mockedLog = log as jest.MockedFunction; + +const RID = 'room-1'; +const T0 = Date.UTC(2026, 6, 22, 12, 0, 0); +const T1 = T0 + 1000; +const T2 = T0 + 2000; +const T3 = T0 + 3000; +const T5 = T0 + 5000; + +const message = (updatedAt?: number): IMessage => + ({ _updatedAt: updatedAt === undefined ? undefined : new Date(updatedAt) } as unknown as IMessage); + +const flush = () => new Promise(resolve => setImmediate(resolve)); + +const persistedLastOpen = async (rid = RID): Promise => { + const subscription = (await mockActiveDatabase.get('subscriptions').find(rid)) as TSubscriptionModel; + return subscription.lastOpen?.getTime(); +}; + +describe('advanceSyncCursor (LokiJS integration)', () => { + beforeAll(() => { + mockActiveDatabase = createLokiTestDatabase(); + }); + + afterAll(() => closeLokiTestDatabase(mockActiveDatabase)); + + beforeEach(async () => { + await resetLokiTestDatabase(mockActiveDatabase); + mockedLog.mockClear(); + }); + + it('advances lastOpen to the max _updatedAt of an out-of-order batch', async () => { + await seedSubscription(mockActiveDatabase, { rid: RID, lastOpen: new Date(T0) }); + + await advanceSyncCursor(RID, maxUpdatedAt([message(T1), message(T3), message(T2)])); + + expect(await persistedLastOpen()).toBe(T3); + }); + + it('is forward-only: a batch older than the cursor is a no-op', async () => { + await seedSubscription(mockActiveDatabase, { rid: RID, lastOpen: new Date(T5) }); + + await advanceSyncCursor(RID, maxUpdatedAt([message(T2)])); + + expect(await persistedLastOpen()).toBe(T5); + }); + + it('is forward-only: latest equal to the cursor is a no-op', async () => { + await seedSubscription(mockActiveDatabase, { rid: RID, lastOpen: new Date(T3) }); + + await advanceSyncCursor(RID, maxUpdatedAt([message(T3)])); + + expect(await persistedLastOpen()).toBe(T3); + }); + + it('never advances past an empty batch', async () => { + await seedSubscription(mockActiveDatabase, { rid: RID, lastOpen: new Date(T2) }); + + await advanceSyncCursor(RID, maxUpdatedAt([])); + + expect(await persistedLastOpen()).toBe(T2); + }); + + it('never advances when every message is missing _updatedAt', async () => { + await seedSubscription(mockActiveDatabase, { rid: RID, lastOpen: new Date(T2) }); + + await advanceSyncCursor(RID, maxUpdatedAt([message(undefined), message(undefined)])); + + expect(await persistedLastOpen()).toBe(T2); + }); + + it('skips messages missing _updatedAt without poisoning the max', async () => { + await seedSubscription(mockActiveDatabase, { rid: RID, lastOpen: new Date(T0) }); + + await advanceSyncCursor(RID, maxUpdatedAt([message(undefined), message(T3), message(undefined)])); + + expect(await persistedLastOpen()).toBe(T3); + }); + + // There is nowhere to persist a cursor without a row, so the no-op stands. What must not be + // lost is the batch itself — `nullCursorRecurrence.integration.test.ts` covers the recovery, + // where the row lands cursor-less and the next sync re-delivers off `ls`/`ts`. + it('returns cleanly when no subscription exists for the rid', async () => { + await expect(advanceSyncCursor('missing-room', maxUpdatedAt([message(T3)]))).resolves.toBeUndefined(); + + const count = await mockActiveDatabase.get('subscriptions').query().fetchCount(); + expect(count).toBe(0); + expect(mockedLog).not.toHaveBeenCalled(); + }); + + it('re-read guard: a concurrent higher cursor is not regressed by an in-flight lower advance', async () => { + await seedSubscription(mockActiveDatabase, { rid: RID, lastOpen: new Date(T0) }); + const subscription = (await mockActiveDatabase.get('subscriptions').find(RID)) as TSubscriptionModel; + + // A concurrent sync holds the writer lock and, once released, commits the higher T3. + let openGate: () => void = () => {}; + const gate = new Promise(resolve => { + openGate = resolve; + }); + let signalInside: () => void = () => {}; + const insideWriteLock = new Promise(resolve => { + signalInside = resolve; + }); + const concurrentAdvance = mockActiveDatabase.write(async () => { + signalInside(); + await gate; + await subscription.update((s: TSubscriptionModel) => { + s.lastOpen = new Date(T3); + }); + }); + + await insideWriteLock; + + // advanceSyncCursor reads the cursor still at T0 (reads don't take the writer lock), + // passes the forward-only gate, then queues its own db.write behind the concurrent one. + // Queueing behind a running writer arms WatermelonDB's dev-only ~1.5s diagnostic timer, + // which would outlive Jest's exit window — clear it around the contended section. + await withWriterQueueDiagnosticCleared(async () => { + const inFlightAdvance = advanceSyncCursor(RID, maxUpdatedAt([message(T2)])); + await flush(); + + // Release the concurrent write: it commits T3 first, then the in-flight write body runs, + // re-reads the now-committed T3, and must skip its stale T2 rather than regress the cursor. + openGate(); + await Promise.all([concurrentAdvance, inFlightAdvance]); + }); + + expect(await persistedLastOpen()).toBe(T3); + }); + + it('swallows and logs a thrown db error instead of propagating', async () => { + await seedSubscription(mockActiveDatabase, { rid: RID, lastOpen: new Date(T0) }); + const error = new Error('boom'); + const writeSpy = jest.spyOn(mockActiveDatabase, 'write').mockRejectedValueOnce(error); + + await expect(advanceSyncCursor(RID, maxUpdatedAt([message(T3)]))).resolves.toBeUndefined(); + + expect(mockedLog).toHaveBeenCalledWith(error); + expect(await persistedLastOpen()).toBe(T0); + writeSpy.mockRestore(); + }); +}); diff --git a/app/lib/database/__tests__/cursorClockPoisoning.integration.test.ts b/app/lib/database/__tests__/cursorClockPoisoning.integration.test.ts new file mode 100644 index 00000000000..8819432476c --- /dev/null +++ b/app/lib/database/__tests__/cursorClockPoisoning.integration.test.ts @@ -0,0 +1,190 @@ +import { Q } from '@nozbe/watermelondb'; + +import type { TAppDatabase } from '../interfaces'; +import type { IMessage, TMessageModel, TSubscriptionModel } from '../../../definitions'; +import { closeLokiTestDatabase, createLokiTestDatabase, resetLokiTestDatabase, seedSubscription } from './lokiTestDatabase'; +import { createFakeSyncServer, type IFakeServerMessage } from './fakeSyncServer'; +import { MessageTypeLoad } from '../../constants/messageTypeLoad'; +import sdk from '../../services/sdk'; +import { loadMessagesForRoom } from '../../methods/loadMessagesForRoom'; +import { loadMissedMessages } from '../../methods/loadMissedMessages'; + +// Real persistence + real cursor: `database.active` points at the live LokiJS DB so +// `loadMessagesForRoom` drives the real `updateMessages` (and therefore the real +// `buildMessage`/`normalizeMessage` in-place mutation) plus the real `advanceSyncCursor`. +// Only the outer edges are mocked: `sdk` (the *.history + chat.syncMessages network), +// `auxStore` (server version + settings + dispatch), `encryption` (pass-through), `log`. +let mockActiveDatabase: TAppDatabase; +jest.mock('../../database', () => ({ + __esModule: true, + default: { + get active() { + return mockActiveDatabase; + } + } +})); + +jest.mock('../../store/auxStore', () => ({ + store: { + getState: () => ({ + server: { version: '8.5.1' }, + settings: { Hide_System_Messages: ['uj'] } + }), + dispatch: jest.fn() + } +})); + +jest.mock('../../services/sdk', () => ({ + __esModule: true, + default: { get: jest.fn() } +})); + +jest.mock('../../encryption', () => ({ + Encryption: { decryptMessages: (messages: unknown) => Promise.resolve(messages) } +})); + +jest.mock('../../methods/helpers/log', () => ({ __esModule: true, default: jest.fn() })); + +/** Structural shape of the `sdk.get` mock, so this file needn't depend on the `jest` mock types. */ +type TSdkGetRouter = { + mockImplementation: (handler: (endpoint: string, params: { type?: string; next?: number | null }) => unknown) => void; +}; + +const mockedSdkGet = sdk.get as jest.MockedFunction; +const server = createFakeSyncServer(); + +const RID = 'room-1'; +const COUNT = 50; +// Deliberately years behind the device clock: any cursor stamped from `new Date()` +// lands unmistakably above every server timestamp in these fixtures. +const SERVER_EPOCH = Date.UTC(2020, 0, 1, 12, 0, 0); +const SEED_CURSOR = SERVER_EPOCH - 60_000; +const MISSED_TS = SERVER_EPOCH + 60_000; + +const historyMessage = (id: string, ts: number, t?: string): IMessage => + ({ + _id: id, + rid: RID, + msg: `msg-${id}`, + ts: new Date(ts).toISOString(), + _updatedAt: new Date(ts).toISOString(), + u: { _id: 'user-1', username: 'user-1' }, + ...(t && { t }) + } as unknown as IMessage); + +const serverMessage = (id: string, updatedAt: number): IFakeServerMessage => ({ + _id: id, + rid: RID, + msg: `msg-${id}`, + ts: new Date(updatedAt).toISOString(), + _updatedAt: updatedAt +}); + +// Newest-first, exactly COUNT long: what the server returns when more history exists above. +const fullPage = (idPrefix: string, newest: number, hiddenType?: string): IMessage[] => + Array.from({ length: COUNT }, (_, index) => historyMessage(`${idPrefix}-${index}`, newest - index * 1000, hiddenType)); + +const historyPages: IMessage[][] = []; +const answerHistoryOnce = (messages: IMessage[]) => historyPages.push(messages); + +// One router for both endpoints a room open hits: the *.history pages queued by the test, +// and the shared `chat.syncMessages` fake reading the cursor the history load left behind. +const installSdkRouter = () => + (mockedSdkGet as unknown as TSdkGetRouter).mockImplementation((endpoint, params) => { + if (endpoint === 'chat.syncMessages') { + return Promise.resolve(server.handleSyncMessages(params)); + } + if (endpoint.endsWith('.history')) { + return Promise.resolve({ success: true, messages: historyPages.shift() ?? [] }); + } + throw new Error(`Unexpected endpoint ${endpoint}`); + }); + +const newestUpdatedAt = (messages: IMessage[]): number => + messages.reduce((max, message) => Math.max(max, new Date(message._updatedAt).getTime()), 0); + +const persistedMessageIds = async (): Promise => { + const rows = (await mockActiveDatabase.get('messages').query(Q.where('rid', RID)).fetch()) as TMessageModel[]; + return rows.map(row => row.id).sort(); +}; + +const persistedLastOpen = async (rid = RID): Promise => { + const subscription = (await mockActiveDatabase.get('subscriptions').find(rid)) as TSubscriptionModel; + return subscription.lastOpen?.getTime(); +}; + +const persistedLoadMoreCount = async (): Promise => { + const rows = (await mockActiveDatabase + .get('messages') + .query(Q.where('rid', RID), Q.where('t', MessageTypeLoad.MORE)) + .fetch()) as TMessageModel[]; + return rows.length; +}; + +describe('sync cursor is never stamped from the device clock (LokiJS integration)', () => { + beforeAll(() => { + mockActiveDatabase = createLokiTestDatabase(); + }); + + afterAll(() => closeLokiTestDatabase(mockActiveDatabase)); + + beforeEach(async () => { + await resetLokiTestDatabase(mockActiveDatabase); + server.reset(); + historyPages.length = 0; + mockedSdkGet.mockReset(); + installSdkRouter(); + }); + + // The user-visible symptom of issue #7499: a full first history page appends a synthetic + // load-more row to the array handed to advanceSyncCursor, normalizeMessage stamps that row's + // missing `_updatedAt` with `new Date()`, and the next sync asks the server for changes + // newer than the DEVICE clock — so everything the server wrote in between is skipped forever. + it('delivers a message written after the history window but before the device clock', async () => { + await seedSubscription(mockActiveDatabase, { rid: RID, lastOpen: new Date(SEED_CURSOR) }); + answerHistoryOnce(fullPage('m', SERVER_EPOCH)); + server.updated.push(serverMessage('missed-1', MISSED_TS)); + + await loadMessagesForRoom({ rid: RID, t: 'c' }); + await loadMissedMessages({ rid: RID }); + + expect(await persistedMessageIds()).toContain('missed-1'); + expect(await persistedLastOpen()).toBe(MISSED_TS); + }); + + it('advances the cursor to the newest server timestamp when the first history page is full', async () => { + await seedSubscription(mockActiveDatabase, { rid: RID, lastOpen: new Date(SEED_CURSOR) }); + const page = fullPage('m', SERVER_EPOCH); + answerHistoryOnce(page); + + await loadMessagesForRoom({ rid: RID, t: 'c' }); + + // the synthetic loader really was inserted — this is the poisoning path, not a no-op + expect(await persistedLoadMoreCount()).toBe(1); + expect(await persistedLastOpen()).toBe(newestUpdatedAt(page)); + }); + + it('advances the cursor to the server max on a partial page', async () => { + await seedSubscription(mockActiveDatabase, { rid: RID, lastOpen: new Date(SEED_CURSOR) }); + const page = [historyMessage('m1', SERVER_EPOCH - 2000), historyMessage('m2', SERVER_EPOCH)]; + answerHistoryOnce(page); + + await loadMessagesForRoom({ rid: RID, t: 'c' }); + + expect(await persistedLoadMoreCount()).toBe(0); + expect(await persistedLastOpen()).toBe(newestUpdatedAt(page)); + }); + + it('advances to the server max when a full page is followed by a partial one', async () => { + await seedSubscription(mockActiveDatabase, { rid: RID, lastOpen: new Date(SEED_CURSOR) }); + // every message in the first page is a hidden system message, so the loader recurses + const hiddenPage = fullPage('hidden', SERVER_EPOCH, 'uj'); + const tailPage = [historyMessage('tail-1', SERVER_EPOCH - COUNT * 1000)]; + answerHistoryOnce(hiddenPage); + answerHistoryOnce(tailPage); + + await loadMessagesForRoom({ rid: RID, t: 'c' }); + + expect(await persistedLastOpen()).toBe(newestUpdatedAt([...hiddenPage, ...tailPage])); + }); +}); diff --git a/app/lib/database/__tests__/fakeSyncServer.ts b/app/lib/database/__tests__/fakeSyncServer.ts new file mode 100644 index 00000000000..72a9e897cbc --- /dev/null +++ b/app/lib/database/__tests__/fakeSyncServer.ts @@ -0,0 +1,143 @@ +/** + * In-memory fake of the `chat.syncMessages` (>=7.1) cursor API, generalized from + * `app/lib/methods/loadMissedMessages.test.ts` so every sync-integration test can + * share one server. `sdk` stays the only mock — this drives `sdk.get`. + * + * The real endpoint answers two independent cursors keyed by `type`: UPDATED + * (created/edited) paginates by `_updatedAt`, DELETED by `_deletedAt`. Both return + * records strictly newer than the requested `next` cursor, page by page, and succeed + * on an empty page. Deleted records are projected down to `{ _id, _deletedAt }`, so + * they carry no message body and no `_updatedAt`. + * + * This models the DELETED cursor as intended, not as shipped: the server's + * `mountCursorQuery` sorts unconditionally by `_updatedAt`, which the DELETED projection + * strips, so its real `next` is not guaranteed to be the page's max `_deletedAt`. Do not + * read this fake as ground truth for that stream's ordering. + */ + +export interface IFakeServerMessage { + _id: string; + rid: string; + msg: string; + ts: string; + _updatedAt: number; +} + +export interface IFakeServerDeletedMessage { + _id: string; + _deletedAt: number; + /** + * Not part of the real projection. Only for tests proving the client ignores an + * overstated deleted payload rather than feeding it to the sync cursor. + */ + _updatedAt?: number; +} + +/** As sent over the wire: the server serializes `_updatedAt` to an ISO string. */ +export type TFakeServerMessageWire = Omit & { _updatedAt: string }; + +/** As sent over the wire: `{ _id, _deletedAt }`, both timestamps ISO strings. */ +export type TFakeServerDeletedWire = { _id: string; _deletedAt: string; _updatedAt?: string }; + +export interface IFakeSyncServerOptions { + updatedPageSize?: number; + deletedPageSize?: number; +} + +/** Structural shape of an `sdk.get` jest mock — avoids depending on the `jest` global type here. */ +type TSdkGetMock = { + mockImplementation: (fn: (endpoint: string, params: { type?: string; next?: number | null }) => unknown) => void; +}; + +export interface IFakeSyncServer { + updated: IFakeServerMessage[]; + deleted: IFakeServerDeletedMessage[]; + /** Reject the Nth UPDATED page request (1-based) to model a mid-pagination network error. */ + failUpdatedPageAtRequest: number | null; + /** Reject the Nth DELETED page request (1-based). */ + failDeletedPageAtRequest: number | null; + /** Clears messages, cursors, counters and failure injection — call in `beforeEach`. */ + reset(): void; + /** Answers a single `chat.syncMessages` request the way the >=7.1 API does. */ + handleSyncMessages(params: { type?: string; next?: number | null }): { + result: { + updated?: TFakeServerMessageWire[]; + deleted?: TFakeServerDeletedWire[]; + cursor: { next: number | null; previous: number | null }; + }; + }; + /** Wires this server into an `sdk.get` jest mock; throws on any other endpoint. */ + installOn(sdkGet: TSdkGetMock): void; +} + +const serialize = (message: IFakeServerMessage): TFakeServerMessageWire => ({ + ...message, + _updatedAt: new Date(message._updatedAt).toISOString() +}); + +const serializeDeleted = (message: IFakeServerDeletedMessage): TFakeServerDeletedWire => ({ + _id: message._id, + _deletedAt: new Date(message._deletedAt).toISOString(), + ...(message._updatedAt !== undefined && { _updatedAt: new Date(message._updatedAt).toISOString() }) +}); + +export const createFakeSyncServer = (options: IFakeSyncServerOptions = {}): IFakeSyncServer => { + const updatedPageSize = options.updatedPageSize ?? Number.POSITIVE_INFINITY; + const deletedPageSize = options.deletedPageSize ?? Number.POSITIVE_INFINITY; + + let updatedPageRequests = 0; + let deletedPageRequests = 0; + + const paginate = (source: T[], next: number | null | undefined, pageSize: number, sortKey: (message: T) => number) => { + const cursor = next ?? 0; + const matching = source.filter(message => sortKey(message) > cursor).sort((a, b) => sortKey(a) - sortKey(b)); + const page = matching.slice(0, pageSize); + const nextCursor = matching.length > page.length ? sortKey(page[page.length - 1]) : null; + return { page, nextCursor }; + }; + + const server: IFakeSyncServer = { + updated: [], + deleted: [], + failUpdatedPageAtRequest: null, + failDeletedPageAtRequest: null, + + reset() { + server.updated.length = 0; + server.deleted.length = 0; + server.failUpdatedPageAtRequest = null; + server.failDeletedPageAtRequest = null; + updatedPageRequests = 0; + deletedPageRequests = 0; + }, + + handleSyncMessages(params) { + if (params.type === 'DELETED') { + deletedPageRequests += 1; + if (server.failDeletedPageAtRequest && deletedPageRequests === server.failDeletedPageAtRequest) { + throw new Error('DELETED page request failed'); + } + const { page, nextCursor } = paginate(server.deleted, params.next, deletedPageSize, message => message._deletedAt); + return { result: { deleted: page.map(serializeDeleted), cursor: { next: nextCursor, previous: null } } }; + } + + updatedPageRequests += 1; + if (server.failUpdatedPageAtRequest && updatedPageRequests === server.failUpdatedPageAtRequest) { + throw new Error('UPDATED page request failed'); + } + const { page, nextCursor } = paginate(server.updated, params.next, updatedPageSize, message => message._updatedAt); + return { result: { updated: page.map(serialize), cursor: { next: nextCursor, previous: null } } }; + }, + + installOn(sdkGet) { + sdkGet.mockImplementation((endpoint: string, params: { type?: string; next?: number | null }) => { + if (endpoint === 'chat.syncMessages') { + return Promise.resolve(server.handleSyncMessages(params)); + } + throw new Error(`Unexpected endpoint ${endpoint}`); + }); + } + }; + + return server; +}; diff --git a/app/lib/database/__tests__/loadMessagesForRoom.integration.test.ts b/app/lib/database/__tests__/loadMessagesForRoom.integration.test.ts new file mode 100644 index 00000000000..1c4a9bb7536 --- /dev/null +++ b/app/lib/database/__tests__/loadMessagesForRoom.integration.test.ts @@ -0,0 +1,133 @@ +import { Q } from '@nozbe/watermelondb'; + +import type { TAppDatabase } from '../interfaces'; +import type { IMessage, TMessageModel, TSubscriptionModel } from '../../../definitions'; +import { closeLokiTestDatabase, createLokiTestDatabase, resetLokiTestDatabase, seedSubscription } from './lokiTestDatabase'; +import sdk from '../../services/sdk'; +import { loadMessagesForRoom } from '../../methods/loadMessagesForRoom'; + +// Real persistence + real cursor: point `database.active` at the live LokiJS DB so +// `loadMessagesForRoom` drives the real `updateMessages`, the real `getMessageById` / +// `getSubscriptionByRoomId` services, and the real `advanceSyncCursor`. Only the outer +// edges are mocked: `sdk` (the *.history network), `auxStore` (settings + dispatch), +// `encryption` (pass-through), `log`. +let mockActiveDatabase: TAppDatabase; +jest.mock('../../database', () => ({ + __esModule: true, + default: { + get active() { + return mockActiveDatabase; + } + } +})); + +jest.mock('../../store/auxStore', () => ({ + store: { + getState: () => ({ settings: { Hide_System_Messages: [] } }), + dispatch: jest.fn() + } +})); + +jest.mock('../../services/sdk', () => ({ + __esModule: true, + default: { get: jest.fn() } +})); + +jest.mock('../../encryption', () => ({ + Encryption: { decryptMessages: (messages: unknown) => Promise.resolve(messages) } +})); + +jest.mock('../../methods/helpers/log', () => ({ __esModule: true, default: jest.fn() })); + +const mockedSdkGet = sdk.get as jest.MockedFunction; + +const RID = 'room-1'; +const T0 = Date.UTC(2026, 6, 22, 12, 0, 0); +const T1 = T0 + 1000; +const T2 = T0 + 2000; +const T3 = T0 + 3000; +const T5 = T0 + 5000; + +// A *.history message. `_updatedAt` must be set explicitly: normalizeMessage stamps a +// missing `_updatedAt` with `new Date()`, which would make advanceSyncCursor jump to now. +const historyMessage = (id: string, ts: number, updatedAt: number = ts): IMessage => + ({ + _id: id, + rid: RID, + msg: `msg-${id}`, + ts: new Date(ts).toISOString(), + _updatedAt: new Date(updatedAt).toISOString(), + u: { _id: 'user-1', username: 'user-1' } + } as unknown as IMessage); + +// A partial (< COUNT) batch: single fetch, no recursion, no trailing load-more. +const answerHistoryOnce = (messages: IMessage[]) => mockedSdkGet.mockResolvedValueOnce({ success: true, messages } as never); + +const persistedMessageIds = async (): Promise => { + const rows = (await mockActiveDatabase.get('messages').query(Q.where('rid', RID)).fetch()) as TMessageModel[]; + return rows.map(row => row.id).sort(); +}; + +const persistedLastOpen = async (rid = RID): Promise => { + const subscription = (await mockActiveDatabase.get('subscriptions').find(rid)) as TSubscriptionModel; + return subscription.lastOpen?.getTime(); +}; + +describe('loadMessagesForRoom cursor gate (LokiJS integration)', () => { + beforeAll(() => { + mockActiveDatabase = createLokiTestDatabase(); + }); + + afterAll(() => closeLokiTestDatabase(mockActiveDatabase)); + + beforeEach(async () => { + await resetLokiTestDatabase(mockActiveDatabase); + mockedSdkGet.mockReset(); + }); + + it('advances the cursor to the window max on the newest load (no `latest`)', async () => { + await seedSubscription(mockActiveDatabase, { rid: RID, lastOpen: new Date(T0) }); + answerHistoryOnce([historyMessage('m1', T1), historyMessage('m3', T3), historyMessage('m2', T2)]); + + await loadMessagesForRoom({ rid: RID, t: 'c' }); + + expect(await persistedMessageIds()).toEqual(['m1', 'm2', 'm3']); + expect(await persistedLastOpen()).toBe(T3); + }); + + it('does not advance the cursor for a scroll-up window (`latest` set)', async () => { + await seedSubscription(mockActiveDatabase, { rid: RID, lastOpen: new Date(T3) }); + // older history paged in above the cursor — an ordinary scroll-up + answerHistoryOnce([historyMessage('h0', T0), historyMessage('h1', T1), historyMessage('h2', T2)]); + + await loadMessagesForRoom({ rid: RID, t: 'c', latest: new Date(T3) }); + + expect(await persistedLastOpen()).toBe(T3); + }); + + it('does not jump the cursor past unsynced newer messages when scroll-up surfaces an edited old message', async () => { + // The exact case the gate exists for: cursor at T2; newer messages are still + // unsynced and NOT in this window. Scrolling up loads an OLD message whose recent + // edit stamped `_updatedAt = T5`. Advancing the cursor to T5 would skip those + // unsynced messages on the next sync — the gate must keep it pinned at T2. + await seedSubscription(mockActiveDatabase, { rid: RID, lastOpen: new Date(T2) }); + const editedOldMessage = historyMessage('edited-old', T0, T5); + answerHistoryOnce([historyMessage('older', T0 - 1000), editedOldMessage]); + + await loadMessagesForRoom({ rid: RID, t: 'c', latest: new Date(T2) }); + + // pinned at T2 — not jumped to the edited message's T5 + expect(await persistedLastOpen()).toBe(T2); + }); + + it('persists the window records regardless of the gate (scroll-up path)', async () => { + await seedSubscription(mockActiveDatabase, { rid: RID, lastOpen: new Date(T3) }); + answerHistoryOnce([historyMessage('h0', T0), historyMessage('h1', T1), historyMessage('h2', T2)]); + + await loadMessagesForRoom({ rid: RID, t: 'c', latest: new Date(T3) }); + + // records land even though the cursor never advanced + expect(await persistedMessageIds()).toEqual(['h0', 'h1', 'h2']); + expect(await persistedLastOpen()).toBe(T3); + }); +}); diff --git a/app/lib/database/__tests__/loadMissedMessages.integration.test.ts b/app/lib/database/__tests__/loadMissedMessages.integration.test.ts new file mode 100644 index 00000000000..e3167d6d0fb --- /dev/null +++ b/app/lib/database/__tests__/loadMissedMessages.integration.test.ts @@ -0,0 +1,288 @@ +import { Q } from '@nozbe/watermelondb'; + +import type { TAppDatabase } from '../interfaces'; +import type { TMessageModel, TSubscriptionModel } from '../../../definitions'; +import { + closeLokiTestDatabase, + createLokiTestDatabase, + resetLokiTestDatabase, + seedMessage, + seedSubscription +} from './lokiTestDatabase'; +import { createFakeSyncServer, type IFakeServerMessage } from './fakeSyncServer'; +import sdk from '../../services/sdk'; +import { loadMissedMessages } from '../../methods/loadMissedMessages'; + +// Real persistence + real cursor: point `database.active` at the live LokiJS DB so +// `loadMissedMessages` drives the real `updateMessages` (create/remove through the +// WMDB writer lock) and the real `advanceSyncCursor`. Only the outer edges are mocked: +// `sdk` (the network), `auxStore` (server version), `encryption` (pass-through), `log`. +let mockActiveDatabase: TAppDatabase; +jest.mock('../../database', () => ({ + __esModule: true, + default: { + get active() { + return mockActiveDatabase; + } + } +})); + +let mockServerVersion = '8.5.1'; +jest.mock('../../store/auxStore', () => ({ + store: { + getState: () => ({ server: { version: mockServerVersion } }) + } +})); + +jest.mock('../../services/sdk', () => ({ + __esModule: true, + default: { get: jest.fn() } +})); + +jest.mock('../../encryption', () => ({ + Encryption: { decryptMessages: (messages: unknown) => Promise.resolve(messages) } +})); + +jest.mock('../../methods/helpers/log', () => ({ __esModule: true, default: jest.fn() })); + +const mockedSdkGet = sdk.get as jest.MockedFunction; + +const RID = 'room-1'; +const T0 = Date.UTC(2026, 6, 22, 12, 0, 0); +const T1 = T0 + 1000; +const T2 = T0 + 2000; +const T3 = T0 + 3000; +const T5 = T0 + 5000; + +const server = createFakeSyncServer(); + +const serverMessage = (id: string, updatedAt: number): IFakeServerMessage => ({ + _id: id, + rid: RID, + msg: `msg-${id}`, + ts: new Date(updatedAt).toISOString(), + _updatedAt: updatedAt +}); + +const persistedMessageIds = async (): Promise => { + const rows = (await mockActiveDatabase.get('messages').query(Q.where('rid', RID)).fetch()) as TMessageModel[]; + return rows.map(row => row.id).sort(); +}; + +const persistedLastOpen = async (rid = RID): Promise => { + const subscription = (await mockActiveDatabase.get('subscriptions').find(rid)) as TSubscriptionModel; + return subscription.lastOpen?.getTime(); +}; + +describe('loadMissedMessages (LokiJS integration)', () => { + beforeAll(() => { + mockActiveDatabase = createLokiTestDatabase(); + }); + + afterAll(() => closeLokiTestDatabase(mockActiveDatabase)); + + beforeEach(async () => { + await resetLokiTestDatabase(mockActiveDatabase); + server.reset(); + mockServerVersion = '8.5.1'; + mockedSdkGet.mockReset(); + server.installOn(mockedSdkGet as unknown as Parameters[0]); + }); + + it('initial fetch seeds both cursors from lastOpen: persists new updates and applies deletions', async () => { + await seedSubscription(mockActiveDatabase, { rid: RID, lastOpen: new Date(T0) }); + // a message deleted after the cursor must pre-exist to be removed + await seedMessage(mockActiveDatabase, { id: 'gone', rid: RID, updatedAt: new Date(T2) }); + server.updated.push(serverMessage('new', T1), serverMessage('stale', T0 - 1000)); + server.deleted.push({ _id: 'gone', _deletedAt: T2 }); + + await loadMissedMessages({ rid: RID, lastOpen: new Date(T0) }); + + // the new update (>T0) is persisted; the stale one (<=T0) never arrives; the deletion is applied + expect(await persistedMessageIds()).toEqual(['new']); + // both streams were queried off the initial lastOpen cursor + const cursors = mockedSdkGet.mock.calls.map(([, params]) => params as unknown as Record); + expect(cursors).toContainEqual({ roomId: RID, next: T0, count: 50, type: 'UPDATED' }); + expect(cursors).toContainEqual({ roomId: RID, next: T0, count: 50, type: 'DELETED' }); + }); + + it('follows the UPDATED cursor across pages (awaited recursion) and advances lastOpen once to the batch max', async () => { + await seedSubscription(mockActiveDatabase, { rid: RID, lastOpen: new Date(T0) }); + server.updated.push(serverMessage('u1', T1), serverMessage('u2', T2), serverMessage('u3', T3)); + const paged = createFakeSyncServer({ updatedPageSize: 1 }); + paged.updated.push(...server.updated); + paged.installOn(mockedSdkGet as unknown as Parameters[0]); + + await loadMissedMessages({ rid: RID, lastOpen: new Date(T0) }); + + // every page landed via the awaited recursion + expect(await persistedMessageIds()).toEqual(['u1', 'u2', 'u3']); + // the cursor advanced to the max _updatedAt of the whole chain + expect(await persistedLastOpen()).toBe(T3); + }); + + it('applies a paginated DELETED-only sync as removals, never as upserts (no positional-slot bug)', async () => { + await seedSubscription(mockActiveDatabase, { rid: RID, lastOpen: new Date(T0) }); + await seedMessage(mockActiveDatabase, { id: 'd1', rid: RID, updatedAt: new Date(T1) }); + await seedMessage(mockActiveDatabase, { id: 'd2', rid: RID, updatedAt: new Date(T2) }); + await seedMessage(mockActiveDatabase, { id: 'd3', rid: RID, updatedAt: new Date(T3) }); + const paged = createFakeSyncServer({ deletedPageSize: 2 }); + paged.deleted.push({ _id: 'd1', _deletedAt: T1 }, { _id: 'd2', _deletedAt: T2 }, { _id: 'd3', _deletedAt: T3 }); + paged.installOn(mockedSdkGet as unknown as Parameters[0]); + + await loadMissedMessages({ rid: RID, lastOpen: new Date(T0) }); + + // all three removed; none resurrected into the updated slot + expect(await persistedMessageIds()).toEqual([]); + // deletions carry no update timestamp, so the cursor stays put + expect(await persistedLastOpen()).toBe(T0); + }); + + it('does not advance the cursor when a later page fails, even after earlier pages persisted', async () => { + await seedSubscription(mockActiveDatabase, { rid: RID, lastOpen: new Date(T0) }); + await seedMessage(mockActiveDatabase, { id: 'd1', rid: RID, updatedAt: new Date(T1) }); + await seedMessage(mockActiveDatabase, { id: 'd2', rid: RID, updatedAt: new Date(T2) }); + await seedMessage(mockActiveDatabase, { id: 'd3', rid: RID, updatedAt: new Date(T3) }); + const paged = createFakeSyncServer({ deletedPageSize: 2 }); + paged.deleted.push({ _id: 'd1', _deletedAt: T1 }, { _id: 'd2', _deletedAt: T2 }, { _id: 'd3', _deletedAt: T3 }); + paged.failDeletedPageAtRequest = 2; + paged.installOn(mockedSdkGet as unknown as Parameters[0]); + + await expect(loadMissedMessages({ rid: RID, lastOpen: new Date(T0) })).rejects.toThrow(); + + // page 1 persisted (d1, d2 removed), but the failed chain left the cursor untouched for retry + expect(await persistedMessageIds()).toEqual(['d3']); + expect(await persistedLastOpen()).toBe(T0); + }); + + it('stops following the UPDATED cursor at MAX_PAGES instead of walking the chain unbounded', async () => { + await seedSubscription(mockActiveDatabase, { rid: RID, lastOpen: new Date(T0) }); + const paged = createFakeSyncServer({ updatedPageSize: 1 }); + for (let index = 1; index <= 15; index += 1) { + paged.updated.push(serverMessage(`u${index}`, T0 + index * 1000)); + } + paged.installOn(mockedSdkGet as unknown as Parameters[0]); + + await loadMissedMessages({ rid: RID, lastOpen: new Date(T0) }); + + const updatedRequests = mockedSdkGet.mock.calls.filter( + ([, params]) => (params as unknown as { type?: string }).type === 'UPDATED' + ); + expect(updatedRequests).toHaveLength(10); + // only the ten fetched pages landed, and the cursor sits on the last one so the next + // sync resumes exactly where the bound stopped + expect(await persistedMessageIds()).toEqual(['u1', 'u10', 'u2', 'u3', 'u4', 'u5', 'u6', 'u7', 'u8', 'u9']); + expect(await persistedLastOpen()).toBe(T0 + 10 * 1000); + }); + + it('resumes the DELETED walk where the bound stopped it, with no updates to carry the cursor', async () => { + await seedSubscription(mockActiveDatabase, { rid: RID, lastOpen: new Date(T0) }); + const paged = createFakeSyncServer({ deletedPageSize: 1 }); + for (let index = 1; index <= 15; index += 1) { + paged.deleted.push({ _id: `d${index}`, _deletedAt: T0 + index * 1000 }); + await seedMessage(mockActiveDatabase, { id: `d${index}`, rid: RID, updatedAt: new Date(T0 + index * 1000) }); + } + paged.installOn(mockedSdkGet as unknown as Parameters[0]); + + await loadMissedMessages({ rid: RID, lastOpen: new Date(T0) }); + + // with no DELETED stop-point the cursor would stay at T0 and every later sync would + // re-walk these same ten pages forever + expect(await persistedLastOpen()).toBe(T0 + 10 * 1000); + + mockedSdkGet.mockClear(); + await loadMissedMessages({ rid: RID }); + + const deletedCursors = mockedSdkGet.mock.calls + .map(([, params]) => params as unknown as { type?: string; next?: number }) + .filter(params => params.type === 'DELETED') + .map(params => params.next); + expect(deletedCursors[0]).toBe(T0 + 10 * 1000); + expect(await persistedMessageIds()).toEqual([]); + }); + + it('never advances past an abandoned DELETED stop-point, even when the updates ran further', async () => { + await seedSubscription(mockActiveDatabase, { rid: RID, lastOpen: new Date(T0) }); + const paged = createFakeSyncServer({ deletedPageSize: 1 }); + for (let index = 1; index <= 15; index += 1) { + paged.deleted.push({ _id: `d${index}`, _deletedAt: T0 + index * 1000 }); + await seedMessage(mockActiveDatabase, { id: `d${index}`, rid: RID, updatedAt: new Date(T0 + index * 1000) }); + } + // an update stamped far past every deletion the bound will reach + paged.updated.push(serverMessage('u-late', T0 + 60 * 1000)); + paged.installOn(mockedSdkGet as unknown as Parameters[0]); + + await loadMissedMessages({ rid: RID, lastOpen: new Date(T0) }); + + // advancing to the updated max would strand deletions d11..d15 forever + expect(await persistedLastOpen()).toBe(T0 + 10 * 1000); + expect(await persistedMessageIds()).toEqual(['d11', 'd12', 'd13', 'd14', 'd15', 'u-late']); + }); + + it('never advances the cursor from a deleted record, only from updated ones', async () => { + await seedSubscription(mockActiveDatabase, { rid: RID, lastOpen: new Date(T0) }); + await seedMessage(mockActiveDatabase, { id: 'gone', rid: RID, updatedAt: new Date(T1) }); + server.updated.push(serverMessage('u1', T1)); + // a deletion stamped past every update must not pull the cursor past unfetched messages + server.deleted.push({ _id: 'gone', _deletedAt: T5, _updatedAt: T5 }); + + await loadMissedMessages({ rid: RID, lastOpen: new Date(T0) }); + + expect(await persistedMessageIds()).toEqual(['u1']); + expect(await persistedLastOpen()).toBe(T1); + }); + + it('leaves the cursor unchanged when the fetch is empty (server answers a current/future cursor)', async () => { + await seedSubscription(mockActiveDatabase, { rid: RID, lastOpen: new Date(T5) }); + server.updated.push(serverMessage('old', T2)); + + await loadMissedMessages({ rid: RID, lastOpen: new Date(T5) }); + + expect(await persistedMessageIds()).toEqual([]); + expect(await persistedLastOpen()).toBe(T5); + }); + + it('uses the lastOpen argument over the subscription record when both are present', async () => { + await seedSubscription(mockActiveDatabase, { rid: RID, lastOpen: new Date(T0) }); + server.updated.push(serverMessage('between', T1), serverMessage('after', T3)); + + // arg cursor is T2; the record cursor is the older T0 + await loadMissedMessages({ rid: RID, lastOpen: new Date(T2) }); + + // only messages newer than the argument cursor arrive + expect(await persistedMessageIds()).toEqual(['after']); + }); + + it('falls back to the subscription lastOpen when no argument is given', async () => { + await seedSubscription(mockActiveDatabase, { rid: RID, lastOpen: new Date(T2) }); + server.updated.push(serverMessage('between', T1), serverMessage('after', T3)); + + await loadMissedMessages({ rid: RID }); + + expect(await persistedMessageIds()).toEqual(['after']); + }); + + it('issues no request and persists nothing when there is no subscription and no argument cursor', async () => { + server.updated.push(serverMessage('any', T3)); + + await loadMissedMessages({ rid: RID }); + + expect(mockedSdkGet).not.toHaveBeenCalled(); + expect(await persistedMessageIds()).toEqual([]); + }); + + it('legacy <7.1 servers still issue the sync call with an ISO lastUpdate', async () => { + mockServerVersion = '7.0.0'; + await seedSubscription(mockActiveDatabase, { rid: RID, lastOpen: new Date(T0) }); + // the <7.1 endpoint answers a single `{ updated, deleted }` payload, not the cursor API + // the fake server models; stub it directly here. + mockedSdkGet.mockResolvedValue({ result: { updated: [], deleted: [] } } as never); + + await loadMissedMessages({ rid: RID, lastOpen: new Date(T0) }); + + expect(mockedSdkGet).toHaveBeenCalledWith('chat.syncMessages', { + roomId: RID, + lastUpdate: new Date(T0).toISOString() + }); + }); +}); diff --git a/app/lib/database/__tests__/lokiTestDatabase.test.ts b/app/lib/database/__tests__/lokiTestDatabase.test.ts new file mode 100644 index 00000000000..c0be52a1047 --- /dev/null +++ b/app/lib/database/__tests__/lokiTestDatabase.test.ts @@ -0,0 +1,70 @@ +import type { TAppDatabase } from '../interfaces'; +import { + closeLokiTestDatabase, + createLokiTestDatabase, + resetLokiTestDatabase, + seedSubscription, + seedMessage +} from './lokiTestDatabase'; +import { createFakeSyncServer } from './fakeSyncServer'; + +describe('lokiTestDatabase harness', () => { + let database: TAppDatabase; + + beforeEach(async () => { + if (!database) { + database = createLokiTestDatabase(); + } + await resetLokiTestDatabase(database); + }); + + afterAll(() => closeLokiTestDatabase(database)); + + it('round-trips a subscription and a message through a real db.write + query', async () => { + const T0 = Date.UTC(2026, 6, 22, 12, 0, 0); + await seedSubscription(database, { rid: 'room-1', name: 'general', lastOpen: new Date(T0) }); + await seedMessage(database, { + id: 'message-1', + rid: 'room-1', + msg: 'hello world', + ts: new Date(T0 + 1000), + u: { _id: 'user-1', username: 'rocket.cat' } + }); + + const subscriptions = await database.get('subscriptions').query().fetch(); + expect(subscriptions).toHaveLength(1); + expect(subscriptions[0].rid).toBe('room-1'); + expect(subscriptions[0].name).toBe('general'); + expect(subscriptions[0].lastOpen?.getTime()).toBe(T0); + + const messages = await database.get('messages').query().fetch(); + expect(messages).toHaveLength(1); + expect(messages[0].id).toBe('message-1'); + expect(messages[0].msg).toBe('hello world'); + expect(messages[0].subscription?.id).toBe('room-1'); + expect(messages[0].u).toEqual({ _id: 'user-1', username: 'rocket.cat' }); + }); + + it('resets between tests: previous records are gone', async () => { + const count = await database.get('messages').query().fetchCount(); + expect(count).toBe(0); + }); + + it('serves paginated chat.syncMessages from the fake server', () => { + const server = createFakeSyncServer({ deletedPageSize: 2 }); + const T0 = Date.UTC(2026, 6, 22, 12, 0, 0); + server.deleted.push( + { _id: 'deleted-1', _deletedAt: T0 + 10_000 }, + { _id: 'deleted-2', _deletedAt: T0 + 20_000 }, + { _id: 'deleted-3', _deletedAt: T0 + 30_000 } + ); + + const firstPage = server.handleSyncMessages({ type: 'DELETED', next: T0 }); + expect(firstPage.result.deleted?.map(message => message._id)).toEqual(['deleted-1', 'deleted-2']); + expect(firstPage.result.cursor.next).toBe(T0 + 20_000); + + const secondPage = server.handleSyncMessages({ type: 'DELETED', next: firstPage.result.cursor.next }); + expect(secondPage.result.deleted?.map(message => message._id)).toEqual(['deleted-3']); + expect(secondPage.result.cursor.next).toBeNull(); + }); +}); diff --git a/app/lib/database/__tests__/lokiTestDatabase.ts b/app/lib/database/__tests__/lokiTestDatabase.ts new file mode 100644 index 00000000000..0c7f52199a9 --- /dev/null +++ b/app/lib/database/__tests__/lokiTestDatabase.ts @@ -0,0 +1,198 @@ +import { Database } from '@nozbe/watermelondb'; +import LokiJSAdapter from '@nozbe/watermelondb/adapters/lokijs'; + +import type { TSubscriptionModel, TMessageModel, SubscriptionType, MessageType } from '../../../definitions'; +import appSchema from '../schema/app'; +import migrations from '../model/migrations'; +import Subscription from '../model/Subscription'; +import Room from '../model/Room'; +import Message from '../model/Message'; +import Thread from '../model/Thread'; +import ThreadMessage from '../model/ThreadMessage'; +import CustomEmoji from '../model/CustomEmoji'; +import FrequentlyUsedEmoji from '../model/FrequentlyUsedEmoji'; +import Upload from '../model/Upload'; +import Setting from '../model/Setting'; +import Role from '../model/Role'; +import Permission from '../model/Permission'; +import SlashCommand from '../model/SlashCommand'; +import User from '../model/User'; +import type { TAppDatabase } from '../interfaces'; + +/** + * Builds a real WatermelonDB `Database` backed by an in-memory LokiJS adapter, + * over the production `appSchema`. Everything WatermelonDB (the JS core, the + * writer lock, `db.write` serialization, queries) is real — only persistence is + * swapped from native SQLite to LokiJS, so integration tests can exercise the + * sync/subscription code without a native bridge. + * + * `useWebWorker: false` keeps the adapter in-process (no worker file to resolve + * under Jest); `useIncrementalIndexedDB: false` keeps it in-memory (no IndexedDB + * shims needed in the node test environment). + */ +export const createLokiTestDatabase = (): TAppDatabase => { + const adapter = new LokiJSAdapter({ + schema: appSchema, + migrations, + useWebWorker: false, + useIncrementalIndexedDB: false + }); + + return new Database({ + adapter, + modelClasses: [ + Subscription, + Room, + Message, + Thread, + ThreadMessage, + CustomEmoji, + FrequentlyUsedEmoji, + Upload, + Setting, + Role, + Permission, + SlashCommand, + User + ] + }) as TAppDatabase; +}; + +/** Truncates every collection so a single database instance can be reused across tests. */ +export const resetLokiTestDatabase = async (database: Database): Promise => { + await database.write(async () => { + await database.unsafeResetDatabase(); + }); +}; + +/** + * Closes the underlying in-memory LokiJS driver so Jest's event loop drains and + * the process exits on its own — without `--forceExit`. The adapter exposes no + * public close, so we reach the in-process driver the same way the adapter's own + * `testClone` does (`_dispatcher._worker._bridge.driver`). Call from `afterAll`. + */ +export const closeLokiTestDatabase = (database: TAppDatabase): Promise => { + const compat = database.adapter as unknown as { underlyingAdapter?: unknown }; + const adapter = (compat.underlyingAdapter ?? database.adapter) as { + _dispatcher?: { _worker?: { _bridge?: { driver?: { loki?: { close?: (callback: () => void) => void } } } } }; + }; + const loki = adapter._dispatcher?._worker?._bridge?.driver?.loki; + if (!loki?.close) { + return Promise.resolve(); + } + return new Promise(resolve => loki.close!(() => resolve())); +}; + +/** + * Runs `block` while capturing and then clearing WatermelonDB's dev-only + * "writer queued behind a running writer" diagnostic timer. + * + * In non-production builds `WorkQueue.enqueue` schedules a ~1.5s `setTimeout` + * whenever a write is enqueued behind an in-flight one, to warn about a possibly + * stuck queue. It's a harmless no-op once the queued write runs, but as a live + * timer it outlives Jest's 1s post-run window and prints the benign + * "Jest did not exit one second after the test run has completed" warning. + * Tests that deliberately contend the writer lock wrap that section here so the + * event loop drains without `--forceExit`. + */ +export const withWriterQueueDiagnosticCleared = async (block: () => Promise): Promise => { + const scheduled: ReturnType[] = []; + const realSetTimeout = global.setTimeout; + global.setTimeout = ((...timeoutArgs: Parameters) => { + const timer = realSetTimeout(...timeoutArgs); + const delay = timeoutArgs[1]; + if (typeof delay === 'number' && delay >= 1000) { + scheduled.push(timer); + } + return timer; + }) as typeof setTimeout; + try { + return await block(); + } finally { + global.setTimeout = realSetTimeout; + scheduled.forEach(clearTimeout); + } +}; + +export interface ISeedSubscription { + id?: string; + rid?: string; + name?: string; + fname?: string; + t?: string; + lastOpen?: Date; + ls?: Date; + ts?: Date; + encrypted?: boolean; + E2EKey?: string; +} + +/** Creates a `subscriptions` record with sensible defaults; override any field. */ +export const seedSubscription = (database: TAppDatabase, overrides: ISeedSubscription = {}): Promise => { + const rid = overrides.rid ?? overrides.id ?? 'room-1'; + return database.write(() => + database.get('subscriptions').create(record => { + record._raw.id = overrides.id ?? rid; + record._id = overrides.id ?? rid; + record.rid = rid; + record.name = overrides.name ?? 'test-room'; + record.fname = overrides.fname ?? overrides.name ?? 'test-room'; + record.t = (overrides.t ?? 'c') as SubscriptionType; + record.open = true; + if (overrides.lastOpen) { + record.lastOpen = overrides.lastOpen; + } + if (overrides.ls) { + record.ls = overrides.ls; + } + if (overrides.ts) { + record.ts = overrides.ts; + } + if (overrides.encrypted !== undefined) { + record.encrypted = overrides.encrypted; + } + if (overrides.E2EKey !== undefined) { + record.E2EKey = overrides.E2EKey; + } + }) + ); +}; + +export interface ISeedMessage { + id?: string; + rid?: string; + msg?: string; + ts?: Date; + updatedAt?: Date; + u?: { _id: string; username: string }; + tmid?: string; + t?: string; + status?: number; +} + +/** Creates a `messages` record with sensible defaults; override any field. */ +export const seedMessage = (database: TAppDatabase, overrides: ISeedMessage = {}): Promise => { + const rid = overrides.rid ?? 'room-1'; + const ts = overrides.ts ?? new Date(); + return database.write(() => + database.get('messages').create(record => { + if (overrides.id) { + record._raw.id = overrides.id; + } + record.subscription!.id = rid; + record.msg = overrides.msg ?? 'hello'; + record.ts = ts; + record._updatedAt = overrides.updatedAt ?? ts; + record.u = overrides.u ?? { _id: 'user-1', username: 'user-1' }; + if (overrides.tmid) { + record.tmid = overrides.tmid; + } + if (overrides.t) { + record.t = overrides.t as MessageType; + } + if (overrides.status !== undefined) { + record.status = overrides.status; + } + }) + ); +}; diff --git a/app/lib/database/__tests__/nullCursorRecurrence.integration.test.ts b/app/lib/database/__tests__/nullCursorRecurrence.integration.test.ts new file mode 100644 index 00000000000..fcbb453bceb --- /dev/null +++ b/app/lib/database/__tests__/nullCursorRecurrence.integration.test.ts @@ -0,0 +1,196 @@ +import { Q } from '@nozbe/watermelondb'; + +import type { TAppDatabase } from '../interfaces'; +import type { TMessageModel, TSubscriptionModel } from '../../../definitions'; +import { + closeLokiTestDatabase, + createLokiTestDatabase, + resetLokiTestDatabase, + seedMessage, + seedSubscription +} from './lokiTestDatabase'; +import { createFakeSyncServer, type IFakeServerMessage } from './fakeSyncServer'; +import { MessageTypeLoad } from '../../constants/messageTypeLoad'; +import { messagesStatus } from '../../constants/messagesStatus'; +import sdk from '../../services/sdk'; +import { loadMissedMessages } from '../../methods/loadMissedMessages'; + +// Recurrence of #7499 for rooms whose subscription row has NO sync cursor: a room reached by +// tapping a push notification is persisted without `lastOpen`, so every case here deliberately +// leaves it unseeded — the gap that let the defect survive `reconnectCycle.integration.test.ts`. +// Same real-WMDB setup as that suite: `database.active` is the live LokiJS DB, only the outer +// edges (`sdk`, `auxStore`, `encryption`, `log`) are mocked. +let mockActiveDatabase: TAppDatabase; +jest.mock('../../database', () => ({ + __esModule: true, + default: { + get active() { + return mockActiveDatabase; + } + } +})); + +jest.mock('../../store/auxStore', () => ({ + store: { + getState: () => ({ + server: { version: '8.5.1' }, + encryption: { enabled: true }, + settings: { Hide_System_Messages: [] } + }), + dispatch: jest.fn() + } +})); + +jest.mock('../../services/sdk', () => ({ + __esModule: true, + default: { get: jest.fn(), post: jest.fn() } +})); + +jest.mock('../../encryption', () => ({ + Encryption: { decryptMessages: (messages: unknown) => Promise.resolve(messages) } +})); + +jest.mock('../../encryption/utils', () => ({ hasE2EEWarning: () => false })); + +jest.mock('../../methods/helpers/log', () => ({ __esModule: true, default: jest.fn() })); + +const mockedSdkGet = sdk.get as jest.MockedFunction; + +const RID = 'room-1'; +const JOINED = Date.UTC(2026, 6, 22, 10, 0, 0); // subscription ts: the user joined the room +const LOCAL_TIP = Date.UTC(2026, 6, 22, 11, 0, 0); // newest message this device actually persisted +const GAP_TS = Date.UTC(2026, 6, 22, 11, 30, 0); // a message that only ever reached the server +const READ = Date.UTC(2026, 6, 22, 12, 0, 0); // subscription ls: last read, still no lastOpen +const MSG_TS = READ + 60_000; // a message lands on the server 1 min after the last read +const DEVICE_CLOCK = Date.UTC(2026, 6, 22, 13, 0, 0); // a device clock running an hour ahead + +const server = createFakeSyncServer(); + +const serverMessage = (id: string, updatedAt: number): IFakeServerMessage => ({ + _id: id, + rid: RID, + msg: `msg-${id}`, + ts: new Date(updatedAt).toISOString(), + _updatedAt: updatedAt +}); + +const persistedMessageIds = async (): Promise => { + const rows = (await mockActiveDatabase.get('messages').query(Q.where('rid', RID)).fetch()) as TMessageModel[]; + return rows.map(row => row.id).sort(); +}; + +const persistedLastOpen = async (): Promise => { + const subscription = (await mockActiveDatabase.get('subscriptions').find(RID)) as TSubscriptionModel; + return subscription.lastOpen?.getTime(); +}; + +describe('null sync cursor recurrence (#7499, LokiJS integration)', () => { + beforeAll(() => { + mockActiveDatabase = createLokiTestDatabase(); + }); + + afterAll(() => closeLokiTestDatabase(mockActiveDatabase)); + + beforeEach(async () => { + await resetLokiTestDatabase(mockActiveDatabase); + server.reset(); + mockedSdkGet.mockReset(); + server.installOn(mockedSdkGet as unknown as Parameters[0]); + }); + + it('notification-tap room (no lastOpen): syncs off `ls` instead of fetching nothing', async () => { + await seedSubscription(mockActiveDatabase, { rid: RID, ts: new Date(JOINED), ls: new Date(READ) }); + server.updated.push(serverMessage('missed-1', MSG_TS)); + + await loadMissedMessages({ rid: RID }); + + expect(mockedSdkGet).toHaveBeenCalledWith('chat.syncMessages', expect.objectContaining({ roomId: RID, next: READ })); + expect(await persistedMessageIds()).toEqual(['missed-1']); + // the recovered sync seeds the cursor from the server timestamp it just read + expect(await persistedLastOpen()).toBe(MSG_TS); + }); + + // `ls` is the server-side last-read stamp and ANY device writes it: reading the room on desktop + // pushes it past messages this device never received. The fallback has to express the newest + // point THIS device is in sync with, so what it has persisted outranks what someone else read. + it('`ls` ahead of the device (read elsewhere): syncs off the newest persisted message, not `ls`', async () => { + await seedSubscription(mockActiveDatabase, { rid: RID, ts: new Date(JOINED), ls: new Date(READ) }); + await seedMessage(mockActiveDatabase, { id: 'local-1', rid: RID, updatedAt: new Date(LOCAL_TIP) }); + // posted after this device went quiet, but before the desktop read moved `ls` past it + server.updated.push(serverMessage('gap-1', GAP_TS)); + + await loadMissedMessages({ rid: RID }); + + expect(mockedSdkGet).toHaveBeenCalledWith('chat.syncMessages', expect.objectContaining({ next: LOCAL_TIP })); + expect(await persistedMessageIds()).toEqual(['gap-1', 'local-1']); + expect(await persistedLastOpen()).toBe(GAP_TS); + }); + + // Not every messages row carries a SERVER stamp: the synthetic load-more sentinel and outgoing + // TEMP/ERROR sends are stamped with `new Date()`. On a device whose clock runs ahead, taking + // those as the cursor asks the server for changes newer than the device clock — the very + // clock-poisoning this cursor exists to prevent. + it.each([ + ['load-more sentinel', { t: MessageTypeLoad.MORE }], + ['failed outgoing send', { status: messagesStatus.ERROR }] + ])('device-stamped row (%s) is not taken as the cursor', async (_label, poison) => { + await seedSubscription(mockActiveDatabase, { rid: RID, ts: new Date(JOINED), ls: new Date(READ) }); + await seedMessage(mockActiveDatabase, { id: 'local-1', rid: RID, updatedAt: new Date(LOCAL_TIP) }); + await seedMessage(mockActiveDatabase, { id: 'poison-1', rid: RID, updatedAt: new Date(DEVICE_CLOCK), ...poison }); + server.updated.push(serverMessage('gap-1', GAP_TS)); + + await loadMissedMessages({ rid: RID }); + + expect(mockedSdkGet).toHaveBeenCalledWith('chat.syncMessages', expect.objectContaining({ next: LOCAL_TIP })); + expect(await persistedMessageIds()).toContain('gap-1'); + }); + + it('never-read room (no lastOpen, no ls): syncs off the subscription `ts`', async () => { + await seedSubscription(mockActiveDatabase, { rid: RID, ts: new Date(JOINED) }); + server.updated.push(serverMessage('missed-1', MSG_TS)); + + await loadMissedMessages({ rid: RID }); + + expect(mockedSdkGet).toHaveBeenCalledWith('chat.syncMessages', expect.objectContaining({ roomId: RID, next: JOINED })); + expect(await persistedMessageIds()).toEqual(['missed-1']); + expect(await persistedLastOpen()).toBe(MSG_TS); + }); + + it('deleted messages are synced off the fallback cursor too', async () => { + await seedSubscription(mockActiveDatabase, { rid: RID, ts: new Date(JOINED), ls: new Date(READ) }); + server.deleted.push({ _id: 'gone-1', _deletedAt: MSG_TS }); + + await loadMissedMessages({ rid: RID }); + + expect(mockedSdkGet).toHaveBeenCalledWith('chat.syncMessages', expect.objectContaining({ type: 'DELETED', next: READ })); + // a deletion-only sync leaves the cursor where it was: the server projects deleted records + // to `{ _id, _deletedAt }`, so that stream can never stand in for an update timestamp + expect(await persistedLastOpen()).toBeUndefined(); + }); + + // `advanceSyncCursor` cannot persist a cursor for a room it has no subscription row for, so a + // sync that races ahead of the rooms sync drops it. That must cost a redundant fetch, never a + // message: once the row lands (still cursor-less), the next sync has to re-deliver. + it('subscription row missing at sync time: the message still arrives once the row lands', async () => { + server.updated.push(serverMessage('missed-1', MSG_TS)); + + await loadMissedMessages({ rid: RID }); + expect(await persistedMessageIds()).toEqual([]); + + await seedSubscription(mockActiveDatabase, { rid: RID, ts: new Date(JOINED), ls: new Date(READ) }); + await loadMissedMessages({ rid: RID }); + + expect(await persistedMessageIds()).toEqual(['missed-1']); + expect(await persistedLastOpen()).toBe(MSG_TS); + }); + + it('an explicit lastOpen argument still wins over the fallback', async () => { + await seedSubscription(mockActiveDatabase, { rid: RID, ts: new Date(JOINED), ls: new Date(READ) }); + server.updated.push(serverMessage('missed-1', MSG_TS)); + + await loadMissedMessages({ rid: RID, lastOpen: new Date(JOINED) }); + + expect(mockedSdkGet).toHaveBeenCalledWith('chat.syncMessages', expect.objectContaining({ next: JOINED })); + expect(await persistedMessageIds()).toEqual(['missed-1']); + }); +}); diff --git a/app/lib/database/__tests__/reconnectCycle.integration.test.ts b/app/lib/database/__tests__/reconnectCycle.integration.test.ts new file mode 100644 index 00000000000..d6c31b6885c --- /dev/null +++ b/app/lib/database/__tests__/reconnectCycle.integration.test.ts @@ -0,0 +1,178 @@ +import { Q } from '@nozbe/watermelondb'; + +import type { TAppDatabase } from '../interfaces'; +import type { TMessageModel, TSubscriptionModel } from '../../../definitions'; +import { closeLokiTestDatabase, createLokiTestDatabase, resetLokiTestDatabase, seedSubscription } from './lokiTestDatabase'; +import { createFakeSyncServer, type IFakeServerMessage } from './fakeSyncServer'; +import sdk from '../../services/sdk'; +import { loadMissedMessages } from '../../methods/loadMissedMessages'; +import { readMessages } from '../../methods/readMessages'; + +// The full open -> sync -> read cycle on a real WMDB core: `database.active` points at the +// live LokiJS DB so `loadMissedMessages` (real `updateMessages` + `advanceSyncCursor`) and +// `readMessages` both drive real persisted rows and the real cursor (`subscription.lastOpen`). +// Only the outer edges are mocked: `sdk` (network), `auxStore` (server version / encryption / +// settings / dispatch), `encryption` (pass-through), `encryption/utils` (no E2EE warning), `log`. +let mockActiveDatabase: TAppDatabase; +jest.mock('../../database', () => ({ + __esModule: true, + default: { + get active() { + return mockActiveDatabase; + } + } +})); + +jest.mock('../../store/auxStore', () => ({ + store: { + getState: () => ({ + server: { version: '8.5.1' }, + encryption: { enabled: true }, + settings: { Hide_System_Messages: [] } + }), + dispatch: jest.fn() + } +})); + +jest.mock('../../services/sdk', () => ({ + __esModule: true, + default: { get: jest.fn(), post: jest.fn() } +})); + +jest.mock('../../encryption', () => ({ + Encryption: { decryptMessages: (messages: unknown) => Promise.resolve(messages) } +})); + +jest.mock('../../encryption/utils', () => ({ hasE2EEWarning: () => false })); + +jest.mock('../../methods/helpers/log', () => ({ __esModule: true, default: jest.fn() })); + +const mockedSdkGet = sdk.get as jest.MockedFunction; +const mockedSdkPost = sdk.post as jest.MockedFunction; + +const RID = 'room-1'; +const T0 = Date.UTC(2026, 6, 22, 12, 0, 0); // server time the room last synced to +const MSG_TS = T0 + 60_000; // a message lands on the server 1 min later +const READ_1 = T0 + 300_000; // client reads/leaves 5 min after T0 +const READ_2 = T0 + 600_000; // a later reopen, 10 min after T0 +const SKEW = 120_000; // client clock 2 min ahead of the server + +const server = createFakeSyncServer(); + +const serverMessage = (id: string, updatedAt: number): IFakeServerMessage => ({ + _id: id, + rid: RID, + msg: `msg-${id}`, + ts: new Date(updatedAt).toISOString(), + _updatedAt: updatedAt +}); + +const persistedMessageIds = async (): Promise => { + const rows = (await mockActiveDatabase.get('messages').query(Q.where('rid', RID)).fetch()) as TMessageModel[]; + return rows.map(row => row.id).sort(); +}; + +const persistedSubscription = async (rid = RID): Promise => + (await mockActiveDatabase.get('subscriptions').find(rid)) as TSubscriptionModel; + +const persistedLastOpen = async (): Promise => (await persistedSubscription()).lastOpen?.getTime(); + +// Mirrors RoomView.init: pull whatever was missed off the server-derived cursor, then mark the +// room read at the client clock. loadMissedMessages reads the cursor from the live subscription. +const openRoom = async (clientNow: Date): Promise => { + await loadMissedMessages({ rid: RID }); + await readMessages(RID, clientNow); +}; + +describe('reconnect / offline coverage-gap cycle (LokiJS integration)', () => { + beforeAll(() => { + mockActiveDatabase = createLokiTestDatabase(); + }); + + afterAll(() => closeLokiTestDatabase(mockActiveDatabase)); + + beforeEach(async () => { + await resetLokiTestDatabase(mockActiveDatabase); + server.reset(); + mockedSdkGet.mockReset(); + server.installOn(mockedSdkGet as unknown as Parameters[0]); + mockedSdkPost.mockReset(); + mockedSdkPost.mockResolvedValue({ success: true } as never); + }); + + it('offline gap: reading/leaving never advances the cursor, and the next sync delivers the missed message', async () => { + await seedSubscription(mockActiveDatabase, { rid: RID, lastOpen: new Date(T0) }); + // a correspondent posts while the device is offline; the client never receives it over the + // stream, so it lives only on the server, newer than the cursor + server.updated.push(serverMessage('offline-1', MSG_TS)); + + // user reads/leaves the room 5 min later — plain wall clock, no skew + await readMessages(RID, new Date(READ_1)); + + // the read stamped `ls` (client clock) but must NOT touch the sync cursor, and nothing synced yet + const afterRead = await persistedSubscription(); + expect(afterRead.lastOpen?.getTime()).toBe(T0); + expect(afterRead.ls?.getTime()).toBe(READ_1); + expect(await persistedMessageIds()).toEqual([]); + + // next sync (back online) loads the message off the untouched cursor and advances to its server ts + await loadMissedMessages({ rid: RID }); + expect(await persistedMessageIds()).toEqual(['offline-1']); + expect(await persistedLastOpen()).toBe(MSG_TS); + }); + + it('clock skew (#7499): a message inside the skew window is delivered on tap and survives later reopens', async () => { + await seedSubscription(mockActiveDatabase, { rid: RID, lastOpen: new Date(T0) }); + + // user opens/reads the room, then backgrounds the app — client clock 2 min ahead of server + await openRoom(new Date(T0 + SKEW)); + // empty sync + a client-clock read must leave the cursor pinned at the server time + expect(await persistedLastOpen()).toBe(T0); + expect((await persistedSubscription()).ls?.getTime()).toBe(T0 + SKEW); + + // correspondent replies 1 min later — its server ts (MSG_TS) is BEHIND the client's read clock + server.updated.push(serverMessage('missed-1', MSG_TS)); + + // notification tap 5 min later: the message is recovered because the cursor tracks server ts, + // not the client read time (which had already passed MSG_TS) + await openRoom(new Date(READ_1 + SKEW)); + expect(await persistedMessageIds()).toEqual(['missed-1']); + expect(await persistedLastOpen()).toBe(MSG_TS); + + // a later reopen keeps it and never re-advances the cursor from the client clock + await openRoom(new Date(READ_2 + SKEW)); + expect(await persistedMessageIds()).toEqual(['missed-1']); + expect(await persistedLastOpen()).toBe(MSG_TS); + }); + + it('backgrounded delivery: a message that arrived while backgrounded is delivered on reopen', async () => { + await seedSubscription(mockActiveDatabase, { rid: RID, lastOpen: new Date(T0) }); + + // first open: nothing missed yet + await openRoom(new Date(T0)); + expect(await persistedMessageIds()).toEqual([]); + expect(await persistedLastOpen()).toBe(T0); + + // message arrives on the server while the app is backgrounded + server.updated.push(serverMessage('missed-1', MSG_TS)); + + // reopen delivers it and advances the cursor to its server ts + await openRoom(new Date(READ_1)); + expect(await persistedMessageIds()).toEqual(['missed-1']); + expect(await persistedLastOpen()).toBe(MSG_TS); + }); + + it('readMessages marks the room read (ls, unread) without ever moving the sync cursor', async () => { + await seedSubscription(mockActiveDatabase, { rid: RID, lastOpen: new Date(T0) }); + + await readMessages(RID, new Date(READ_1)); + + const subscription = await persistedSubscription(); + // read state is updated at the client clock... + expect(subscription.ls?.getTime()).toBe(READ_1); + expect(subscription.unread).toBe(0); + expect(subscription.alert).toBe(false); + // ...but the server-derived cursor is left untouched + expect(subscription.lastOpen?.getTime()).toBe(T0); + }); +}); diff --git a/app/lib/database/__tests__/room.integration.test.ts b/app/lib/database/__tests__/room.integration.test.ts new file mode 100644 index 00000000000..cd2da753cf8 --- /dev/null +++ b/app/lib/database/__tests__/room.integration.test.ts @@ -0,0 +1,179 @@ +import { Q } from '@nozbe/watermelondb'; + +import type { TAppDatabase } from '../interfaces'; +import type { IMessage, TMessageModel } from '../../../definitions'; +import { + closeLokiTestDatabase, + createLokiTestDatabase, + resetLokiTestDatabase, + seedMessage, + seedSubscription, + withWriterQueueDiagnosticCleared +} from './lokiTestDatabase'; +import RoomSubscription from '../../methods/subscriptions/room'; +import { Encryption } from '../../encryption'; +import log from '../../methods/helpers/log'; + +// Real persistence + real writer lock: point `database.active` at the live LokiJS DB so +// `RoomSubscription.updateMessage` drives the real WMDB `db.write` serialization and the +// real `getMessageById` / `getThreadById` / `getThreadMessageById` services. Only the outer +// edges are mocked: `sdk`/`readMessages`/`loadMissedMessages` (never reached by updateMessage), +// `auxStore` (dispatch), `encryption` (pass-through decryption hook), `log`. +let mockActiveDatabase: TAppDatabase; +jest.mock('../../database', () => ({ + __esModule: true, + default: { + get active() { + return mockActiveDatabase; + } + } +})); + +jest.mock('../../store/auxStore', () => ({ + store: { + getState: () => ({ login: {}, settings: {}, room: {} }), + dispatch: jest.fn() + } +})); + +jest.mock('../../services/sdk', () => ({ + __esModule: true, + default: { + subscribeRoom: jest.fn(() => Promise.resolve([])), + onStreamData: jest.fn(() => Promise.resolve({ stop: jest.fn() })) + } +})); + +jest.mock('../../encryption', () => ({ + Encryption: { decryptMessage: jest.fn((message: unknown) => Promise.resolve(message)) } +})); + +jest.mock('../../methods/helpers/log', () => ({ __esModule: true, default: jest.fn() })); + +jest.mock('../../methods/readMessages', () => ({ readMessages: jest.fn() })); +jest.mock('../../methods/loadMissedMessages', () => ({ loadMissedMessages: jest.fn(() => Promise.resolve()) })); +jest.mock('../../methods/helpers/markMessagesRead', () => ({ __esModule: true, default: jest.fn() })); + +const decryptMessage = Encryption.decryptMessage as unknown as jest.Mock; +const mockedLog = log as unknown as jest.Mock; + +const RID = 'room-1'; +const T1 = Date.UTC(2026, 6, 22, 12, 0, 0); +const T2 = T1 + 1000; + +// A stream message as it reaches `updateMessage` (already built/decrypted upstream). `ts` and +// `_updatedAt` are explicit Dates: updateMessage does `Object.assign(record, message)` verbatim, +// so whatever is passed is what persists. +const streamMessage = (overrides: Partial = {}): IMessage => + ({ + _id: 'msg-1', + rid: RID, + msg: 'hello', + ts: new Date(T1), + _updatedAt: new Date(T1), + u: { _id: 'user-1', username: 'user-1' }, + ...overrides + } as unknown as IMessage); + +const findMessage = async (id: string): Promise => { + try { + return (await mockActiveDatabase.get('messages').find(id)) as TMessageModel; + } catch { + return null; + } +}; + +const messageCount = async (): Promise => + (await mockActiveDatabase.get('messages').query(Q.where('rid', RID)).fetch()).length; + +describe('RoomSubscription.updateMessage (LokiJS integration)', () => { + let sub: RoomSubscription; + + beforeAll(() => { + mockActiveDatabase = createLokiTestDatabase(); + }); + + afterAll(() => closeLokiTestDatabase(mockActiveDatabase)); + + beforeEach(async () => { + await resetLokiTestDatabase(mockActiveDatabase); + decryptMessage.mockClear(); + decryptMessage.mockImplementation((message: unknown) => Promise.resolve(message)); + mockedLog.mockClear(); + sub = new RoomSubscription(RID); + }); + + it('creates a new message row and runs the decryption hook (create path)', async () => { + await seedSubscription(mockActiveDatabase, { rid: RID }); + + await sub.updateMessage(streamMessage({ _id: 'created', msg: 'first' })); + + const row = await findMessage('created'); + expect(row?.msg).toBe('first'); + expect(row?.subscription?.id).toBe(RID); + expect(decryptMessage).toHaveBeenCalledTimes(1); + expect(mockedLog).not.toHaveBeenCalled(); + }); + + it('updates an existing message row in place without duplicating it (update path)', async () => { + await seedSubscription(mockActiveDatabase, { rid: RID }); + await seedMessage(mockActiveDatabase, { id: 'edited', rid: RID, msg: 'before', updatedAt: new Date(T1) }); + + await sub.updateMessage(streamMessage({ _id: 'edited', msg: 'after', _updatedAt: new Date(T2) })); + + const row = await findMessage('edited'); + expect(row?.msg).toBe('after'); + expect(await messageCount()).toBe(1); + expect(mockedLog).not.toHaveBeenCalled(); + }); + + it('returns early without decrypting or writing when the message rid does not match', async () => { + await seedSubscription(mockActiveDatabase, { rid: RID }); + + await sub.updateMessage(streamMessage({ _id: 'wrong-room', rid: 'different-room' })); + + expect(decryptMessage).not.toHaveBeenCalled(); + expect(await findMessage('wrong-room')).toBeNull(); + }); + + it('batches the message, thread and thread_message records under a single write', async () => { + await seedSubscription(mockActiveDatabase, { rid: RID }); + const batchSpy = jest.spyOn(mockActiveDatabase, 'batch'); + + await sub.updateMessage(streamMessage({ _id: 'threaded', msg: 'root', tlm: new Date(T1), tmid: 'parent-thread' })); + + // one batch, three prepared records → all committed under the same db.write + expect(batchSpy).toHaveBeenCalledTimes(1); + expect(batchSpy.mock.calls[0]).toHaveLength(3); + expect(await findMessage('threaded')).not.toBeNull(); + expect((await mockActiveDatabase.get('threads').find('threaded')).id).toBe('threaded'); + expect((await mockActiveDatabase.get('thread_messages').find('threaded')).id).toBe('threaded'); + + batchSpy.mockRestore(); + }); + + it('serializes two concurrent stream events for the same id without a "pending changes" throw', async () => { + await seedSubscription(mockActiveDatabase, { rid: RID }); + await seedMessage(mockActiveDatabase, { id: 'racing', rid: RID, msg: 'v0', updatedAt: new Date(T1) }); + + // Both events prepareUpdate the SAME cached record. Pre-fix (reads/prepares outside the + // writer lock) the second prepareUpdate hits a record with pending changes and throws; + // the fix serializes read → prepare → batch under one db.write so it can't. + await withWriterQueueDiagnosticCleared(() => + Promise.all([ + sub.updateMessage(streamMessage({ _id: 'racing', msg: 'v1', _updatedAt: new Date(T2) })), + sub.updateMessage(streamMessage({ _id: 'racing', msg: 'v2', _updatedAt: new Date(T2) })) + ]) + ); + + const pendingChangesLogged = mockedLog.mock.calls.some(([error]) => + /pending changes|was modified/i.test((error as Error)?.message ?? '') + ); + expect(pendingChangesLogged).toBe(false); + + // the record survived consistent: one coherent row, both writes applied in order (v2 last) + const row = await findMessage('racing'); + expect(row?.msg).toBe('v2'); + expect(await messageCount()).toBe(1); + }); +}); diff --git a/app/lib/database/services/Message.ts b/app/lib/database/services/Message.ts index 184ea29a0e9..202851483ba 100644 --- a/app/lib/database/services/Message.ts +++ b/app/lib/database/services/Message.ts @@ -1,9 +1,49 @@ +import { Q } from '@nozbe/watermelondb'; + import database from '..'; +import { type TMessageModel } from '../../../definitions'; +import { MESSAGE_TYPE_ANY_LOAD, type MessageTypeLoad } from '../../constants/messageTypeLoad'; +import { messagesStatus } from '../../constants/messagesStatus'; +import log from '../../methods/helpers/log'; import { type TAppDatabase } from '../interfaces'; import { MESSAGES_TABLE } from '../model/Message'; const getCollection = (db: TAppDatabase) => db.get(MESSAGES_TABLE); +// Head of the room scanned for a server-stamped row. Device-stamped rows cluster at the head (one +// load-more sentinel per loaded window, plus the unsent queue), and the `rid` index keeps the read +// cheap, so this is sized well past the handful expected rather than at the minimum. +const NEWEST_MESSAGE_SCAN = 20; + +// Synthetic load-more sentinels and unsent TEMP/ERROR rows are stamped with the device clock, so +// they are skipped: taking one as a sync cursor would ask the server for changes newer than a +// clock that may run ahead. They are discriminated in JS rather than by a query predicate because +// `t` and `status` are both nullable for ordinary messages, and adapter null semantics for +// `Q.notIn` differ between LokiJS and SQLite. +const isServerStamped = (message: TMessageModel) => + !MESSAGE_TYPE_ANY_LOAD.includes(message.t as MessageTypeLoad) && + (message.status ?? messagesStatus.SENT) === messagesStatus.SENT; + +/** + * Newest server stamp this device holds for a room — the point it is provably in sync with. + * Returns null when the whole head of the room is device-stamped, leaving the caller to fall + * back; the window scanned is sized so that only a large unsent backlog can exhaust it. + */ +export const getNewestMessageUpdatedAt = async (rid: string): Promise => { + const db = database.active; + const messageCollection = getCollection(db); + try { + const rows = (await messageCollection + .query(Q.where('rid', rid), Q.sortBy('_updated_at', Q.desc), Q.take(NEWEST_MESSAGE_SCAN)) + .fetch()) as TMessageModel[]; + const newest = rows.find(isServerStamped)?._updatedAt; + return newest ? new Date(newest) : null; + } catch (e) { + log(e); + return null; + } +}; + export const getMessageById = async (messageId: string | null) => { if (!messageId) { return null; diff --git a/app/lib/methods/helpers/advanceSyncCursor.ts b/app/lib/methods/helpers/advanceSyncCursor.ts new file mode 100644 index 00000000000..0a505ff48c5 --- /dev/null +++ b/app/lib/methods/helpers/advanceSyncCursor.ts @@ -0,0 +1,53 @@ +import database from '../../database'; +import { getSubscriptionByRoomId } from '../../database/services/Subscription'; +import { type ILastMessage, type IMessage, type TSubscriptionModel } from '../../../definitions'; +import log from './log'; + +export const maxUpdatedAt = (messages: (IMessage | ILastMessage)[]): number => { + let max = 0; + messages.forEach(message => { + if (!message._updatedAt) { + return; + } + const updatedAt = new Date(message._updatedAt).getTime(); + if (updatedAt > max) { + max = updatedAt; + } + }); + return max; +}; + +// lastOpen doubles as the chat.syncMessages cursor. `serverLatest` must be a server-stamped +// `_updatedAt` or `_deletedAt` (see maxUpdatedAt) and is 0 for an empty batch, so a client +// clock ahead of the server can't skip messages. +export const advanceSyncCursor = async (rid: string, serverLatest: number): Promise => { + try { + if (!serverLatest) { + return; + } + const subscription = await getSubscriptionByRoomId(rid); + // A sync that ran ahead of the rooms sync has no row to persist the cursor on. Dropping it + // costs a redundant fetch, not a message: the batch is already persisted, so once the row + // lands loadMissedMessages resolves the cursor off those messages and re-delivers the gap. + if (!subscription) { + return; + } + const current = subscription.lastOpen ? new Date(subscription.lastOpen).getTime() : 0; + if (serverLatest <= current) { + return; + } + const db = database.active; + await db.write(async () => { + await subscription.update((s: TSubscriptionModel) => { + // re-read inside the write: a concurrent sync may have advanced the + // cursor between the outer read and this commit (forward-only). + const committed = s.lastOpen ? new Date(s.lastOpen).getTime() : 0; + if (serverLatest > committed) { + s.lastOpen = new Date(serverLatest); + } + }); + }); + } catch (e) { + log(e); + } +}; diff --git a/app/lib/methods/loadMessagesForRoom.ts b/app/lib/methods/loadMessagesForRoom.ts index ce3d213c978..104994a1574 100644 --- a/app/lib/methods/loadMessagesForRoom.ts +++ b/app/lib/methods/loadMessagesForRoom.ts @@ -5,6 +5,7 @@ import { type IMessage, type TMessageModel } from '../../definitions'; import log from './helpers/log'; import { getMessageById } from '../database/services/Message'; import { getSubscriptionByRoomId } from '../database/services/Subscription'; +import { advanceSyncCursor, maxUpdatedAt } from './helpers/advanceSyncCursor'; import { type RoomTypes, roomTypeToApiType } from './roomTypeToApiType'; import sdk from '../services/sdk'; import { store } from '../store/auxStore'; @@ -128,8 +129,12 @@ export async function loadMessagesForRoom(args: { if (messages?.length) { const lastMessage = messages[messages.length - 1]; const lastMessageRecord = await getMessageById(lastMessage._id as string); + // The synthetic load-more row stays out of `messages`: it has no server _updatedAt, + // so normalizeMessage would stamp it with the device clock and advanceSyncCursor + // would read that wall clock as the cursor + const update = [...messages]; if (!lastMessageRecord && lastBatchWasFull) { - messages.push({ + update.push({ _id: generateLoadMoreId(lastMessage._id as string), rid: lastMessage.rid, ts: dayjs(lastMessage.ts).subtract(1, 'millisecond').toString(), @@ -137,7 +142,12 @@ export async function loadMessagesForRoom(args: { msg: lastMessage.msg } as IMessage); } - await updateMessages({ rid: args.rid, update: messages, loaderItem: args.loaderItem }); + await updateMessages({ rid: args.rid, update, loaderItem: args.loaderItem }); + // Older-history windows (latest set) can't advance the cursor: an edited old + // message's recent _updatedAt would jump it past unsynced newer messages + if (!args.latest) { + await advanceSyncCursor(args.rid, maxUpdatedAt(messages)); + } } } catch (e) { log(e); diff --git a/app/lib/methods/loadMissedMessages.ts b/app/lib/methods/loadMissedMessages.ts index ba2e6c23458..7b98ddad4c9 100644 --- a/app/lib/methods/loadMissedMessages.ts +++ b/app/lib/methods/loadMissedMessages.ts @@ -1,11 +1,14 @@ import { type ILastMessage } from '../../definitions'; import { compareServerVersion } from './helpers'; +import { advanceSyncCursor, maxUpdatedAt } from './helpers/advanceSyncCursor'; import updateMessages from './updateMessages'; import sdk from '../services/sdk'; import { store } from '../store/auxStore'; import { getSubscriptionByRoomId } from '../database/services/Subscription'; +import { getNewestMessageUpdatedAt } from '../database/services/Message'; const count = 50; +const MAX_PAGES = 10; const syncMessages = async ({ roomId, next, type }: { roomId: string; next: number; type: 'UPDATED' | 'DELETED' }) => { // @ts-ignore // this method dont have type @@ -19,34 +22,35 @@ const getSyncMessagesFromCursor = async ( updatedNext?: number | null, deletedNext?: number | null ) => { - const promises = []; + const isInitialFetch = !!lastOpen && !updatedNext && !deletedNext; + const updatedCursor = updatedNext || (isInitialFetch ? lastOpen : undefined); + const deletedCursor = deletedNext || (isInitialFetch ? lastOpen : undefined); - if (lastOpen && !updatedNext && !deletedNext) { - promises.push(syncMessages({ roomId, next: lastOpen, type: 'UPDATED' })); - promises.push(syncMessages({ roomId, next: lastOpen, type: 'DELETED' })); - } - if (updatedNext) { - promises.push(syncMessages({ roomId, next: updatedNext, type: 'UPDATED' })); - } - if (deletedNext) { - promises.push(syncMessages({ roomId, next: deletedNext, type: 'DELETED' })); - } + const [updatedResult, deletedResult] = await Promise.all([ + updatedCursor ? syncMessages({ roomId, next: updatedCursor, type: 'UPDATED' }) : undefined, + deletedCursor ? syncMessages({ roomId, next: deletedCursor, type: 'DELETED' }) : undefined + ]); - const [updatedMessages, deletedMessages] = await Promise.all(promises); return { - deleted: deletedMessages?.deleted ?? [], - deletedNext: deletedMessages?.cursor.next, - updated: updatedMessages?.updated ?? [], - updatedNext: updatedMessages?.cursor.next + deleted: deletedResult?.deleted ?? [], + deletedNext: deletedResult?.cursor?.next ?? null, + updated: updatedResult?.updated ?? [], + updatedNext: updatedResult?.cursor?.next ?? null }; }; -const getLastUpdate = async (rid: string) => { +// A room never opened on this device (a push notification tap) has no lastOpen, and an undefined +// cursor makes the sync fetch nothing at all. Candidates are ordered by how well each answers +// "newest point THIS device is in sync with" — `ls` is written by every device, so it can sit past +// messages this one never received. Epoch 0 means an unseeded non-optional column, not a cursor. +const getLastUpdate = async (rid: string): Promise => { const sub = await getSubscriptionByRoomId(rid); if (!sub) { return null; } - return sub.lastOpen; + const candidates = [sub.lastOpen, await getNewestMessageUpdatedAt(rid), sub.ls, sub.ts]; + const cursor = candidates.find(candidate => candidate && new Date(candidate).getTime() > 0); + return cursor ? new Date(cursor) : null; }; async function load({ @@ -86,35 +90,86 @@ async function load({ return result; } -export async function loadMissedMessages(args: { +interface ISyncPagesResult { + /** Newest server `_updatedAt` seen across the whole chain, 0 when no update arrived. */ + highestUpdatedAt: number; + /** DELETED cursor the walk stopped at, null when that stream was exhausted. */ + deletedStop: number | null; +} + +// Persists each page as it arrives (idempotent) and threads the newest server `_updatedAt` +// seen so far, so no page payload outlives the page. The cursor advances only after the whole +// chain succeeds: advancing per page can skip records the other stream still has pending if a +// later page fails. The chain is bounded, and UPDATED and DELETED are independent cursors, so +// the DELETED stop-point travels back out — the persisted cursor must not outrun either stream. +async function syncPages(args: { rid: string; lastOpen?: Date; updatedNext?: number | null; deletedNext?: number | null; -}): Promise { + page: number; + highestUpdatedAt: number; +}): Promise { const data = await load({ rid: args.rid, lastOpen: args.lastOpen, updatedNext: args.updatedNext, deletedNext: args.deletedNext }); - if (data) { - const { - updated, - updatedNext, - deleted, - deletedNext - }: { updated: ILastMessage[]; deleted: ILastMessage[]; updatedNext: number | null; deletedNext: number | null } = data; - // @ts-ignore // TODO: remove loaderItem obligatoriness - await updateMessages({ rid: args.rid, update: updated, remove: deleted }); + if (!data) { + return { highestUpdatedAt: args.highestUpdatedAt, deletedStop: args.deletedNext ?? null }; + } + const { + updated, + updatedNext, + deleted, + deletedNext + }: { + updated: ILastMessage[]; + // the server projects deleted records down to these two fields, so they carry no + // `_updatedAt` and can never feed the cursor + deleted: { _id: string; _deletedAt: string }[]; + updatedNext: number | null; + deletedNext: number | null; + } = data; + // @ts-ignore // the sync payload is ILastMessage[]; updateMessages types update as IMessage[] + await updateMessages({ rid: args.rid, update: updated, remove: deleted }); + const highestUpdatedAt = Math.max(args.highestUpdatedAt, maxUpdatedAt(updated)); - if (deletedNext || updatedNext) { - loadMissedMessages({ + if (deletedNext || updatedNext) { + if (args.page + 1 < MAX_PAGES) { + return syncPages({ rid: args.rid, lastOpen: args.lastOpen, updatedNext, - deletedNext + deletedNext, + page: args.page + 1, + highestUpdatedAt }); } + return { highestUpdatedAt, deletedStop: deletedNext }; + } + return { highestUpdatedAt, deletedStop: null }; +} + +// Both streams are filtered exclusively (`$gt`), so resuming from the lower of the two +// stop-points costs a redundant refetch on the stream that ran further, never a record. +const resolveCursor = ({ highestUpdatedAt, deletedStop }: ISyncPagesResult): number => { + if (deletedStop === null) { + return highestUpdatedAt; + } + if (!highestUpdatedAt) { + return deletedStop; } + return Math.min(highestUpdatedAt, deletedStop); +}; + +export async function loadMissedMessages(args: { + rid: string; + lastOpen?: Date; + updatedNext?: number | null; + deletedNext?: number | null; +}): Promise { + const result = await syncPages({ ...args, page: 0, highestUpdatedAt: 0 }); + await advanceSyncCursor(args.rid, resolveCursor(result)); } diff --git a/app/lib/methods/readMessages.ts b/app/lib/methods/readMessages.ts index 790cafbe505..d086ef54488 100644 --- a/app/lib/methods/readMessages.ts +++ b/app/lib/methods/readMessages.ts @@ -5,7 +5,7 @@ import sdk from '../services/sdk'; import { hasE2EEWarning } from '../encryption/utils'; import { store } from '../store/auxStore'; -export async function readMessages(rid: string, ls: Date, updateLastOpen = false): Promise { +export async function readMessages(rid: string, ls: Date): Promise { try { const db = database.active; let subscription; @@ -45,9 +45,6 @@ export async function readMessages(rid: string, ls: Date, updateLastOpen = false s.userMentions = 0; s.groupMentions = 0; s.ls = ls; - if (updateLastOpen) { - s.lastOpen = ls; - } }); } catch (e) { // Do nothing diff --git a/app/lib/methods/subscriptions/room.test.ts b/app/lib/methods/subscriptions/room.test.ts index 830d0d0bf1d..d4f4bbe7acf 100644 --- a/app/lib/methods/subscriptions/room.test.ts +++ b/app/lib/methods/subscriptions/room.test.ts @@ -1,9 +1,6 @@ import RoomSubscription from './room'; import { loadMissedMessages } from '../loadMissedMessages'; import { clearUserTyping } from '../../../actions/usersTyping'; -import { getMessageById } from '../../database/services/Message'; -import { getThreadById } from '../../database/services/Thread'; -import log from '../helpers/log'; const mockSubscribeRoom = jest.fn, [string]>(() => Promise.resolve([])); const mockOnStreamData = jest.fn, [string, (...args: unknown[]) => void]>(() => @@ -61,10 +58,6 @@ jest.mock('../helpers/markMessagesRead', () => ({ default: jest.fn() })); -jest.mock('../updateLastOpen', () => ({ - updateLastOpen: jest.fn() -})); - jest.mock('../../../actions/usersTyping', () => ({ addUserTyping: jest.fn(), clearUserTyping: jest.fn().mockReturnValue({ type: 'CLEAR_USER_TYPING' }), @@ -193,42 +186,47 @@ describe('RoomSubscription', () => { expect(mockSubscribeRoom).not.toHaveBeenCalled(); }); - it('tears down stale subscriptions on reconnect and tracks fresh ones for later cleanup', async () => { + it('re-subscribes on reconnect without tearing down the stale subscriptions', async () => { const staleSub = { unsubscribe: jest.fn(() => Promise.resolve()) }; const freshSub = { unsubscribe: jest.fn(() => Promise.resolve()) }; mockSubscribeRoom.mockResolvedValueOnce([staleSub]).mockResolvedValueOnce([freshSub]); await sub.subscribe(); await sub.handleLogin(); + + // A fresh DDP session already killed the server-side subs, so handleLogin must not churn them. + expect(staleSub.unsubscribe).not.toHaveBeenCalled(); + await sub.unsubscribe(); - expect(staleSub.unsubscribe).toHaveBeenCalledTimes(1); expect(freshSub.unsubscribe).toHaveBeenCalledTimes(1); }); it('does not accumulate subscriptions across repeated handleLogin calls (simulates sequential reopen)', async () => { const first = { unsubscribe: jest.fn(() => Promise.resolve()) }; const second = { unsubscribe: jest.fn(() => Promise.resolve()) }; - mockSubscribeRoom.mockResolvedValueOnce([first]).mockResolvedValueOnce([second]); + const third = { unsubscribe: jest.fn(() => Promise.resolve()) }; + mockSubscribeRoom.mockResolvedValueOnce([first]).mockResolvedValueOnce([second]).mockResolvedValueOnce([third]); await sub.subscribe(); expect(mockSubscribeRoom).toHaveBeenCalledTimes(1); - // First reopen → tears down [first], creates [second] + // First reopen → tracks [second]; [first] is left alone (SDK owns the replay/dedup). await sub.handleLogin(); expect(mockSubscribeRoom).toHaveBeenCalledTimes(2); - expect(first.unsubscribe).toHaveBeenCalledTimes(1); - expect(second.unsubscribe).not.toHaveBeenCalled(); + expect(first.unsubscribe).not.toHaveBeenCalled(); - // Second reopen → tears down [second], creates [] + // Second reopen → tracks [third]; [second] left alone. await sub.handleLogin(); expect(mockSubscribeRoom).toHaveBeenCalledTimes(3); - expect(second.unsubscribe).toHaveBeenCalledTimes(1); + expect(second.unsubscribe).not.toHaveBeenCalled(); - // Final cleanup → empty batch, no more unsubscribes + // The app-side promise list holds one batch (replaced, not grown): final cleanup tears down + // only the current [third]. SDK-side dedup of the replays is proven in ddpSocket.test.ts. await sub.unsubscribe(); - expect(first.unsubscribe).toHaveBeenCalledTimes(1); - expect(second.unsubscribe).toHaveBeenCalledTimes(1); + expect(first.unsubscribe).not.toHaveBeenCalled(); + expect(second.unsubscribe).not.toHaveBeenCalled(); + expect(third.unsubscribe).toHaveBeenCalledTimes(1); }); it('does not call onStreamData inside handleLogin (listeners persist across reopen)', async () => { @@ -283,48 +281,4 @@ describe('RoomSubscription', () => { expect(mockSubscribeRoom).toHaveBeenCalledWith(rid); }); }); - - describe('updateMessage concurrency', () => { - const makeRecord = (debugName: string) => ({ - _preparedState: null as string | null, - prepareUpdate(recordUpdater: (m: any) => void) { - if (this._preparedState) { - throw new Error(`Cannot update a record with pending changes (${debugName})`); - } - recordUpdater(this); - this._preparedState = 'update'; - return this; - } - }); - - it('does not throw "pending changes" when two stream events for the same message id arrive concurrently', async () => { - const _id = 'KXse45i7gGYE8j4Xb'; - const messageRecord = makeRecord(`messages#${_id}`); - const threadRecord = makeRecord(`threads#${_id}`); - (getMessageById as jest.Mock).mockResolvedValue(messageRecord); - (getThreadById as jest.Mock).mockResolvedValue(threadRecord); - // db.batch commits prepared records, clearing their pending state (like the real writer). - mockDbBatch.mockImplementation((...items: any[]) => { - items.forEach(item => { - if (item && typeof item === 'object' && '_preparedState' in item) { - item._preparedState = null; - } - }); - return Promise.resolve(undefined); - }); - - const message = { _id, rid, tlm: { $date: 1 } } as any; - - // updateMessage's promise never resolves on the happy path, so fire both and flush the queues. - sub.updateMessage({ ...message }); - sub.updateMessage({ ...message }); - await Array.from({ length: 10 }).reduce>( - chain => chain.then(() => new Promise(resolve => setImmediate(resolve))), - Promise.resolve() - ); - - const loggedPendingChanges = (log as jest.Mock).mock.calls.some(([err]) => /pending changes/.test(err?.message)); - expect(loggedPendingChanges).toBe(false); - }); - }); }); diff --git a/app/lib/methods/subscriptions/room.ts b/app/lib/methods/subscriptions/room.ts index 1e88d165453..564137a6e6f 100644 --- a/app/lib/methods/subscriptions/room.ts +++ b/app/lib/methods/subscriptions/room.ts @@ -27,7 +27,6 @@ import { type IDDPMessage } from '../../../definitions/IDDPMessage'; import sdk from '../../services/sdk'; import { readMessages } from '../readMessages'; import { loadMissedMessages } from '../loadMissedMessages'; -import { updateLastOpen } from '../updateLastOpen'; import markMessagesRead from '../helpers/markMessagesRead'; export default class RoomSubscription { @@ -64,7 +63,6 @@ export default class RoomSubscription { unsubscribe = async () => { console.log(`[RCRN] Unsubscribing from room ${this.rid}`); - updateLastOpen(this.rid); this.isAlive = false; reduxStore.dispatch(unsubscribeRoom(this.rid)); if (this.promises) { @@ -98,10 +96,8 @@ export default class RoomSubscription { return; } try { - if (this.promises) { - const oldSubs = await this.promises; - oldSubs?.forEach(sub => sub?.unsubscribe?.().catch(() => {})); - } + // A fresh DDP session means the prior server-side subs are already dead, so tearing them + // down here is pointless churn; the SDK replays and dedups the re-subscribe on login. this.promises = sdk.subscribeRoom(this.rid); await this.promises; if (!this.isAlive) { diff --git a/app/lib/methods/updateLastOpen.ts b/app/lib/methods/updateLastOpen.ts deleted file mode 100644 index e264648afce..00000000000 --- a/app/lib/methods/updateLastOpen.ts +++ /dev/null @@ -1,21 +0,0 @@ -import database from '../database'; -import { getSubscriptionByRoomId } from '../database/services/Subscription'; -import log from './helpers/log'; -import { type TSubscriptionModel } from '../../definitions'; - -export async function updateLastOpen(rid: string, lastOpen = new Date()): Promise { - try { - const db = database.active; - const subscription = await getSubscriptionByRoomId(rid); - if (!subscription) { - return; - } - await db.write(async () => { - await subscription.update((s: TSubscriptionModel) => { - s.lastOpen = lastOpen; - }); - }); - } catch (e) { - log(e); - } -} diff --git a/app/lib/services/connect.test.ts b/app/lib/services/connect.test.ts index d3ac06b24ed..1955987ca52 100644 --- a/app/lib/services/connect.test.ts +++ b/app/lib/services/connect.test.ts @@ -602,4 +602,47 @@ describe('connect — stream-notify-logged updateAvatar', () => { }); }); +describe('connect — connected-guard (Zombie Connection resume)', () => { + beforeEach(() => { + jest.clearAllMocks(); + mockOnStreamDataStops.length = 0; + }); + + type ResumeAction = { credentials?: { resume?: string } }; + const resumeActions = (): ResumeAction[] => + mockStoreDispatch.mock.calls.map(([action]) => action as ResumeAction).filter(action => action?.credentials?.resume); + + it('re-runs resume-login on a Zombie Connection (connected + !isAuthenticated + token)', async () => { + mockStoreGetState.mockReturnValue({ + meteor: { connected: true }, + login: { user: { token: 'tok' }, isAuthenticated: false }, + settings: {} + }); + + await connect({ server: 'https://example.com' }); + mockStoreDispatch.mockClear(); + + getHandlersByEvent('connected')[0](); + + const resumes = resumeActions(); + expect(resumes).toHaveLength(1); + expect(resumes[0].credentials).toEqual({ resume: 'tok' }); + }); + + it('early-returns on a healthy connected + isAuthenticated (no relogin storm)', async () => { + mockStoreGetState.mockReturnValue({ + meteor: { connected: true }, + login: { user: { token: 'tok' }, isAuthenticated: true }, + settings: {} + }); + + await connect({ server: 'https://example.com' }); + mockStoreDispatch.mockClear(); + + getHandlersByEvent('connected')[0](); + + expect(resumeActions()).toHaveLength(0); + }); +}); + // Note: Apple authentication when isIOS is true is tested in connect.ios.test.ts diff --git a/app/lib/services/connect.ts b/app/lib/services/connect.ts index dce7f343396..c32818c78d7 100644 --- a/app/lib/services/connect.ts +++ b/app/lib/services/connect.ts @@ -118,11 +118,13 @@ function connect({ server, logoutOnError = false }: { server: string; logoutOnEr connectedListener = sdk.current.onStreamData('connected', () => { const { connected } = store.getState().meteor; - if (connected) { + const { user, isAuthenticated } = store.getState().login; + // A real handshake on an unauthenticated session (a Zombie Connection: connected + + // !isAuthenticated + token) must re-run resume-login instead of stranding with 0 subs. + if (connected && isAuthenticated) { return; } store.dispatch(connectSuccess()); - const { user } = store.getState().login; if (user?.token) { store.dispatch(loginRequest({ resume: user.token }, logoutOnError)); } diff --git a/app/lib/services/ddpSocket.test.ts b/app/lib/services/ddpSocket.test.ts index 1fe4a165697..db388d3321b 100644 --- a/app/lib/services/ddpSocket.test.ts +++ b/app/lib/services/ddpSocket.test.ts @@ -106,6 +106,80 @@ describe('Socket.forceReopen awaitability', () => { }); }); +describe('Socket subscription restoration (lib-led)', () => { + const seed = (socket: any, name: string, params: unknown[], id: string) => { + socket.subscriptions[id] = { + id, + name, + params, + unsubscribe: jest.fn(() => Promise.resolve()), + onEvent: jest.fn() + }; + }; + + const subFrames = (send: jest.Mock) => + send.mock.calls.map(([raw]) => JSON.parse(raw as string)).filter(frame => frame.msg === 'sub'); + + it('forceReopen keeps the subscription registry (subtraction 1)', async () => { + const { socket } = buildSocket(); + socket.open = jest.fn(() => Promise.resolve()); + seed(socket, 'stream-notify-user', ['u1/message'], 'ddp-1'); + + await socket.forceReopen(); + + expect(Object.keys(socket.subscriptions)).toEqual(['ddp-1']); + }); + + it('subscribeAll replays each preserved sub exactly once, reusing its original id', () => { + const { socket, send } = buildSocket(); + seed(socket, 'stream-notify-user', ['u1/message'], 'ddp-1'); + seed(socket, 'stream-room-messages', ['rid-1'], 'ddp-2'); + + socket.subscribeAll(); + + const frames = subFrames(send); + expect(frames).toHaveLength(2); + expect(frames.map(f => f.id).sort()).toEqual(['ddp-1', 'ddp-2']); + }); + + it('dedups two subscribe calls with identical name+params: one sub, one entry, both callbacks fire', async () => { + const { socket, send } = buildSocket(); + const cb1 = jest.fn(); + const cb2 = jest.fn(); + + const first = socket.subscribe('stream-room-messages', ['rid-1'], cb1); + socket.subscribe('stream-room-messages', ['rid-1'], cb2); + + const frames = subFrames(send); + expect(frames).toHaveLength(1); + + // Resolve the single sub so the first subscriber's callback attaches (confirmed path). + socket.emit('ready', { subs: [frames[0].id] }); + await first; + + const entries = Object.values(socket.subscriptions).filter((s: any) => s.name === 'stream-room-messages'); + expect(entries).toHaveLength(1); + + socket.emit('stream-room-messages', { fields: {} }); + expect(cb1).toHaveBeenCalledTimes(1); + expect(cb2).toHaveBeenCalledTimes(1); + }); + + it('keeps the registry flat across replay+re-subscribe cycles and per-room params separate', () => { + const { socket } = buildSocket(); + seed(socket, 'stream-room-messages', ['rid-1'], 'ddp-1'); + seed(socket, 'stream-room-messages', ['rid-2'], 'ddp-2'); + + for (let cycle = 0; cycle < 3; cycle += 1) { + socket.subscribeAll(); + socket.subscribe('stream-room-messages', ['rid-1']); + socket.subscribe('stream-room-messages', ['rid-2']); + } + + expect(Object.keys(socket.subscriptions)).toHaveLength(2); + }); +}); + describe('Socket.checkAndReopen bucket dispatch', () => { const buildWithSpies = () => { const { socket } = buildSocket(); diff --git a/app/sagas/__tests__/state.test.ts b/app/sagas/__tests__/state.test.ts new file mode 100644 index 00000000000..c0be2626603 --- /dev/null +++ b/app/sagas/__tests__/state.test.ts @@ -0,0 +1,73 @@ +jest.mock('../../lib/methods/helpers/localAuthentication', () => ({ + localAuthenticate: jest.fn(() => Promise.resolve()), + saveLastLocalAuthenticationSession: jest.fn(() => Promise.resolve()) +})); + +jest.mock('../../lib/services/connect', () => ({ + checkAndReopen: jest.fn(() => Promise.resolve()) +})); + +jest.mock('../../lib/services/restApi', () => ({ + setUserPresenceOnline: jest.fn(() => Promise.resolve()), + setUserPresenceAway: jest.fn(() => Promise.resolve()) +})); + +jest.mock('../../lib/notifications', () => ({ + checkPendingNotification: jest.fn(() => Promise.resolve()) +})); + +jest.mock('../../lib/methods/helpers/log', () => ({ + __esModule: true, + default: jest.fn() +})); + +import { put } from 'redux-saga/effects'; + +import { RootEnum } from '../../definitions'; +import { loginRequest } from '../../actions/login'; +import { checkAndReopen } from '../../lib/services/connect'; +import { appHasComeBackToForeground } from '../state'; + +// Manual saga-test style (no redux-saga-test-plan): drive the generator by feeding each +// `select` result and inspecting the yielded effects. Enters at ROOT_INSIDE for every case. +describe('state saga — appHasComeBackToForeground resume gate', () => { + beforeEach(() => { + jest.clearAllMocks(); + }); + + it('dispatches loginRequest({ resume }) when inside root, unauthenticated, token present', () => { + const gen = appHasComeBackToForeground(); + gen.next(); // yields select(app.root) + gen.next(RootEnum.ROOT_INSIDE); // yields select(login) + const effect = gen.next({ isAuthenticated: false, user: { token: 'tok' } }).value; + + expect(effect).toEqual(put(loginRequest({ resume: 'tok' }))); + expect(gen.next().done).toBe(true); + }); + + it('bails without dispatching when unauthenticated and no token', () => { + const gen = appHasComeBackToForeground(); + gen.next(); + gen.next(RootEnum.ROOT_INSIDE); + const result = gen.next({ isAuthenticated: false, user: null }); + + expect(result.done).toBe(true); + expect(result.value).toBeUndefined(); + }); + + it('runs the authenticated reopen path with no redundant resume', () => { + const gen = appHasComeBackToForeground(); + gen.next(); + gen.next(RootEnum.ROOT_INSIDE); + + const values: unknown[] = []; + let result = gen.next({ isAuthenticated: true, user: { token: 'tok' } }); + while (!result.done) { + values.push(result.value); + result = gen.next(undefined); + } + + expect(checkAndReopen).toHaveBeenCalledTimes(1); + expect(values).not.toContainEqual(put(loginRequest({ resume: 'tok' }))); + }); +}); diff --git a/app/sagas/state.js b/app/sagas/state.js index 44f0b1430fa..fd63a796489 100644 --- a/app/sagas/state.js +++ b/app/sagas/state.js @@ -1,7 +1,8 @@ -import { select, takeLatest } from 'redux-saga/effects'; +import { put, select, takeLatest } from 'redux-saga/effects'; import log from '../lib/methods/helpers/log'; import { localAuthenticate, saveLastLocalAuthenticationSession } from '../lib/methods/helpers/localAuthentication'; +import { loginRequest } from '../actions/login'; import { APP_STATE } from '../actions/actionsTypes'; import { RootEnum } from '../definitions'; import { checkAndReopen } from '../lib/services/connect'; @@ -14,13 +15,18 @@ const isAuthAndConnected = function* isAuthAndConnected() { return login.isAuthenticated && meteor.connected; }; -const appHasComeBackToForeground = function* appHasComeBackToForeground() { +export const appHasComeBackToForeground = function* appHasComeBackToForeground() { const appRoot = yield select(state => state.app.root); if (appRoot !== RootEnum.ROOT_INSIDE) { return; } const login = yield select(state => state.login); if (!login.isAuthenticated) { + // A Zombie Connection (inside root, token present but not authenticated) resumes login + // instead of bailing; the authenticated path below owns the reconnect. + if (login.user?.token) { + yield put(loginRequest({ resume: login.user.token })); + } return; } try { diff --git a/app/views/RoomView/index.tsx b/app/views/RoomView/index.tsx index edffea7126a..1df9f12f9f7 100644 --- a/app/views/RoomView/index.tsx +++ b/app/views/RoomView/index.tsx @@ -115,6 +115,7 @@ import { type IRoomFederated, isRoomFederated, isRoomNativeFederated } from '../ import { InvitedRoom } from './components/InvitedRoom'; import { getInvitationData } from '../../lib/methods/getInvitationData'; import { isInviteSubscription } from '../../lib/methods/isInviteSubscription'; +import { InitRetryScheduler } from './services/initRetryScheduler'; const EMPTY_HIDE_SYSTEM_MESSAGES: string[] = []; @@ -134,7 +135,7 @@ class RoomView extends Component { private subObserveQuery?: Subscription; private subSubscription?: Subscription; private queryUnreads?: Subscription; - private retryInitTimeout?: ReturnType; + private initRetry = new InitRetryScheduler(); private messageErrorActions?: IMessageErrorActions | null; private messageActions?: IMessageActions | null; // Type of InteractionManager.runAfterInteractions @@ -379,9 +380,7 @@ class RoomView extends Component { if (this.queryUnreads && this.queryUnreads.unsubscribe) { this.queryUnreads.unsubscribe(); } - if (this.retryInitTimeout) { - clearTimeout(this.retryInitTimeout); - } + this.initRetry.cancel(); if (this.unsubscribeBlur) { this.unsubscribeBlur(); } @@ -678,7 +677,7 @@ class RoomView extends Component { this.consumeJumpParam(messageId); } } else { - const newLastOpen = new Date(); + const readAt = new Date(); await RoomServices.getMessages({ rid: room.rid, t: room.t as RoomType, @@ -692,7 +691,7 @@ class RoomView extends Component { } else { this.setLastOpen(null); } - readMessages(room.rid, newLastOpen, true).catch(e => console.log(e)); + readMessages(room.rid, readAt).catch(e => console.log(e)); } } @@ -700,11 +699,14 @@ class RoomView extends Component { const member = await this.getRoomMember(); this.setState({ canAutoTranslate, member, loading: false }); + this.initRetry.reset(); } catch (e) { this.setState({ loading: false }); - this.retryInitTimeout = setTimeout(() => { - this.init(); - }, 300); + // one record per broken room open: the retry loop itself must not spam the log + if (this.initRetry.isFirstAttempt) { + log(e); + } + this.initRetry.schedule(this.init); } }; diff --git a/app/views/RoomView/services/initRetryScheduler.test.ts b/app/views/RoomView/services/initRetryScheduler.test.ts new file mode 100644 index 00000000000..75653a6c8d8 --- /dev/null +++ b/app/views/RoomView/services/initRetryScheduler.test.ts @@ -0,0 +1,93 @@ +import { INIT_MAX_RETRY_DELAY, InitRetryScheduler } from './initRetryScheduler'; + +describe('InitRetryScheduler', () => { + beforeEach(() => jest.useFakeTimers()); + + afterEach(() => { + jest.clearAllTimers(); + jest.useRealTimers(); + }); + + const delaysOf = (scheduler: InitRetryScheduler, retry: () => void, attempts: number): number[] => { + const delays: number[] = []; + for (let attempt = 0; attempt < attempts; attempt += 1) { + const before = Date.now(); + scheduler.schedule(retry); + jest.runOnlyPendingTimers(); + delays.push(Date.now() - before); + } + return delays; + }; + + it('backs off exponentially instead of retrying at a fixed interval', () => { + const retry = jest.fn(); + + expect(delaysOf(new InitRetryScheduler(), retry, 5)).toEqual([300, 600, 1200, 2400, 4800]); + expect(retry).toHaveBeenCalledTimes(5); + }); + + it('clamps the backoff at the delay floor and keeps retrying there indefinitely', () => { + const retry = jest.fn(); + + const delays = delaysOf(new InitRetryScheduler(), retry, 12); + + expect(delays.slice(6)).toEqual(Array(6).fill(INIT_MAX_RETRY_DELAY)); + // self-healing survives any number of failures: the room never stops trying + expect(retry).toHaveBeenCalledTimes(12); + }); + + it('reset restores the initial delay', () => { + const scheduler = new InitRetryScheduler(); + const retry = jest.fn(); + delaysOf(scheduler, retry, 8); + + scheduler.reset(); + + expect(delaysOf(scheduler, retry, 1)).toEqual([300]); + }); + + it('flags only the first arm since reset, so a caller logs one record per broken open', () => { + const scheduler = new InitRetryScheduler(); + const retry = jest.fn(); + + expect(scheduler.isFirstAttempt).toBe(true); + delaysOf(scheduler, retry, 3); + expect(scheduler.isFirstAttempt).toBe(false); + + scheduler.reset(); + expect(scheduler.isFirstAttempt).toBe(true); + }); + + it('cancel drops a pending retry', () => { + const scheduler = new InitRetryScheduler(); + const retry = jest.fn(); + + scheduler.schedule(retry); + scheduler.cancel(); + jest.runOnlyPendingTimers(); + + expect(retry).not.toHaveBeenCalled(); + }); + + it('reset also drops a pending retry', () => { + const scheduler = new InitRetryScheduler(); + const retry = jest.fn(); + + scheduler.schedule(retry); + scheduler.reset(); + jest.runOnlyPendingTimers(); + + expect(retry).not.toHaveBeenCalled(); + }); + + it('keeps only the newest pending retry when scheduled twice', () => { + const scheduler = new InitRetryScheduler(); + const retry = jest.fn(); + + scheduler.schedule(retry); + scheduler.schedule(retry); + jest.runOnlyPendingTimers(); + + expect(retry).toHaveBeenCalledTimes(1); + }); +}); diff --git a/app/views/RoomView/services/initRetryScheduler.ts b/app/views/RoomView/services/initRetryScheduler.ts new file mode 100644 index 00000000000..55f59121d11 --- /dev/null +++ b/app/views/RoomView/services/initRetryScheduler.ts @@ -0,0 +1,36 @@ +const INITIAL_RETRY_DELAY = 300; +export const INIT_MAX_RETRY_DELAY = 10000; + +// A failing room init used to re-fire every 300ms forever, so a server that keeps +// rejecting turned one open room into a request storm. What is capped is the rate, not +// the number of attempts: the delay doubles up to the floor and stays there, because +// nothing else re-enters init() for an authenticated room and a room that stopped +// retrying has no error UI and no way back other than leaving it. +export class InitRetryScheduler { + private attempts = 0; + private timeout?: ReturnType; + + /** True until the first retry since the last reset is armed: the failure worth recording. */ + get isFirstAttempt(): boolean { + return this.attempts === 0; + } + + schedule(retry: () => void): void { + const delay = Math.min(INITIAL_RETRY_DELAY * 2 ** this.attempts, INIT_MAX_RETRY_DELAY); + this.attempts += 1; + this.cancel(); + this.timeout = setTimeout(retry, delay); + } + + reset(): void { + this.attempts = 0; + this.cancel(); + } + + cancel(): void { + if (this.timeout) { + clearTimeout(this.timeout); + this.timeout = undefined; + } + } +} diff --git a/jest.config.js b/jest.config.js index aa5e3a720c6..f5ccdae3cb8 100644 --- a/jest.config.js +++ b/jest.config.js @@ -1,6 +1,13 @@ module.exports = { modulePathIgnorePatterns: ['/.*worktrees/'], - testPathIgnorePatterns: ['e2e', 'node_modules', '/.*worktrees/', '/__tests__/testHelpers\\.tsx$'], + testPathIgnorePatterns: [ + 'e2e', + 'node_modules', + '/.*worktrees/', + '/__tests__/testHelpers\\.tsx$', + '/__tests__/lokiTestDatabase\\.ts$', + '/__tests__/fakeSyncServer\\.ts$' + ], transformIgnorePatterns: [ 'node_modules/(?!((jest-)?react-native|@react-native(-community)?)|expo(nent)?|@expo(nent)?/.*|@expo-google-fonts/.*|react-navigation|@react-navigation/.*|@unimodules/.*|unimodules|sentry-expo|native-base|react-native-svg|@rocket.chat/ui-kit|@rocket.chat/sdk|tiny-events)' ], diff --git a/patches/@rocket.chat+sdk+1.3.3-mobile.patch b/patches/@rocket.chat+sdk+1.3.3-mobile.patch index 3cac42dab33..67fe6050f10 100644 --- a/patches/@rocket.chat+sdk+1.3.3-mobile.patch +++ b/patches/@rocket.chat+sdk+1.3.3-mobile.patch @@ -1,8 +1,18 @@ diff --git a/node_modules/@rocket.chat/sdk/lib/drivers/ddp.ts b/node_modules/@rocket.chat/sdk/lib/drivers/ddp.ts -index 19d31ae..07bda5f 100644 +index 19d31ae..6e94d91 100644 --- a/node_modules/@rocket.chat/sdk/lib/drivers/ddp.ts +++ b/node_modules/@rocket.chat/sdk/lib/drivers/ddp.ts -@@ -101,9 +101,25 @@ export class Socket extends EventEmitter { +@@ -42,6 +42,9 @@ import { sha256 } from 'js-sha256' + + const userDisconnectCloseCode = 4000; + ++/** Idempotency key for a subscription: identical (name, params) collapse to one sub. */ ++const subscriptionKey = (name: string, params: any): string => `${name}\u0000${JSON.stringify(params)}` ++ + /** Websocket handler class, manages connections and subscriptions by DDP */ + export class Socket extends EventEmitter { + sent = 0 +@@ -101,9 +104,25 @@ export class Socket extends EventEmitter { this.logger.error(err) return reject(err) } @@ -29,7 +39,7 @@ index 19d31ae..07bda5f 100644 this.connection.onopen = this.onOpen.bind(this, resolve) this.emit('connecting') }) -@@ -125,7 +141,14 @@ export class Socket extends EventEmitter { +@@ -125,7 +144,14 @@ export class Socket extends EventEmitter { } /** Emit close event so it can be used for promise resolve in close() */ @@ -45,7 +55,7 @@ index 19d31ae..07bda5f 100644 this.emit('close', e) try { if (e?.code !== userDisconnectCloseCode) { -@@ -175,15 +198,115 @@ export class Socket extends EventEmitter { +@@ -175,15 +201,119 @@ export class Socket extends EventEmitter { return Promise.resolve() } @@ -85,6 +95,10 @@ index 19d31ae..07bda5f 100644 + + _reopenInFlight: Promise | null = null + ++ // In-flight (name, params) claims, keyed synchronously at subscribe() entry so a twin ++ // subscribe racing the un-awaited subscribeAll() replay dedups before the registry write lands. ++ _pendingSubscriptions: { [key: string]: Promise } = {} ++ + /** + * Tear down the existing connection and start a fresh one. Used when the + * socket is known or suspected dead. Emits `disconnected` so any in-flight @@ -112,7 +126,7 @@ index 19d31ae..07bda5f 100644 + } + this.emit('disconnected') + this.emit('close', { code: userDisconnectCloseCode }) -+ this.subscriptions = {} ++ // Registry survives the reopen so the next login replays every sub via subscribeAll(). + if (this.connection) { + try { + this.connection.onopen = null as any @@ -148,7 +162,7 @@ index 19d31ae..07bda5f 100644 + this.logger.info(`[ddp] checkAndReopen bucket=stale elapsed=${elapsed}`) + await this.forceReopen() + return -+ } + } + if (elapsed > this.config.ping * 2) { + this.logger.info(`[ddp] checkAndReopen bucket=stale elapsed=${elapsed}`) + await this.forceReopen() @@ -162,13 +176,92 @@ index 19d31ae..07bda5f 100644 + if (alive) { + this.logger.info(`[ddp] checkAndReopen bucket=probe-pass elapsed=${elapsed}`) + return - } ++ } + this.logger.info(`[ddp] checkAndReopen bucket=probe-fail elapsed=${elapsed}`) + await this.forceReopen() } /** Clear connection and try to connect again. */ -@@ -549,7 +672,9 @@ export class DDPDriver extends EventEmitter implements ISocket, IDriver { +@@ -239,7 +369,8 @@ export class Socket extends EventEmitter { + + if (/^sub$/.test(obj.msg)) { + const { name, params } = obj; +- this.subscriptions[id] = { id, name, params, unsubscribe: this.unsubscribe.bind(this, id) }; ++ // Normalize eager entries with onEvent so a deduped app subscribe can attach a late listener. ++ this.subscriptions[id] = { id, name, params, unsubscribe: this.unsubscribe.bind(this, id), onEvent: this.onEvent.bind(this, name) }; + } + + try { +@@ -354,16 +485,43 @@ export class Socket extends EventEmitter { + * @param params Params sent to the subscription request + */ + subscribe = (name: string, params: any[], callback ?: ISocketMessageCallback, id?: string) => { ++ // Idempotency guard: an identical (name, params) already tracked (confirmed or in-flight) ++ // reuses that entry and attaches the late callback instead of sending a second `sub`. This ++ // collapses the app's re-subscribes against subscribeAll()'s replay after a reopen; distinct ++ // per-room params stay separate. subscribeAll() replays via _sendSub to skip this guard. ++ const key = subscriptionKey(name, params) ++ const existing = Object.keys(this.subscriptions) ++ .map(subId => this.subscriptions[subId]) ++ .find(sub => subscriptionKey(sub.name, sub.params) === key) ++ if (existing) { ++ if (callback && existing.onEvent) existing.onEvent(callback) ++ return Promise.resolve(existing) ++ } ++ const pending = this._pendingSubscriptions[key] ++ if (pending) { ++ if (callback) this.onEvent(name, callback) ++ return pending ++ } ++ return this._sendSub(name, params, callback, id) ++ } ++ ++ /** ++ * Send a `sub` frame and register/refresh its confirmed registry entry. No dedup — callers ++ * that must dedup (subscribe) guard before calling. The (name, params) claim is registered ++ * synchronously here, before send()'s first await, so a racing twin subscribe sees it. ++ */ ++ _sendSub = (name: string, params: any[], callback ?: ISocketMessageCallback, id?: string) => { + this.logger.info(`[ddp] Subscribe to ${name}, param: ${JSON.stringify(params)}`) +- return this.send({ msg: 'sub', id, name, params }) ++ const key = subscriptionKey(name, params) ++ const promise = this.send({ msg: 'sub', id, name, params }) + .then((result) => { +- const id = (result.subs) ? result.subs[0] : undefined +- if (id) { +- const unsubscribe = this.unsubscribe.bind(this, id) ++ const subId = (result.subs) ? result.subs[0] : undefined ++ if (subId) { ++ const unsubscribe = this.unsubscribe.bind(this, subId) + const onEvent = this.onEvent.bind(this, name) +- const subscription = { id, name, params, unsubscribe, onEvent } ++ const subscription = { id: subId, name, params, unsubscribe, onEvent } + if (callback) subscription.onEvent(callback) +- this.subscriptions[id] = subscription ++ this.subscriptions[subId] = subscription + return subscription + } + }) +@@ -371,13 +529,17 @@ export class Socket extends EventEmitter { + this.logger.error(`[ddp] Subscribe error: ${err.message}`) + // throw err + }) ++ this._pendingSubscriptions[key] = promise ++ const clear = () => { if (this._pendingSubscriptions[key] === promise) delete this._pendingSubscriptions[key] } ++ promise.then(clear, clear) ++ return promise + } + + /** Subscribe to all pre-configured streams (e.g. on login resume) */ + subscribeAll = () => { + const subscriptions = Object.keys(this.subscriptions || {}).map((key) => { + const { name, params, id } = this.subscriptions[key] +- return this.subscribe(name, params, undefined, id) ++ return this._sendSub(name, params, undefined, id) + }) + return Promise.all(subscriptions) + } +@@ -549,7 +711,9 @@ export class DDPDriver extends EventEmitter implements ISocket, IDriver { 'uiInteraction', 'e2ekeyRequest', 'userData',