diff --git a/modules/database/README.mdx b/modules/database/README.mdx index 10703937b..f573efb1e 100644 --- a/modules/database/README.mdx +++ b/modules/database/README.mdx @@ -60,7 +60,7 @@ subscribe({ schema: 'Order', documentId?: string }) unsubscribe({ schema: 'Order', documentId?: string }) ``` -Events arrive as `change` with `{ version, operation, schema, documentId, occurredAt }` and contain no document fields and no resume token. This is **live-tail, not backfill**. After a full reconnect, subscribe again and refetch over authorized REST. +Events arrive as `change` with `{ version, operation, schema, documentId, occurredAt }` and contain no document fields and no resume token. This is **live-tail, not backfill**: the change stream starts at the end of the oplog. A leader restart or cursor drop does not replay missed events; clients subscribe again and refetch over authorized REST. The leader also publishes `database:change:${schema}` on the Redis bus. **Do not also relay `database:change:*` on `/events/` if the same client is on `/database/`** — that duplicates notifications. Keep the bus for other modules; just do not dual-subscribe. diff --git a/modules/database/src/realtime/MongoChangeStreamCoordinator.ts b/modules/database/src/realtime/MongoChangeStreamCoordinator.ts index 45644fa78..b6a8349e2 100644 --- a/modules/database/src/realtime/MongoChangeStreamCoordinator.ts +++ b/modules/database/src/realtime/MongoChangeStreamCoordinator.ts @@ -1,16 +1,7 @@ import { ConduitGrpcSdk } from '@conduitplatform/grpc-sdk'; -import { - normalizeChangeEvent, - parseResumeToken, - serializeResumeToken, - type RawChangeEvent, -} from './normalize.js'; +import { normalizeChangeEvent, type RawChangeEvent } from './normalize.js'; import { authorizedDocumentRoom, roomsForPublicChange } from './rooms.js'; -import { - isResumeTokenUnusable, - topologyFromHello, - type TopologyResult, -} from './topology.js'; +import { topologyFromHello, type TopologyResult } from './topology.js'; import type { ChangeStreamLike, DatabaseChangeEvent, @@ -28,7 +19,6 @@ import { } from './watchPipeline.js'; const LEADER_LOCK = 'realtime:change-stream:leader'; -const RESUME_TOKEN_KEY = 'realtime:resumeToken'; const LOCK_TTL_MS = 15_000; const LOCK_RENEW_MS = 5_000; const RETRY_BASE_MS = 1_000; @@ -38,10 +28,7 @@ type LeaderLock = NonNullable< Awaited['tryAcquireLock']>> >; -export type WatchFactory = (options: { - resumeAfter?: unknown; - pipeline: WatchPipeline; -}) => ChangeStreamLike; +export type WatchFactory = (pipeline: WatchPipeline) => ChangeStreamLike; export type CoordinatorOptions = { grpcSdk: ConduitGrpcSdk; @@ -232,21 +219,10 @@ export class MongoChangeStreamCoordinator { this.ignoreClose = false; const generation = this.lockGeneration; try { - const resumeAfter = parseResumeToken( - await this.options.grpcSdk.state!.getKey(RESUME_TOKEN_KEY), - ); - if ( - this.watching || - this.closed || - !this.lock || - generation !== this.lockGeneration - ) { - return; - } const collections = this.collectionNames(); const pipeline = buildWatchPipeline(collections); this.watchedCollectionsKey = optedInCollectionsKey(collections); - const stream = this.options.watch({ resumeAfter, pipeline }); + const stream = this.options.watch(pipeline); if (generation !== this.lockGeneration || this.closed) { try { await stream.close(); @@ -299,13 +275,9 @@ export class MongoChangeStreamCoordinator { private async handleChange(change: RawChangeEvent, generation: number) { if (generation !== this.lockGeneration) return; - const token = serializeResumeToken(change._id); const schema = this.resolveSchema(change.ns?.coll); const event = schema ? normalizeChangeEvent(change, schema.name) : null; if (!event || !schema) { - if (token && generation === this.lockGeneration) { - await this.persistResumeToken(token); - } if (change.operationType && WATCH_RESTART_OPERATIONS.has(change.operationType)) { await this.stopStream('starting'); this.scheduleRetry(); @@ -315,13 +287,6 @@ export class MongoChangeStreamCoordinator { this.lastEventAt = event.occurredAt; this.lastError = undefined; await this.emitChange(schema, event); - if (token) { - await this.persistResumeToken(token); - } - } - - private async persistResumeToken(token: string) { - await this.options.grpcSdk.state!.setKey(RESUME_TOKEN_KEY, token); } private async emitChange(schema: OptedInSchema, event: DatabaseChangeEvent) { @@ -407,9 +372,6 @@ export class MongoChangeStreamCoordinator { this.streamState = 'degraded'; ConduitGrpcSdk.Metrics?.increment('database_realtime_stream_errors_total'); ConduitGrpcSdk.Logger.error(err as Error); - if (isResumeTokenUnusable(err)) { - await this.options.grpcSdk.state!.clearKey(RESUME_TOKEN_KEY); - } await this.stopStream('degraded'); this.scheduleRetry(); } diff --git a/modules/database/src/realtime/RealtimeService.ts b/modules/database/src/realtime/RealtimeService.ts index 96d21f485..90369690e 100644 --- a/modules/database/src/realtime/RealtimeService.ts +++ b/modules/database/src/realtime/RealtimeService.ts @@ -41,7 +41,7 @@ export class RealtimeService { if (adapter instanceof MongooseAdapter) { this.coordinator = new MongoChangeStreamCoordinator({ grpcSdk, - watch: options => this.openWatch(adapter, options), + watch: pipeline => this.openWatch(adapter, pipeline), hello: () => this.hello(adapter), getOptedInSchemas: () => this.getOptedInSchemas(), subscriptions: this.subscriptions, @@ -131,16 +131,13 @@ export class RealtimeService { private openWatch( adapter: MongooseAdapter, - options: { resumeAfter?: unknown; pipeline: WatchPipeline }, + pipeline: WatchPipeline, ): ChangeStreamLike { const db = adapter.mongoose.connection.db; if (!db) { throw new Error('MongoDB connection is not ready'); } - return db.watch( - options.pipeline, - options.resumeAfter ? { resumeAfter: options.resumeAfter as never } : {}, - ) as unknown as ChangeStreamLike; + return db.watch(pipeline) as unknown as ChangeStreamLike; } private async hello( diff --git a/modules/database/src/realtime/__tests__/coordinator.test.ts b/modules/database/src/realtime/__tests__/coordinator.test.ts index 3720d5a41..c789ea94d 100644 --- a/modules/database/src/realtime/__tests__/coordinator.test.ts +++ b/modules/database/src/realtime/__tests__/coordinator.test.ts @@ -1,6 +1,6 @@ import { EventEmitter } from 'node:events'; import { afterEach, describe, expect, it, jest } from '@jest/globals'; -import { EJSON, ObjectId } from 'bson'; +import { ObjectId } from 'bson'; import { MongoChangeStreamCoordinator } from '../MongoChangeStreamCoordinator.js'; import { RealtimeSubscriptionTracker } from '../subscriptions.js'; import { roomsForPublicChange } from '../rooms.js'; @@ -50,13 +50,11 @@ function createCoordinator(overrides?: { authorizationEnabled: boolean; cmsReadEnabled?: boolean; }[]; - getKeyDelayMs?: number; }) { const stream = new EventEmitter() as EventEmitter & { close: () => Promise }; stream.close = async () => { stream.emit('close'); }; - const state = new Map(); const lock = { extend: jest.fn(async () => lock), release: jest.fn(async () => undefined), @@ -70,18 +68,6 @@ function createCoordinator(overrides?: { state: { tryAcquireLock: jest.fn(async () => lock), releaseLock: jest.fn(async () => undefined), - getKey: jest.fn(async (key: string) => { - if (overrides?.getKeyDelayMs) { - await new Promise(resolve => setTimeout(resolve, overrides.getKeyDelayMs)); - } - return state.get(key) ?? null; - }), - setKey: jest.fn(async (key: string, value: string) => { - state.set(key, value); - }), - clearKey: jest.fn(async (key: string) => { - state.delete(key); - }), }, bus: { publish }, router: { socketPush: routerPush }, @@ -120,36 +106,40 @@ function createCoordinator(overrides?: { publish, subscriptions, grpcSdk, - state, lock, }; } -function insertChange(collection: string, id: string, token: unknown) { +function insertChange(collection: string, id: string) { return { operationType: 'insert', ns: { coll: collection }, documentKey: { _id: new ObjectId(id) }, - _id: token, }; } +function expectWatchFromNow( + watch: { mock: { calls: unknown[][] } }, + callIndex = 0, +) { + expect(watch.mock.calls[callIndex]).toHaveLength(1); + expect(watch.mock.calls[callIndex][0]).not.toHaveProperty('resumeAfter'); +} + describe('MongoChangeStreamCoordinator', () => { afterEach(() => { jest.useRealTimers(); }); it('emits one normalized event to public rooms and ignores other collections', async () => { - const { coordinator, stream, routerPush, adminPush, publish, state } = - createCoordinator(); + const { coordinator, stream, routerPush, adminPush, publish } = createCoordinator(); await coordinator.reconcile(); - const resume = { _data: 'token' }; stream.emit('change', { - ...insertChange('orders', '64b64c4c4c4c4c4c4c4c4c4c', resume), + ...insertChange('orders', '64b64c4c4c4c4c4c4c4c4c4c'), fullDocument: { secret: 'nope' }, wallTime: new Date('2026-01-02T00:00:00.000Z'), }); - stream.emit('change', insertChange('other', '64b64c4c4c4c4c4c4c4c4c4d', resume)); + stream.emit('change', insertChange('other', '64b64c4c4c4c4c4c4c4c4c4d')); await coordinator.waitForIdle(); expect(publish).toHaveBeenCalledTimes(1); expect(publish.mock.calls[0][0]).toBe('database:change:Order'); @@ -172,23 +162,11 @@ describe('MongoChangeStreamCoordinator', () => { expect( JSON.parse((adminPush.mock.calls[0][0] as { data: string }).data), ).not.toHaveProperty('resumeToken'); - expect(state.get('realtime:resumeToken')).toBe(EJSON.stringify(resume)); - await coordinator.shutdown(); - }); - - it('advances the resume token for filtered events', async () => { - const { coordinator, stream, publish, state } = createCoordinator(); - await coordinator.reconcile(); - const skip = { _data: 'skip-token' }; - stream.emit('change', insertChange('other', '64b64c4c4c4c4c4c4c4c4c4d', skip)); - await coordinator.waitForIdle(); - expect(publish).not.toHaveBeenCalled(); - expect(state.get('realtime:resumeToken')).toBe(EJSON.stringify(skip)); await coordinator.shutdown(); }); - it('serializes overlapping handlers and persists after emit', async () => { - const { coordinator, stream, state, adminPush } = createCoordinator(); + it('serializes overlapping handlers', async () => { + const { coordinator, stream, adminPush } = createCoordinator(); let release!: () => void; const gate = new Promise(resolve => { release = resolve; @@ -201,27 +179,23 @@ describe('MongoChangeStreamCoordinator', () => { } }); await coordinator.reconcile(); - const tokenA = { _data: 'token-a' }; - const tokenB = { _data: 'token-b' }; - stream.emit('change', insertChange('orders', '64b64c4c4c4c4c4c4c4c4c4c', tokenA)); - stream.emit('change', insertChange('orders', '64b64c4c4c4c4c4c4c4c4c4d', tokenB)); + stream.emit('change', insertChange('orders', '64b64c4c4c4c4c4c4c4c4c4c')); + stream.emit('change', insertChange('orders', '64b64c4c4c4c4c4c4c4c4c4d')); await Promise.resolve(); await new Promise(resolve => setImmediate(resolve)); - expect(state.get('realtime:resumeToken')).toBeUndefined(); expect(adminPush).toHaveBeenCalledTimes(1); release(); await coordinator.waitForIdle(); expect(adminPush).toHaveBeenCalledTimes(2); - expect(state.get('realtime:resumeToken')).toBe(EJSON.stringify(tokenB)); await coordinator.shutdown(); }); - it('watches opted-in collections with $match and $project', async () => { + it('watches opted-in collections with $match and $project from now', async () => { const { coordinator, watch } = createCoordinator(); await coordinator.reconcile(); expect(watch).toHaveBeenCalledTimes(1); - const pipeline = (watch.mock.calls[0][0] as { pipeline: Record[] }) - .pipeline; + const pipeline = watch.mock.calls[0][0] as Record[]; + expectWatchFromNow(watch); expect(pipeline[0]).toEqual( expect.objectContaining({ $match: expect.objectContaining({ @@ -257,14 +231,14 @@ describe('MongoChangeStreamCoordinator', () => { }); await coordinator.reconcile(); expect(watch).toHaveBeenCalledTimes(2); - const pipeline = (watch.mock.calls[1][0] as { pipeline: Record[] }) - .pipeline; + const pipeline = watch.mock.calls[1][0] as Record[]; const match = pipeline[0] as { $match: { $or: Array<{ 'ns.coll'?: { $in: string[] } }> }; }; expect(match.$match.$or[0]['ns.coll']?.$in).toEqual( expect.arrayContaining(['orders', 'items']), ); + expectWatchFromNow(watch, 1); await coordinator.shutdown(); }); @@ -280,10 +254,7 @@ describe('MongoChangeStreamCoordinator', () => { 'user-1', ); await coordinator.reconcile(); - stream.emit( - 'change', - insertChange('orders', '64b64c4c4c4c4c4c4c4c4c4c', { _data: 'token' }), - ); + stream.emit('change', insertChange('orders', '64b64c4c4c4c4c4c4c4c4c4c')); await coordinator.waitForIdle(); expect(routerPush).not.toHaveBeenCalled(); expect(await subscriptions.listUsers('Order', '64b64c4c4c4c4c4c4c4c4c4c')).toEqual( @@ -305,39 +276,16 @@ describe('MongoChangeStreamCoordinator', () => { }); it('opens a single watch when reconcile runs concurrently', async () => { - const { coordinator, watch } = createCoordinator({ getKeyDelayMs: 40 }); + const { coordinator, watch } = createCoordinator(); await Promise.all([coordinator.reconcile(), coordinator.reconcile()]); expect(watch).toHaveBeenCalledTimes(1); await coordinator.shutdown(); }); - it('clears an unusable resume token on 280 and retries', async () => { - const { coordinator, stream, grpcSdk } = createCoordinator(); - await coordinator.reconcile(); - stream.emit('error', { code: 280, message: 'ChangeStreamHistoryLost' }); - await new Promise(resolve => setImmediate(resolve)); - expect(grpcSdk.state.clearKey).toHaveBeenCalled(); - await coordinator.shutdown(); - }); - - it('keeps the resume token on CursorKilled 237', async () => { - const { coordinator, stream, grpcSdk, state } = createCoordinator(); - await coordinator.reconcile(); - const token = { _data: 'keep-me' }; - stream.emit('change', insertChange('orders', '64b64c4c4c4c4c4c4c4c4c4c', token)); - await coordinator.waitForIdle(); - expect(state.get('realtime:resumeToken')).toBe(EJSON.stringify(token)); - stream.emit('error', { code: 237, message: 'CursorKilled' }); - await new Promise(resolve => setImmediate(resolve)); - expect(grpcSdk.state.clearKey).not.toHaveBeenCalled(); - expect(state.get('realtime:resumeToken')).toBe(EJSON.stringify(token)); - await coordinator.shutdown(); - }); - - it('does not persist a later token when emit fails and reopens from the last good token', async () => { + it('stops the stream and retries from now when emit fails', async () => { jest.useFakeTimers(); const streams: Array Promise }> = []; - const { coordinator, adminPush, state, watch } = createCoordinator(); + const { coordinator, adminPush, watch } = createCoordinator(); watch.mockImplementation(() => { const next = new EventEmitter() as EventEmitter & { close: () => Promise }; next.close = async () => { @@ -351,39 +299,28 @@ describe('MongoChangeStreamCoordinator', () => { .mockRejectedValueOnce(new Error('push failed')) .mockResolvedValue(undefined); await coordinator.reconcile(); - const tokenGood = { _data: 'token-good' }; - const tokenA = { _data: 'token-a' }; - const tokenB = { _data: 'token-b' }; - streams[0].emit( - 'change', - insertChange('orders', '64b64c4c4c4c4c4c4c4c4c4b', tokenGood), - ); + streams[0].emit('change', insertChange('orders', '64b64c4c4c4c4c4c4c4c4c4b')); await coordinator.waitForIdle(); - expect(state.get('realtime:resumeToken')).toBe(EJSON.stringify(tokenGood)); - streams[0].emit('change', insertChange('orders', '64b64c4c4c4c4c4c4c4c4c4c', tokenA)); - streams[0].emit('change', insertChange('orders', '64b64c4c4c4c4c4c4c4c4c4d', tokenB)); + streams[0].emit('change', insertChange('orders', '64b64c4c4c4c4c4c4c4c4c4c')); + streams[0].emit('change', insertChange('orders', '64b64c4c4c4c4c4c4c4c4c4d')); await coordinator.waitForIdle(); - expect(state.get('realtime:resumeToken')).toBe(EJSON.stringify(tokenGood)); expect(adminPush).toHaveBeenCalledTimes(2); await jest.advanceTimersByTimeAsync(1_000); expect(watch).toHaveBeenCalledTimes(2); - expect((watch.mock.calls[1][0] as { resumeAfter?: unknown }).resumeAfter).toEqual( - tokenGood, - ); - streams[1].emit('change', insertChange('orders', '64b64c4c4c4c4c4c4c4c4c4c', tokenA)); + expectWatchFromNow(watch, 1); + streams[1].emit('change', insertChange('orders', '64b64c4c4c4c4c4c4c4c4c4c')); await coordinator.waitForIdle(); - expect(state.get('realtime:resumeToken')).toBe(EJSON.stringify(tokenA)); expect(adminPush).toHaveBeenCalledTimes(3); await coordinator.shutdown(); jest.useRealTimers(); }); it.each(['drop', 'rename', 'invalidate', 'dropDatabase'] as const)( - 'reopens the watch on %s', + 'reopens the watch on %s from now', async operationType => { jest.useFakeTimers(); const streams: Array Promise }> = []; - const { coordinator, watch, state } = createCoordinator(); + const { coordinator, watch } = createCoordinator(); watch.mockImplementation(() => { const next = new EventEmitter() as EventEmitter & { close: () => Promise }; next.close = async () => { @@ -393,19 +330,14 @@ describe('MongoChangeStreamCoordinator', () => { return next as never; }); await coordinator.reconcile(); - const token = { _data: `${operationType}-token` }; streams[0].emit('change', { operationType, ns: { coll: 'orders' }, - _id: token, }); await coordinator.waitForIdle(); - expect(state.get('realtime:resumeToken')).toBe(EJSON.stringify(token)); await jest.advanceTimersByTimeAsync(1_000); expect(watch).toHaveBeenCalledTimes(2); - expect((watch.mock.calls[1][0] as { resumeAfter?: unknown }).resumeAfter).toEqual( - token, - ); + expectWatchFromNow(watch, 1); await coordinator.shutdown(); jest.useRealTimers(); }, @@ -427,10 +359,7 @@ describe('MongoChangeStreamCoordinator', () => { 'user-1', ]); await coordinator.reconcile(); - stream.emit( - 'change', - insertChange('orders', '64b64c4c4c4c4c4c4c4c4c4c', { _data: 'token' }), - ); + stream.emit('change', insertChange('orders', '64b64c4c4c4c4c4c4c4c4c4c')); await coordinator.waitForIdle(); expect(routerPush).not.toHaveBeenCalled(); expect(removeUser).not.toHaveBeenCalled(); @@ -451,19 +380,15 @@ describe('MongoChangeStreamCoordinator', () => { it('ignores draining watch events after lock renew failure', async () => { jest.useFakeTimers(); - const { coordinator, stream, lock, adminPush, state } = createCoordinator(); + const { coordinator, stream, lock, adminPush } = createCoordinator(); lock.extend.mockResolvedValueOnce(lock).mockRejectedValueOnce(new Error('lost lock')); await coordinator.reconcile(); expect(coordinator.getState()).toBe('live'); await jest.advanceTimersByTimeAsync(5_000); expect(coordinator.getState()).toBe('idle'); - stream.emit( - 'change', - insertChange('orders', '64b64c4c4c4c4c4c4c4c4c4c', { _data: 'stale' }), - ); + stream.emit('change', insertChange('orders', '64b64c4c4c4c4c4c4c4c4c4c')); await coordinator.waitForIdle(); expect(adminPush).not.toHaveBeenCalled(); - expect(state.get('realtime:resumeToken')).toBeUndefined(); await coordinator.shutdown(); jest.useRealTimers(); }); @@ -489,10 +414,7 @@ describe('MongoChangeStreamCoordinator', () => { 'user-1', ); await coordinator.reconcile(); - stream.emit( - 'change', - insertChange('orders', '64b64c4c4c4c4c4c4c4c4c4c', { _data: 'token' }), - ); + stream.emit('change', insertChange('orders', '64b64c4c4c4c4c4c4c4c4c4c')); await coordinator.waitForIdle(); expect(adminPush).toHaveBeenCalledTimes(1); expect(routerPush).not.toHaveBeenCalled(); diff --git a/modules/database/src/realtime/__tests__/normalize.test.ts b/modules/database/src/realtime/__tests__/normalize.test.ts index f7f933d33..223009676 100644 --- a/modules/database/src/realtime/__tests__/normalize.test.ts +++ b/modules/database/src/realtime/__tests__/normalize.test.ts @@ -1,21 +1,15 @@ import { describe, expect, it } from '@jest/globals'; import { ObjectId } from 'bson'; -import { - normalizeChangeEvent, - parseResumeToken, - serializeResumeToken, -} from '../normalize.js'; +import { normalizeChangeEvent } from '../normalize.js'; describe('normalizeChangeEvent', () => { it('normalizes insert/update/replace/delete into metadata-only events', () => { - const resume = { _data: 'token-1' }; const event = normalizeChangeEvent( { operationType: 'insert', documentKey: { _id: new ObjectId('64b64c4c4c4c4c4c4c4c4c4c') }, fullDocument: { secret: 'nope' }, wallTime: new Date('2026-01-01T00:00:00.000Z'), - _id: resume, }, 'Order', ); @@ -26,17 +20,12 @@ describe('normalizeChangeEvent', () => { documentId: '64b64c4c4c4c4c4c4c4c4c4c', occurredAt: '2026-01-01T00:00:00.000Z', }); - expect(parseResumeToken(serializeResumeToken(resume))).toEqual(resume); expect(event).not.toHaveProperty('resumeToken'); expect(JSON.parse(JSON.stringify(event))).not.toHaveProperty('fullDocument'); }); it('ignores drop/invalidate and missing document ids', () => { - expect( - normalizeChangeEvent({ operationType: 'drop', _id: { _data: 'x' } }, 'Order'), - ).toBeNull(); - expect( - normalizeChangeEvent({ operationType: 'insert', _id: { _data: 'x' } }, 'Order'), - ).toBeNull(); + expect(normalizeChangeEvent({ operationType: 'drop' }, 'Order')).toBeNull(); + expect(normalizeChangeEvent({ operationType: 'insert' }, 'Order')).toBeNull(); }); }); diff --git a/modules/database/src/realtime/__tests__/topology.test.ts b/modules/database/src/realtime/__tests__/topology.test.ts index 79afac87c..51b52be54 100644 --- a/modules/database/src/realtime/__tests__/topology.test.ts +++ b/modules/database/src/realtime/__tests__/topology.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from '@jest/globals'; -import { isResumeTokenUnusable, topologyFromHello } from '../topology.js'; +import { topologyFromHello } from '../topology.js'; describe('topology helpers', () => { it('accepts replica sets and mongos, rejects standalone', () => { @@ -7,13 +7,4 @@ describe('topology helpers', () => { expect(topologyFromHello({ msg: 'isdbgrid' })).toEqual({ supported: true }); expect(topologyFromHello({}).supported).toBe(false); }); - - it('detects unusable resume tokens', () => { - expect(isResumeTokenUnusable({ code: 280 })).toBe(true); - expect(isResumeTokenUnusable({ code: 237 })).toBe(false); - expect(isResumeTokenUnusable({ code: 136 })).toBe(false); - expect(isResumeTokenUnusable(new Error('ChangeStreamHistoryLost'))).toBe(true); - expect(isResumeTokenUnusable(new Error('cannot resume'))).toBe(false); - expect(isResumeTokenUnusable(new Error('socket hang up'))).toBe(false); - }); }); diff --git a/modules/database/src/realtime/normalize.ts b/modules/database/src/realtime/normalize.ts index b2f4b8ff3..b5055c3d2 100644 --- a/modules/database/src/realtime/normalize.ts +++ b/modules/database/src/realtime/normalize.ts @@ -1,4 +1,3 @@ -import { EJSON } from 'bson'; import { DATABASE_CHANGE_EVENT_VERSION, DATABASE_CHANGE_OPERATIONS, @@ -13,8 +12,6 @@ export type RawChangeEvent = { ns?: { coll?: string }; documentKey?: { _id?: unknown }; wallTime?: Date; - clusterTime?: { toString?: () => string }; - _id?: unknown; }; export function normalizeChangeEvent( @@ -30,9 +27,6 @@ export function normalizeChangeEvent( if (!documentId) { return null; } - if (!change._id) { - return null; - } return { version: DATABASE_CHANGE_EVENT_VERSION, operation: operation as DatabaseChangeOperation, @@ -45,24 +39,6 @@ export function normalizeChangeEvent( }; } -export function serializeResumeToken(id: unknown): string | null { - if (id === undefined || id === null) return null; - try { - return EJSON.stringify(id); - } catch { - return null; - } -} - -export function parseResumeToken(token: string | null | undefined): unknown | undefined { - if (!token) return undefined; - try { - return EJSON.parse(token); - } catch { - return undefined; - } -} - function extractDocumentId(id: unknown): string | null { if (id === undefined || id === null) return null; if (typeof id === 'string' || typeof id === 'number') return String(id); diff --git a/modules/database/src/realtime/topology.ts b/modules/database/src/realtime/topology.ts index 4d62fa3fa..a1ab72d71 100644 --- a/modules/database/src/realtime/topology.ts +++ b/modules/database/src/realtime/topology.ts @@ -1,8 +1,3 @@ -const UNUSABLE_RESUME_TOKEN_CODES = new Set([ - 280, // ChangeStreamHistoryLost - 286, // ChangeStreamFatalError -]); - export type TopologyResult = { supported: boolean; message?: string; @@ -28,23 +23,3 @@ export function topologyFromHello( message: 'A replica set or sharded MongoDB deployment is required for live updates', }; } - -export function isResumeTokenUnusable(error: unknown): boolean { - const code = extractErrorCode(error); - if (code === 237 || code === 136) { - return false; - } - if (code !== undefined && UNUSABLE_RESUME_TOKEN_CODES.has(code)) { - return true; - } - const message = error instanceof Error ? error.message : String(error ?? ''); - return /ChangeStreamHistoryLost/i.test(message); -} - -function extractErrorCode(error: unknown): number | undefined { - if (!error || typeof error !== 'object') return undefined; - const candidate = error as { code?: unknown; errorCode?: unknown }; - if (typeof candidate.code === 'number') return candidate.code; - if (typeof candidate.errorCode === 'number') return candidate.errorCode; - return undefined; -}