test: tests, eslint, ts fixed

Author: arturwojnar

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

@@ -62,9 +62,9 @@ export type MongoDBEventStoreConsumerConfig<
   CheckpointType
 > & {
   // from?: any;
-  pulling?: {
-    batchSize?: number;
-  };
+  // pulling?: {
+  //   batchSize?: number;
+  // };
   resilience?: {
     resubscribeOptions?: AsyncRetryOptions;
   };
@@ -224,9 +224,15 @@ export const mongoDBMessagesConsumer = <
           },
         });
 
+        const db = options.client?.db?.();
+
+        if (!db) {
+          throw new EmmettError('MongoDB client is not connected');
+        }
+
         // TODO: Remember to fix.
-        const policy = (await generateVersionPolicies(options.client?.db()!))
-          .changeStreamFullDocumentValuePolicy;
+        const versionPolicies = await generateVersionPolicies(db);
+        const policy = versionPolicies.changeStreamFullDocumentValuePolicy;
 
         await stream.start({
           getFullDocumentValue: policy,

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

@@ -91,7 +91,6 @@ const isResumeToken = (value: any): value is MongoDBResumeToken =>
 
 export const getCheckpoint = <
   MessageType extends AnyMessage = AnyMessage,
-  // eslint-disable-next-line @typescript-eslint/no-explicit-any
   CheckpointType = MongoDBCheckpointer,
   MessageMetadataType extends
     ReadEventMetadataWithGlobalPosition<CheckpointType> = ReadEventMetadataWithGlobalPosition<CheckpointType>,

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

@@ -36,11 +36,13 @@ export type MongoDBSubscriptionOptions<
   MessageType extends Message = Message,
   MessageMetadataType extends
     MongoDBReadEventMetadata = MongoDBRecordedMessageMetadata,
-  CheckpointType = MongoDBResumeToken,
+  // CheckpointType = MongoDBResumeToken,
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  CheckpointType = any,
 > = {
   from?: CurrentMessageProcessorPosition<CheckpointType>;
   client: MongoClient;
-  batchSize: number;
+  // batchSize: number;
   eachBatch: BatchRecordedMessageHandlerWithoutContext<
     MessageType,
     MessageMetadataType
@@ -346,7 +348,6 @@ const createChangeStream = <
 
 const subscribe =
   (getFullDocumentValue: ChangeStreamFullDocumentValuePolicy, db: Db) =>
-  // eslint-disable-next-line @typescript-eslint/no-explicit-any
   <EventType extends Message = AnyMessage, CheckpointType = MongoDBResumeToken>(
     resumeToken?: MongoDBSubscriptionStartFrom<CheckpointType>,
   ) => {
@@ -361,7 +362,7 @@ export const mongoDBSubscription = <
 >({
   client,
   from,
-  batchSize,
+  // batchSize,
   eachBatch,
   resilience,
 }: MongoDBSubscriptionOptions<
@@ -396,14 +397,18 @@ export const mongoDBSubscription = <
           callback?.();
           resolve();
         } catch (error) {
-          reject(error);
+          reject(
+            error instanceof Error
+              ? error
+              : typeof error === 'string'
+                ? new Error(error)
+                : new Error('Unknown error'),
+          );
         }
       });
     } else {
       try {
         await subscription.close();
-      } catch (error) {
-        throw error;
       } finally {
         callback?.();
       }
@@ -434,7 +439,7 @@ export const mongoDBSubscription = <
           >({
             client,
             from,
-            batchSize,
+            // batchSize,
             eachBatch,
             resilience,
           });
@@ -449,7 +454,7 @@ export const mongoDBSubscription = <
 
               if (isMongoDBResumeToken(result)) {
                 options.startFrom = {
-                  lastCheckpoint: result,
+                  lastCheckpoint: result as ResumeToken,
                 };
                 done();
                 return;
@@ -513,8 +518,7 @@ export const mongoDBSubscription = <
 
       start = (async () => {
         isRunning = true;
-        const a = pipeMessages(options);
-        return a;
+        return pipeMessages(options);
       })();
 
       return start;

src/packages/emmett-mongodb/src/eventStore/mongoDBEventStore.subscription.e2e.spec.ts

@@ -67,24 +67,6 @@ void describe('MongoDBEventStore subscription', () => {
       quantity: 10,
       price: 3,
     }) as PricedProductItem;
-  const timeoutGuard = async (
-    action: () => Promise<void>,
-    timeoutAfterMs = 1000,
-  ) => {
-    return new Promise<void>((resolve, reject) => {
-      const timer = setTimeout(() => {
-        reject(new Error('timeout'));
-        clearTimeout(timer);
-      }, timeoutAfterMs);
-
-      action()
-        .catch(noop)
-        .finally(() => {
-          clearTimeout(timer);
-          resolve();
-        });
-    });
-  };
 
   before(async () => {
     mongodb = await new MongoDBContainer('mongo:8.0.10').start();
@@ -126,11 +108,11 @@ void describe('MongoDBEventStore subscription', () => {
       stopAfter: (event) => {
         if (event.data.productItem.productId === lastProductItemIdTest1) {
           messageProcessingPromise1.resolve();
-          consumer.stop();
+          consumer.stop().catch(noop);
         }
         if (event.data.productItem.productId === lastProductItemIdTest2) {
           messageProcessingPromise2.resolve();
-          consumer.stop();
+          consumer.stop().catch(noop);
         }
 
         return (
@@ -185,11 +167,7 @@ void describe('MongoDBEventStore subscription', () => {
       { expectedStreamVersion: 2n },
     );
 
-    try {
-      await consumer.start();
-    } catch (err) {
-      console.error(err);
-    }
+    await consumer.start();
 
     const stream = await collection.findOne(
       { streamName },
@@ -204,7 +182,7 @@ void describe('MongoDBEventStore subscription', () => {
     assertTrue(stream.metadata.updatedAt instanceof Date);
   });
 
-  void it.skip('should renew after the last event', async () => {
+  void it('should renew after the last event', async () => {
     assertTrue(!!processor);
     assert(processor);
 
@@ -215,8 +193,6 @@ void describe('MongoDBEventStore subscription', () => {
     assertIsNotNull(stream);
     assertEqual(3n, stream.metadata.streamPosition);
 
-    await consumer.start();
-
     const position = await processor.start({ client });
 
     assertTrue(!!position);
@@ -230,6 +206,10 @@ void describe('MongoDBEventStore subscription', () => {
       compareTwoMongoDBTokens(position.lastCheckpoint, lastResumeToken!),
     );
 
+    const consumerPromise = consumer.start();
+
+    await new Promise((resolve) => setTimeout(resolve, 1000));
+
     await eventStore.appendToStream<ShoppingCartEvent>(
       streamName,
       [
@@ -241,7 +221,7 @@ void describe('MongoDBEventStore subscription', () => {
       { expectedStreamVersion: 3n },
     );
 
-    await timeoutGuard(() => messageProcessingPromise2);
+    await consumerPromise;
 
     stream = await collection.findOne({ streamName }, { useBigInt64: true });
     assertIsNotNull(stream);