Added MongoDB Consumer based on MongoDB ChangeStream subscription

src/packages/emmett-esdb/src/eventStore/consumers/eventStoreDBEventStoreConsumer.ts

@@ -217,8 +217,6 @@ export const eventStoreDBEventStoreConsumer = <
       return start;
     },
     stop,
-    close: async () => {
-      await stop();
-    },
+    close: stop,
   };
 };

src/packages/emmett-esdb/src/eventStore/consumers/subscriptions/index.ts

@@ -208,6 +208,10 @@ export const eventStoreDBSubscription = <
     return asyncRetry(
       () =>
         new Promise<void>((resolve, reject) => {
+          if (!isRunning) {
+            resolve();
+            return;
+          }
           console.info(
             `Starting subscription. ${retry++} retries. From: ${JSONParser.stringify(from ?? '$all')}, Start from: ${JSONParser.stringify(
               options.startFrom,

src/packages/emmett-mongodb/src/eventStore/consumers/CancellablePromise.ts

@@ -0,0 +1,63 @@
+import assert from 'assert';
+
+export class CancellationPromise<T = unknown> extends Promise<T> {
+  private _resolve: (value: T | PromiseLike<T>) => void;
+  private _reject: (reason?: unknown) => void;
+  private _state: 'resolved' | 'rejected' | 'pending' = 'pending';
+
+  constructor(
+    executor: (
+      resolve: (value: T | PromiseLike<T>) => void,
+      reject: (reason?: unknown) => void,
+    ) => void = () => null,
+  ) {
+    let _resolve: ((value: T | PromiseLike<T>) => void) | undefined = undefined;
+    let _reject: ((reason?: unknown) => void) | undefined = undefined;
+
+    super((resolve, reject) => {
+      executor(resolve, reject);
+      _resolve = resolve;
+      _reject = reject;
+    });
+
+    assert(_resolve);
+    assert(_reject);
+
+    this._resolve = _resolve;
+    this._reject = _reject;
+  }
+
+  reject(reason?: unknown): void {
+    this._state = 'rejected';
+    this._reject(reason);
+  }
+
+  resolve(value?: T): void {
+    this._state = 'resolved';
+    this._resolve(value as T);
+  }
+
+  get isResolved() {
+    return this._state === 'resolved';
+  }
+
+  get isRejected() {
+    return this._state === 'rejected';
+  }
+
+  get isPending() {
+    return this._state === 'pending';
+  }
+
+  static resolved<R = unknown>(value?: R) {
+    const promise = new CancellationPromise<R>();
+    promise.resolve(value as R);
+    return promise;
+  }
+
+  static rejected<R extends Error = Error>(value: R) {
+    const promise = new CancellationPromise<R>();
+    promise.reject(value);
+    return promise;
+  }
+}

src/packages/emmett-mongodb/src/eventStore/consumers/mongoDBCheckpointer.ts

@@ -0,0 +1,143 @@
+import {
+  type Message,
+  type ReadProcessorCheckpointResult,
+  getCheckpoint,
+} from '@event-driven-io/emmett';
+import type { MongoClient } from 'mongodb';
+import type { MongoDBCheckpointer } from './mongoDBProcessor';
+import { compareTwoTokens } from './subscriptions';
+import { DefaultProcessotCheckpointCollectionName, defaultTag } from './types';
+
+export const mongoDBCheckpointer = <
+  MessageType extends Message = Message,
+>(): MongoDBCheckpointer<MessageType> => ({
+  read: async (options, context) => {
+    const result = await readProcessorCheckpoint(context.client, options);
+
+    // eslint-disable-next-line @typescript-eslint/no-unsafe-assignment
+    return { lastCheckpoint: result?.lastCheckpoint };
+  },
+  store: async (options, context) => {
+    const newCheckpoint = getCheckpoint(options.message);
+
+    const result = await storeProcessorCheckpoint(context.client, {
+      lastStoredCheckpoint: options.lastCheckpoint,
+      newCheckpoint,
+      processorId: options.processorId,
+      partition: options.partition,
+      version: options.version || 0,
+    });
+
+    return result.success
+      ? { success: true, newCheckpoint: result.newCheckpoint }
+      : result;
+  },
+});
+
+// eslint-disable-next-line @typescript-eslint/no-explicit-any
+type ReadProcessorCheckpointMongoDBResult<Position = any> = {
+  lastProcessedCheckpoint: Position;
+  processorId: string;
+  partitionId: string;
+  version: number;
+};
+
+// eslint-disable-next-line @typescript-eslint/no-explicit-any
+export const readProcessorCheckpoint = async <CheckpointType = any>(
+  client: MongoClient,
+  options: {
+    processorId: string;
+    partition?: string;
+    collectionName?: string;
+    databaseName?: string;
+  },
+): Promise<ReadProcessorCheckpointResult<CheckpointType>> => {
+  const result = await client
+    .db(options.databaseName)
+    .collection<ReadProcessorCheckpointMongoDBResult<CheckpointType>>(
+      options.collectionName || DefaultProcessotCheckpointCollectionName,
+    )
+    .findOne({
+      processorId: options.processorId,
+      partitionId: options.partition || defaultTag,
+    });
+
+  return {
+    lastCheckpoint: result !== null ? result.lastProcessedCheckpoint : null,
+  };
+};
+
+type StoreLastProcessedProcessorPositionResult<Position = unknown> =
+  | {
+      success: true;
+      newCheckpoint: Position;
+    }
+  | { success: false; reason: 'IGNORED' | 'MISMATCH' };
+
+export const storeProcessorCheckpoint = async <Position>(
+  client: MongoClient,
+  {
+    processorId,
+    version,
+    newCheckpoint,
+    lastStoredCheckpoint,
+    partition,
+    collectionName,
+    dbName,
+  }: {
+    processorId: string;
+    version: number;
+    newCheckpoint: Position | null;
+    lastStoredCheckpoint: Position | null;
+    partition?: string;
+    collectionName?: string;
+    dbName?: string;
+  },
+): Promise<
+  StoreLastProcessedProcessorPositionResult<
+    null extends Position ? Position | null : Position
+  >
+> => {
+  const checkpoints = client
+    .db(dbName)
+    .collection<ReadProcessorCheckpointMongoDBResult>(
+      collectionName || DefaultProcessotCheckpointCollectionName,
+    );
+
+  const filter = {
+    processorId: processorId,
+    partitionId: partition || defaultTag,
+  };
+
+  const current = await checkpoints.findOne(filter);
+
+  // MISMATCH: we have a checkpoint but lastProcessedCheckpoint doesn’t match
+  if (
+    current &&
+    compareTwoTokens(current.lastProcessedCheckpoint, lastStoredCheckpoint) !==
+      0
+  ) {
+    return { success: false, reason: 'MISMATCH' };
+  }
+
+  // IGNORED: same or earlier position
+  if (current?.lastProcessedCheckpoint && newCheckpoint) {
+    if (
+      compareTwoTokens(current.lastProcessedCheckpoint, newCheckpoint) !== -1
+    ) {
+      return { success: false, reason: 'IGNORED' };
+    }
+  }
+
+  const updateResult = await checkpoints.updateOne(
+    { ...filter, lastProcessedCheckpoint: lastStoredCheckpoint },
+    { $set: { lastProcessedCheckpoint: newCheckpoint, version } },
+    { upsert: true },
+  );
+
+  if (updateResult.matchedCount > 0 || updateResult.upsertedCount > 0) {
+    return { success: true, newCheckpoint: newCheckpoint! };
+  }
+
+  return { success: false, reason: 'MISMATCH' };
+};