fix(database): merge-commit restack onto #1600 and drop Mongo resume

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.
 

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<ReturnType<NonNullable<ConduitGrpcSdk['state']>['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();
   }

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(