Made MongoDB checkpointers to be aligned with others

Passed explicit checkpoint and used string instead of the nested
structure

Author: oskardudycz

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

@@ -9,11 +9,10 @@ import {
   type Message,
   type MessageConsumer,
   type RecordedMessage,
+  type RecordedMessageMetadataWithGlobalPosition,
 } from '@event-driven-io/emmett';
 import { MongoClient, type MongoClientOptions } from 'mongodb';
 import { v4 as uuid } from 'uuid';
-import type { MongoDBRecordedMessageMetadata } from '../event';
-import type { MongoDBReadEventMetadata } from '../mongoDBEventStore';
 import { CancellationPromise } from './CancellablePromise';
 import {
   changeStreamReactor,
@@ -27,12 +26,15 @@ import {
   zipMongoDBMessageBatchPullerStartFrom,
   type MongoDBSubscription,
 } from './subscriptions';
-import type { MongoDBResumeToken } from './subscriptions/types';
+import type { MongoDBResumeToken } from './subscriptions/mongoDbResumeToken';
+
+export type MongoDBChangeStreamMessageMetadata =
+  RecordedMessageMetadataWithGlobalPosition<MongoDBResumeToken['_data']>;
 
 export type MessageConsumerOptions<
   MessageType extends Message = AnyMessage,
   MessageMetadataType extends
-    MongoDBReadEventMetadata = MongoDBRecordedMessageMetadata,
+    MongoDBChangeStreamMessageMetadata = MongoDBChangeStreamMessageMetadata,
   HandlerContext extends DefaultRecord | undefined = undefined,
   CheckpointType = GlobalPositionTypeOfRecordedMessageMetadata<MessageMetadataType>,
 > = {
@@ -50,7 +52,7 @@ export type MongoDBEventStoreConsumerConfig<
   // eslint-disable-next-line @typescript-eslint/no-explicit-any
   ConsumerMessageType extends Message = any,
   MessageMetadataType extends
-    MongoDBReadEventMetadata = MongoDBRecordedMessageMetadata,
+    MongoDBChangeStreamMessageMetadata = MongoDBChangeStreamMessageMetadata,
   HandlerContext extends DefaultRecord | undefined = undefined,
   CheckpointType = GlobalPositionTypeOfRecordedMessageMetadata<MessageMetadataType>,
 > = MessageConsumerOptions<
@@ -67,7 +69,7 @@ export type MongoDBEventStoreConsumerConfig<
 export type MongoDBConsumerOptions<
   ConsumerEventType extends Message = Message,
   MessageMetadataType extends
-    MongoDBReadEventMetadata = MongoDBRecordedMessageMetadata,
+    MongoDBChangeStreamMessageMetadata = MongoDBChangeStreamMessageMetadata,
   HandlerContext extends DefaultRecord | undefined = undefined,
   CheckpointType = GlobalPositionTypeOfRecordedMessageMetadata<MessageMetadataType>,
 > = MongoDBEventStoreConsumerConfig<
@@ -130,7 +132,7 @@ export type MongoDBConsumerHandlerContext = {
 export const mongoDBEventStoreConsumer = <
   ConsumerMessageType extends Message = AnyMessage,
   MessageMetadataType extends
-    MongoDBReadEventMetadata = MongoDBRecordedMessageMetadata,
+    MongoDBChangeStreamMessageMetadata = MongoDBChangeStreamMessageMetadata,
   HandlerContext extends
     MongoDBConsumerHandlerContext = MongoDBConsumerHandlerContext,
   CheckpointType = MongoDBResumeToken,

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

@@ -9,16 +9,15 @@ import {
   MessageProcessor,
   type ProjectorOptions,
   type ReactorOptions,
-  type RecordedMessage,
+  getCheckpoint,
   projector,
   reactor,
 } from '@event-driven-io/emmett';
 import { MongoClient } from 'mongodb';
-import type { ReadEventMetadataWithGlobalPosition } from '../event';
 import type { MongoDBEventStoreConnectionOptions } from '../mongoDBEventStore';
+import type { MongoDBChangeStreamMessageMetadata } from './mongoDBEventsConsumer';
 import { readProcessorCheckpoint } from './readProcessorCheckpoint';
 import { storeProcessorCheckpoint } from './storeProcessorCheckpoint';
-import type { MongoDBResumeToken } from './subscriptions/types';
 
 type MongoDBConnectionOptions = {
   connectionOptions: MongoDBEventStoreConnectionOptions;
@@ -31,98 +30,51 @@ export type MongoDBProcessorHandlerContext = {
 export type MongoDBProcessor<MessageType extends Message = AnyMessage> =
   MessageProcessor<
     MessageType,
-    ReadEventMetadataWithGlobalPosition,
+    MongoDBChangeStreamMessageMetadata,
     MongoDBProcessorHandlerContext
   >;
 
 export type MongoDBProcessorOptions<MessageType extends Message = Message> =
   ReactorOptions<
     MessageType,
-    ReadEventMetadataWithGlobalPosition,
+    MongoDBChangeStreamMessageMetadata,
     MongoDBProcessorHandlerContext
   > & { connectionOptions: MongoDBEventStoreConnectionOptions };
 
-export type MongoDBCheckpointer<
-  MessageType extends AnyMessage = AnyMessage,
-  // eslint-disable-next-line @typescript-eslint/no-explicit-any
-  CheckpointType = any,
-> = Checkpointer<
-  MessageType,
-  ReadEventMetadataWithGlobalPosition<CheckpointType>,
-  MongoDBProcessorHandlerContext,
-  CheckpointType
->;
+export type MongoDBCheckpointer<MessageType extends AnyMessage = AnyMessage> =
+  Checkpointer<
+    MessageType,
+    MongoDBChangeStreamMessageMetadata,
+    MongoDBProcessorHandlerContext
+  >;
 
 export type MongoDBProjectorOptions<EventType extends AnyEvent = AnyEvent> =
   ProjectorOptions<
     EventType,
-    ReadEventMetadataWithGlobalPosition,
+    MongoDBChangeStreamMessageMetadata,
     MongoDBProcessorHandlerContext
   > &
     MongoDBConnectionOptions;
 
-// eslint-disable-next-line @typescript-eslint/no-explicit-any
-const isResumeToken = (value: any): value is MongoDBResumeToken =>
-  typeof value === 'object' &&
-  '_data' in value &&
-  // eslint-disable-next-line @typescript-eslint/no-unsafe-member-access
-  typeof value._data === 'string';
-
-export const getCheckpoint = <
-  MessageType extends AnyMessage = AnyMessage,
-  CheckpointType = MongoDBCheckpointer,
-  MessageMetadataType extends
-    ReadEventMetadataWithGlobalPosition<CheckpointType> = ReadEventMetadataWithGlobalPosition<CheckpointType>,
->(
-  message: RecordedMessage<MessageType, MessageMetadataType>,
-): CheckpointType | null => {
-  // eslint-disable-next-line @typescript-eslint/no-unsafe-return
-  return 'checkpoint' in message.metadata &&
-    // eslint-disable-next-line @typescript-eslint/no-unsafe-member-access
-    isResumeToken(message.metadata.checkpoint)
-    ? // eslint-disable-next-line @typescript-eslint/no-unsafe-member-access
-      message.metadata.checkpoint
-    : 'globalPosition' in message.metadata &&
-        // eslint-disable-next-line @typescript-eslint/no-unsafe-member-access
-        isResumeToken(message.metadata.globalPosition)
-      ? // eslint-disable-next-line @typescript-eslint/no-unsafe-member-access
-        message.metadata.globalPosition
-      : 'streamPosition' in message.metadata &&
-          // eslint-disable-next-line @typescript-eslint/no-unsafe-member-access
-          isResumeToken(message.metadata.streamPosition)
-        ? // eslint-disable-next-line @typescript-eslint/no-unsafe-member-access
-          message.metadata.streamPosition
-        : null;
-};
-
 export const mongoDBCheckpointer = <
   MessageType extends Message = Message,
-  // eslint-disable-next-line @typescript-eslint/no-explicit-any
-  CheckpointType = any,
->(): MongoDBCheckpointer<MessageType, CheckpointType> => ({
+>(): MongoDBCheckpointer<MessageType> => ({
   read: async (options, context) => {
-    const result = await readProcessorCheckpoint<CheckpointType>(
-      context.client,
-      options,
-    );
+    const result = await readProcessorCheckpoint(context.client, options);
 
-    return { lastCheckpoint: result?.lastProcessedPosition };
+    // eslint-disable-next-line @typescript-eslint/no-unsafe-assignment
+    return { lastCheckpoint: result?.lastCheckpoint };
   },
   store: async (options, context) => {
-    const newPosition = getCheckpoint<MessageType, CheckpointType>(
-      options.message,
-    );
-
-    const result = await storeProcessorCheckpoint<CheckpointType>(
-      context.client,
-      {
-        lastProcessedPosition: options.lastCheckpoint,
-        newPosition,
-        processorId: options.processorId,
-        partition: options.partition,
-        version: options.version || 0,
-      },
-    );
+    const newPosition = getCheckpoint(options.message);
+
+    const result = await storeProcessorCheckpoint(context.client, {
+      lastProcessedPosition: options.lastCheckpoint,
+      newPosition,
+      processorId: options.processorId,
+      partition: options.partition,
+      version: options.version || 0,
+    });
 
     return result.success
       ? { success: true, newCheckpoint: result.newPosition }
@@ -134,8 +86,6 @@ const mongoDBProcessingScope = (options: {
   client: MongoClient;
   processorId: string;
 }): MessageProcessingScope<MongoDBProcessorHandlerContext> => {
-  // const processorConnectionString = options.connectionString;
-
   const processingScope: MessageProcessingScope<
     MongoDBProcessorHandlerContext
   > = async <Result = MessageHandlerResult>(
@@ -144,15 +94,6 @@ const mongoDBProcessingScope = (options: {
     ) => Result | Promise<Result>,
     partialContext: Partial<MongoDBProcessorHandlerContext>,
   ) => {
-    // const connection = partialContext?.connection;
-    // const connectionString =
-    //   processorConnectionString ?? connection?.connectionString;
-
-    // if (!connectionString)
-    //   throw new EmmettError(
-    //     `MongoDB processor '${options.processorId}' is missing connection string. Ensure that you passed it through options`,
-    //   );
-
     return handler({
       client: options.client,
       ...partialContext,
@@ -174,6 +115,10 @@ export const mongoDBProjector = <EventType extends Event = Event>(
         }
       : undefined,
   };
+  // TODO: This should be eventually moved to the mongoDBProcessingScope
+  // In the similar way as it's made in the postgresql processor
+  // So creating client only if it's needed and different than consumer is passing
+  // through handler context
   const client =
     'client' in connectionOptions && connectionOptions.client
       ? connectionOptions.client
@@ -184,7 +129,7 @@ export const mongoDBProjector = <EventType extends Event = Event>(
 
   return projector<
     EventType,
-    ReadEventMetadataWithGlobalPosition,
+    MongoDBChangeStreamMessageMetadata,
     MongoDBProcessorHandlerContext
   >({
     ...options,

src/packages/emmett-mongodb/src/eventStore/consumers/processorCheckpoint.e2e.spec.ts

@@ -7,7 +7,7 @@ import { MongoClient } from 'mongodb';
 import { after, before, describe, it } from 'node:test';
 import { readProcessorCheckpoint } from './readProcessorCheckpoint';
 import { storeProcessorCheckpoint } from './storeProcessorCheckpoint';
-import type { MongoDBResumeToken } from './subscriptions/types';
+import type { MongoDBResumeToken } from './subscriptions/mongoDbResumeToken';
 
 void describe('storeProcessorCheckpoint and readProcessorCheckpoint tests', () => {
   let mongodb: StartedMongoDBContainer;
@@ -117,7 +117,7 @@ void describe('storeProcessorCheckpoint and readProcessorCheckpoint tests', () =
       processorId,
     });
 
-    assertDeepEqual(result, { lastProcessedPosition: resumeToken2 });
+    assertDeepEqual(result, { lastCheckpoint: resumeToken2 });
   });
 
   void it('it can read a position of a processor with a defined partition', async () => {
@@ -126,6 +126,6 @@ void describe('storeProcessorCheckpoint and readProcessorCheckpoint tests', () =
       partition: 'partition-2',
     });
 
-    assertDeepEqual(result, { lastProcessedPosition: resumeToken1 });
+    assertDeepEqual(result, { lastCheckpoint: resumeToken1 });
   });
 });

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

@@ -1,14 +1,6 @@
+import type { ReadProcessorCheckpointResult } from '@event-driven-io/emmett';
 import type { MongoClient } from 'mongodb';
-import {
-  DefaultProcessotCheckpointCollectionName,
-  defaultTag,
-  type ReadProcessorCheckpointSqlResult,
-} from './types';
-
-// eslint-disable-next-line @typescript-eslint/no-explicit-any
-export type ReadProcessorCheckpointResult<CheckpointType = any> = {
-  lastProcessedPosition: CheckpointType | null;
-};
+import { DefaultProcessotCheckpointCollectionName, defaultTag } from './types';
 
 // eslint-disable-next-line @typescript-eslint/no-explicit-any
 export const readProcessorCheckpoint = async <CheckpointType = any>(
@@ -22,7 +14,7 @@ export const readProcessorCheckpoint = async <CheckpointType = any>(
 ): Promise<ReadProcessorCheckpointResult<CheckpointType>> => {
   const result = await client
     .db(options.databaseName)
-    .collection<ReadProcessorCheckpointSqlResult<CheckpointType>>(
+    .collection<ReadProcessorCheckpointResult<CheckpointType>>(
       options.collectionName || DefaultProcessotCheckpointCollectionName,
     )
     .findOne({
@@ -31,6 +23,6 @@ export const readProcessorCheckpoint = async <CheckpointType = any>(
     });
 
   return {
-    lastProcessedPosition: result !== null ? result.lastProcessedToken : null,
+    lastCheckpoint: result !== null ? result.lastCheckpoint : null,
   };
 };

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

@@ -1,7 +1,7 @@
 import type { MongoClient } from 'mongodb';
 import { compareTwoTokens } from './subscriptions';
 import {
-  type ReadProcessorCheckpointSqlResult,
+  type ReadProcessorCheckpointResult,
   DefaultProcessotCheckpointCollectionName,
   defaultTag,
 } from './types';
@@ -39,7 +39,7 @@ export const storeProcessorCheckpoint = async <Position>(
 > => {
   const checkpoints = client
     .db(dbName)
-    .collection<ReadProcessorCheckpointSqlResult>(
+    .collection<ReadProcessorCheckpointResult>(
       collectionName || DefaultProcessotCheckpointCollectionName,
     );
 
@@ -53,21 +53,21 @@ export const storeProcessorCheckpoint = async <Position>(
   // MISMATCH: we have a checkpoint but lastProcessedPosition doesn’t match
   if (
     current &&
-    compareTwoTokens(current.lastProcessedToken, lastProcessedPosition) !== 0
+    compareTwoTokens(current.lastProcessedPosition, lastProcessedPosition) !== 0
   ) {
     return { success: false, reason: 'MISMATCH' };
   }
 
   // IGNORED: same or earlier position
-  if (current?.lastProcessedToken && newPosition) {
-    if (compareTwoTokens(current.lastProcessedToken, newPosition) !== -1) {
+  if (current?.lastProcessedPosition && newPosition) {
+    if (compareTwoTokens(current.lastProcessedPosition, newPosition) !== -1) {
       return { success: false, reason: 'IGNORED' };
     }
   }
 
   const updateResult = await checkpoints.updateOne(
-    { ...filter, lastProcessedToken: lastProcessedPosition },
-    { $set: { lastProcessedToken: newPosition, version } },
+    { ...filter, lastProcessedPosition: lastProcessedPosition },
+    { $set: { lastProcessedPosition: newPosition, version } },
     { upsert: true },
   );