Updating project

src/main.ts

@@ -1,4 +1,3 @@
-/* eslint-disable no-console */
 import {
   AzureEventhubSasFromString,
   KafkaProducerCompact,
@@ -28,6 +27,9 @@ import {
 
 dotenv.config();
 
+const databaseName = "mongo-cdc-poc-mongodb";
+const collectionName = "students";
+
 const getCosmosConnectionURI = (): TE.TaskEither<Error, string> =>
   pipe(
     CosmosDBConfig.decode(MONGO_CONFIG),
@@ -65,7 +67,6 @@ export const log = withLogger(IO.io)(C.log);
 const waitForExit = (client: MongoClient): TE.TaskEither<Error, void> =>
   TE.tryCatch(
     async () => {
-      console.log("waiting");
       process.stdin.resume();
       process.on("SIGINT", () => {
         disconnectMongo(client);
@@ -84,15 +85,17 @@ const exitFromProcess = (): TE.TaskEither<Error, void> =>
 const sendMessageEventHub =
   (messagingClient: KafkaProducerCompact<Student>) =>
   <T = Document>(change: ChangeStreamDocument<T>): void =>
-    pipe(change, transform, void sendMessages(messagingClient));
+    void pipe(change, transform, (students) =>
+      sendMessages(messagingClient)(students)()
+    );
 
 const main = () =>
   pipe(
     TE.Do,
     getCosmosConnectionURI,
     TE.bind("client", (connectionUri) => mongoConnect(connectionUri)),
-    TE.bind("db", ({ client }) => getMongoDb(client, "customer")),
-    TE.bind("collection", ({ db }) => getMongoCollection(db, "students")),
+    TE.bind("db", ({ client }) => getMongoDb(client, databaseName)),
+    TE.bind("collection", ({ db }) => getMongoCollection(db, collectionName)),
     TE.bind("messagingClient", () => getEventHubProducer()),
     TE.chainFirst(({ collection, messagingClient }) =>
       pipe(
@@ -119,4 +122,6 @@ const main = () =>
     TE.orElse(exitFromProcess)
   )();
 
-void main();
+main()
+  .catch((err) => log(() => "Error: " + String(err)))
+  .finally(() => log(() => "Exiting..."));

src/mapper/students.ts

@@ -1,3 +1,4 @@
+/* eslint-disable no-console */
 import * as A from "fp-ts/lib/Array";
 import * as O from "fp-ts/lib/Option";
 
@@ -7,26 +8,26 @@ import { Student } from "../model/student";
 
 type Transformer = <T = Document>(data: ChangeStreamDocument<T>) => Student[];
 
-const isHandled = <T = Document>(data: ChangeStreamDocument<T>) =>
+const isInsert = <T = Document>(data: ChangeStreamDocument<T>) =>
   "documentKey" in (data as ChangeStreamInsertDocument<T>);
 const insertTransformer: Transformer = <T = Document>(
   data: ChangeStreamDocument<T>
 ) =>
   pipe(
     O.fromNullable(data),
-    O.filter(isHandled),
+    O.filter(isInsert),
     O.chain(() =>
       O.fromNullable((data as ChangeStreamInsertDocument<T>).fullDocument)
     ),
-    O.map((newObject) => {
+    O.map((data) => {
       // eslint-disable-next-line @typescript-eslint/no-unused-vars
       const { _id, ...newObj } =
-        newObject as unknown as ChangeStreamInsertDocument<T>;
+        data as unknown as ChangeStreamInsertDocument<T>;
       return newObj;
     }),
     O.fold(
       () => [],
-      (newObject) => A.of(newObject as unknown as Student)
+      (data) => A.of(data as unknown as Student)
     )
   );
 // const deleteTransformer: Transformer = (data) => { ... };

src/mongo/mongoOperation.ts

@@ -32,7 +32,15 @@ export const getMongoCollection = <T = Document>(
   collectionName: string
 ): TE.TaskEither<Error, Collection<T>> =>
   TE.tryCatch(
-    async () => db.collection<T>(collectionName),
+    async () => {
+      const collections = await db
+        .listCollections({ name: collectionName })
+        .toArray();
+      if (collections.length === 0) {
+        return db.createCollection<T>(collectionName);
+      }
+      return db.collection(collectionName);
+    },
     (reason) =>
       new Error(
         `Impossible to connect to Get the ${collectionName} collection: " ${reason}`
@@ -51,7 +59,16 @@ export const watchMongoCollection = <T = Document>(
   collection: Collection<T>
 ): TE.TaskEither<Error, ChangeStream<T, ChangeStreamDocument<T>>> =>
   TE.tryCatch(
-    async () => collection.watch(),
+    async () =>
+      collection.watch(
+        [
+          {
+            $match: { operationType: { $in: ["insert", "update", "replace"] } },
+          },
+          { $project: { _id: 1, fullDocument: 1, ns: 1, documentKey: 1 } },
+        ],
+        { fullDocument: "updateLookup" }
+      ),
     (reason) =>
       new Error(
         `Impossible to watch the ${collection.namespace} collection: " ${reason}`