Merge pull request #66 from tungv/enhance-jerni-dev

votrungquan1999 · web-flow · commit a828f2728f7f · 2024-09-10T22:34:44.000+07:00
Enhance jerni dev
diff --git a/packages/integrations/tests/CLI/cli-integration.spec.ts b/packages/integrations/tests/CLI/cli-integration.spec.ts
@@ -11,7 +11,6 @@ afterAll(cleanUpTestDatabase);
 
 it("CLI call should project events correctly", async () => {
   const dbName = `jerni_integration_test_${nanoid()}`;
-  const eventDbName = `jerni_integration_test_events_${nanoid()}`;
 
   const { server } = createServer();
   const port = server.port;
@@ -23,8 +22,6 @@ it("CLI call should project events correctly", async () => {
   const process = exec(
     `MONGODB_DBNAME=${dbName} \
     MONGODB_URL=mongodb://127.0.0.1:27017 \
-    EVENTS_DB_MONGODB_URL=mongodb://127.0.0.1:27017 \
-    EVENTS_DB_MONGODB_NAME=${eventDbName} \
     EVENTS_SERVER=http://localhost:${port}/ \
     PORT=${healthCheckPort} \
     bun run ${jerniCliPath} \
diff --git a/packages/integrations/tests/CLI/jerni-dev.spec.ts b/packages/integrations/tests/CLI/jerni-dev.spec.ts
@@ -18,7 +18,6 @@ describe("Jerni Dev Integration", () => {
     const port = Math.floor(Math.random() * 10000) + 10000;
     const dbFileName = `./test-events-db-${nanoid()}.yml`;
     const mongodbName = `jerni_integration_test_${nanoid()}`;
-    const eventDbName = `jerni_integration_test_events_${nanoid()}`;
 
     const devCliPath = path.resolve(__dirname, "../../../jerni/src/dev-cli/index.ts");
     const initFileName = path.resolve(__dirname, "./makeTestJourneyCli.ts");
@@ -27,8 +26,6 @@ describe("Jerni Dev Integration", () => {
     const childProcess = exec(
       `PORT=${port}\
         MONGODB_DBNAME=${mongodbName}\
-        EVENTS_DB_MONGODB_URL=mongodb://127.0.0.1:27017 \
-        EVENTS_DB_MONGODB_NAME=${eventDbName} \
         bun run ${devCliPath} ${initFileName} ${dbFilePath}`,
       (error, stdout, stderr) => {
         // console.log(`stdout: ${stdout}`);
diff --git a/packages/integrations/tests/startWorker.ts b/packages/integrations/tests/startWorker.ts
@@ -1,14 +1,10 @@
 import begin from "@jerni/jerni-3/lib/begin";
 import type { JourneyInstance } from "@jerni/jerni-3/types";
-import { nanoid } from "nanoid";
 
 export default async function startWorker(journey: JourneyInstance, signal: AbortSignal) {
-  const eventDbName = `jerni_integration_test_events_${nanoid()}`;
-
-  process.env.EVENTS_DB_MONGODB_URL = "mongodb://127.0.0.1:27017";
-  process.env.EVENTS_DB_MONGODB_NAME = eventDbName;
-
   for await (const events of begin(journey, signal)) {
     // console.log("events", events);
+
+    return events;
   }
 }
diff --git a/packages/jerni-store-mongodb/jsr.json b/packages/jerni-store-mongodb/jsr.json
@@ -1,8 +1,8 @@
 {
   "name": "@jerni/store-mongodb",
-  "version": "0.4.4",
+  "version": "0.4.5",
   "exports": {
     ".": "./src/index.ts",
     "./types": "./src/types.ts"
   }
-}
+}
diff --git a/packages/jerni-store-mongodb/src/optimistic/getBulkOperations.ts b/packages/jerni-store-mongodb/src/optimistic/getBulkOperations.ts
@@ -1,4 +1,4 @@
-import type { AnyBulkWriteOperation, Document, UpdateFilter, UpdateOneModel } from "mongodb";
+import type { AnyBulkWriteOperation, Document, UpdateOneModel } from "mongodb";
 import type {
   DeleteManyOp,
   DeleteOneOp,
@@ -11,16 +11,9 @@ import type {
 
 interface ChangeWithOp<Change> {
   change: Change;
-  __op: number;
   __v: number;
 }
 
-type OptimisticDocument = {
-  __v: number;
-  __op: number;
-  [key: string]: unknown;
-};
-
 interface Newer {
   $or: [{ __v: { $gt: number } }, { __v: number; __op: { $gte: number } }];
 }
diff --git a/packages/jerni-store-mongodb/src/store.ts b/packages/jerni-store-mongodb/src/store.ts
@@ -1,11 +1,11 @@
 import { setTimeout } from "node:timers/promises";
-import { type Collection, type Db, type Document, MongoClient } from "mongodb";
+import { type AnyBulkWriteOperation, type Collection, type Db, type Document, MongoClient } from "mongodb";
 import makeTestLogger from "../tests/helpers/makeTestLogger";
 import getCollectionName from "./getCollectionName";
 import type MongoDBModel from "./model";
 import getBulkOperations from "./optimistic/getBulkOperations";
 import { Signal, clearModelSlots, runWithModel } from "./read";
-import type { Changes, JourneyCommittedEvent, MongoDBStore, MongoDBStoreConfig, MongoOps } from "./types";
+import type { Changes, JourneyCommittedEvent, MongoDBStore, MongoDBStoreConfig } from "./types";
 
 interface SnapshotDocument {
   __v: number;
@@ -294,19 +294,12 @@ export default async function makeMongoDBStore(config: MongoDBStoreConfig): Prom
 
     // output is an array of changes per event.
     // however for efficiency, we need to handle event per model
-    type VersionedChange = {
-      change: MongoOps<Document>;
-      __op: number;
-      __v: number;
-    };
-
-    const changesPerModel: VersionedChange[][] = [];
+    // so we need to group changes per model
+    const bulkWritesPerModel: AnyBulkWriteOperation<Document>[][] = [];
 
     for (const allChangesForAnEvent of outputs) {
       let modelIndex = 0;
       for (const changesForAModel of allChangesForAnEvent) {
-        let __op = 0;
-
         if (changesForAModel === undefined || changesForAModel.length === 0) {
           modelIndex++;
           continue;
@@ -315,37 +308,37 @@ export default async function makeMongoDBStore(config: MongoDBStoreConfig): Prom
         const changesWithOp = changesForAModel.map((change) => {
           return {
             change,
-            __op: __op++,
             __v: events[eventIndex].id,
           };
         });
 
-        // add changes to changesPerModel
-        if (!changesPerModel[modelIndex]) {
-          changesPerModel[modelIndex] = [];
+        const operations = getBulkOperations(changesWithOp);
+
+        // add operations to the bulkWritesPerModel
+        if (!bulkWritesPerModel[modelIndex]) {
+          bulkWritesPerModel[modelIndex] = [];
         }
-        changesPerModel[modelIndex].push(...changesWithOp);
+        bulkWritesPerModel[modelIndex].push(...operations);
         modelIndex++;
       }
       eventIndex++;
     }
 
     await Promise.all(
-      changesPerModel.map(async (changesForAModel, modelIndex) => {
+      bulkWritesPerModel.map(async (bulkWritesForAModel, modelIndex) => {
         const model = models[modelIndex];
         const outputForThisModel = changes[modelIndex];
-        if (changesForAModel === undefined || changesForAModel.length === 0) {
+        if (bulkWritesForAModel === undefined || bulkWritesForAModel.length === 0) {
           return;
         }
 
         const collection = db.collection(getCollectionName(model));
-        const bulkWriteOperations = getBulkOperations(changesForAModel);
 
         if (abortSignal.aborted) {
           return;
         }
 
-        const res = await collection.bulkWrite(bulkWriteOperations, { ordered: true, writeConcern: { w: "majority" } });
+        const res = await collection.bulkWrite(bulkWritesForAModel, { ordered: true, writeConcern: { w: "majority" } });
 
         skipByModel[modelIndex] = Math.max(skipByModel[modelIndex], events[events.length - 1].id);
 
diff --git a/packages/jerni-store-mongodb/src/types.ts b/packages/jerni-store-mongodb/src/types.ts
@@ -63,7 +63,7 @@ export interface MongoDBStore {
     >,
   ) => void;
 
-  getDriver<T extends Document>(model: MongoDBModel<T>): Promise<AsyncDisposable>;
+  getDriver<T extends Document>(model: MongoDBModel<T>): Promise<Collection<T> & AsyncDisposable>;
   handleEvents: (
     events: JourneyCommittedEvent[],
     signal?: AbortSignal,
diff --git a/packages/jerni/global.d.ts b/packages/jerni/global.d.ts
@@ -0,0 +1,8 @@
+declare namespace NodeJS {
+  interface ProcessEnv {
+    readonly NODE_ENV: string;
+    PORT: string;
+    EVENTS_SERVER: string;
+    JERNI_CLI_SQLITE_PATH: string;
+  }
+}
diff --git a/packages/jerni/src/begin.ts b/packages/jerni/src/begin.ts
@@ -174,7 +174,14 @@ export default async function* begin(journey: JourneyInstance, signal: AbortSign
       try {
         // handle events must stop after 10 seconds
         const start = Date.now();
-        const { output, lastId } = await handleEventBatch(config.stores, config.onError, events, logger, timeout);
+        const { output, lastId } = await handleEventBatch(
+          config.stores,
+          config.onError,
+          config.onReport,
+          events,
+          logger,
+          timeout,
+        );
         latestHandled = lastId;
         const total = Date.now() - start;
         const budgetPercentage = (total / timeBudget) * 100;
@@ -207,8 +214,9 @@ export default async function* begin(journey: JourneyInstance, signal: AbortSign
           logger.info(`${INF} [HANDLING_EVENT] retrying with maxEvents = ${maxEvents}`);
           continue mainLoop;
         }
-        console.error(`${ERR} [HANDLING_EVENT] something went wrong while handling events`, ex);
-        process.exit(1);
+
+        logger.error(`${ERR} [HANDLING_EVENT] something went wrong while handling events`, ex);
+        break mainLoop;
       } finally {
         clearInterval(progressBarId);
       }
@@ -225,8 +233,16 @@ export default async function* begin(journey: JourneyInstance, signal: AbortSign
       resolve();
     });
 
+    // also check if the signal is aborted
+    signal.addEventListener("abort", resolve, { once: true, signal: ctrl.signal });
+
     newEventNotifier.addEventListener("latest", resolve, { once: true, signal: ctrl.signal });
 
+    // check if the signal is aborted before waiting
+    if (signal.aborted) {
+      break;
+    }
+
     await promise;
     // logger.debug(
     //   `${DBG} [HANDLING_EVENT] waiting for new events… after ${prettyMilliseconds(Date.now() - lastProcessingTime)}`,
@@ -237,6 +253,7 @@ export default async function* begin(journey: JourneyInstance, signal: AbortSign
 async function handleEventBatch(
   stores: JourneyConfig["stores"],
   onError: JourneyConfig["onError"],
+  onReport: JourneyConfig["onReport"],
   events: JourneyCommittedEvent[],
   logger: Logger,
   signal: AbortSignal,
@@ -257,6 +274,12 @@ async function handleEventBatch(
       stores.map(async (store) => {
         const output = await singleStoreHandleEvents(store, events, 0, events.length, logger, onError, signal);
 
+        onReport?.("store_output", {
+          store: store,
+
+          output,
+        });
+
         return output;
       }),
     ),
diff --git a/packages/jerni/src/createJourney.ts b/packages/jerni/src/createJourney.ts
@@ -112,7 +112,7 @@ export default function createJourney(config: JourneyConfig): JourneyInstance {
       }
     },
     // biome-ignore lint/suspicious/noExplicitAny: because this is a placeholder, the client that uses jerni would override this type
-    async getReader(model: any): Promise<AsyncDisposable> {
+    async getReader(model: any): Promise<any> {
       registerOnce();
       const store = modelToStoreMap.get(model);
 
diff --git a/packages/jerni/src/dev-cli/clearData.ts b/packages/jerni/src/dev-cli/clearData.ts
@@ -0,0 +1,9 @@
+import type { JourneyInstance } from "@jerni/jerni-3/types";
+
+export default async function clearDatabase(journey: JourneyInstance) {
+  const config = journey.getConfig();
+
+  for (const store of config.stores) {
+    await store.clean();
+  }
+}
diff --git a/packages/jerni/src/dev-cli/initJerniDev.ts b/packages/jerni/src/dev-cli/initJerniDev.ts
@@ -7,6 +7,7 @@ import begin from "../begin";
 import { ERR, INF } from "../cli-utils/log-headers";
 import { printErrorObject } from "../printErrorObject";
 import type { JourneyInstance } from "../types/journey";
+import clearData from "./clearData";
 
 interface StartJerniDevOptions {
   cleanStart?: boolean;
@@ -50,7 +51,7 @@ export default async function initJerniDev(filePath: string | undefined) {
           console.log("%s clean start jerni, clearing database…", INF);
 
           // clear database
-          // await clearData(journey);
+          await clearData(journey);
         }
 
         console.log("%s jerni dev start %s", INF, bold(validFilePath));
diff --git a/packages/jerni/src/getEventStream.ts b/packages/jerni/src/getEventStream.ts
@@ -158,13 +158,13 @@ async function* retrieveJourneyCommittedEvents(
 ): AsyncGenerator<EventStreamReturnType, void, unknown> {
   signal.throwIfAborted();
   try {
+    // passing signal to fetch will throw uncatchable error, instead, add the signal to body.pipeThrough
     const resp = await fetch(url.toString(), {
       headers: {
         authorization: `Basic ${btoa(`${url.username}:${url.password}`)}`,
         "burst-count": String(batchSize),
         "last-event-id": String(lastSeenId),
       },
-      signal,
     });
 
     if (!resp.ok) {
@@ -180,7 +180,7 @@ async function* retrieveJourneyCommittedEvents(
       connected_at: Date.now(),
     };
 
-    const stream = resp.body.pipeThrough(new TextDecoderStream());
+    const stream = resp.body.pipeThrough(new TextDecoderStream(), { signal });
 
     const pending = [] as string[];
 
diff --git a/packages/jerni/src/tests/helpers/initJourney.ts b/packages/jerni/src/tests/helpers/initJourney.ts
@@ -2,7 +2,7 @@ import { mock } from "bun:test";
 import createJourney from "src/createJourney";
 import type { JourneyConfig } from "../../types/config";
 
-export async function initJourney(eventServerUrl: string, stores: JourneyConfig["stores"]) {
+export function initJourney(eventServerUrl: string, stores: JourneyConfig["stores"]) {
   const logger = makeTestLogger();
 
   const onReport = mock((type: string, payload: unknown) => {});
diff --git a/packages/jerni/src/tests/helpers/makeTestJourneyCli.ts b/packages/jerni/src/tests/helpers/makeTestJourneyCli.ts
@@ -17,7 +17,7 @@ function getEventsServerUrl() {
 }
 
 export default async function initializeJourney(eventsServer = getEventsServerUrl()) {
-  const { journey } = await initJourney(eventsServer, []);
+  const { journey } = initJourney(eventsServer, []);
 
   return journey;
 }
diff --git a/packages/jerni/src/tests/subscribe-from-last-saved-event-id.spec.ts b/packages/jerni/src/tests/subscribe-from-last-saved-event-id.spec.ts
@@ -34,10 +34,10 @@ test("start subscription from last saved event id", async () => {
 
   const ctrl1 = new AbortController();
 
-  const worker1 = await initJourney(eventServerUrl, []);
+  const worker1 = initJourney(eventServerUrl, []);
 
   // subscribe to events
-  for await (const _events of begin(worker1.journey, ctrl1.signal)) {
+  for await (const _output of begin(worker1.journey, ctrl1.signal)) {
     ctrl1.abort();
   }
 
@@ -54,7 +54,7 @@ test("start subscription from last saved event id", async () => {
   const ctrl2 = new AbortController();
 
   let processedId = 0;
-  const worker2 = await initJourney(eventServerUrl, [
+  const worker2 = initJourney(eventServerUrl, [
     {
       name: "test",
       clean: async () => {},
@@ -83,7 +83,7 @@ test("start subscription from last saved event id", async () => {
   ]);
 
   // subscribe to events
-  for await (const output of begin(worker2.journey, ctrl2.signal)) {
+  for await (const _output of begin(worker2.journey, ctrl2.signal)) {
     if (processedId === 2) {
       ctrl2.abort();
     }
diff --git a/packages/jerni/src/types/config.ts b/packages/jerni/src/types/config.ts
@@ -48,7 +48,7 @@ export interface Store {
   registerModels: (map: Map<any, Store>) => void;
 
   // biome-ignore lint/suspicious/noExplicitAny: Jerni can take any model and store, there is no way to enforce the type here. However, the type of mongodb store is enforced in the store-mongodb package
-  getDriver(model: any): Promise<AsyncDisposable>;
+  getDriver(model: any): Promise<any>;
   // biome-ignore lint/suspicious/noExplicitAny: the return type is dependent on the implementation of the store
   handleEvents: (events: JourneyCommittedEvent[], signal?: AbortSignal) => Promise<any>;
   getLastSeenId: () => Promise<number>;
diff --git a/packages/jerni/src/worker.ts b/packages/jerni/src/worker.ts
@@ -54,6 +54,8 @@ export default async function initWorker(filePath: string | undefined, port: num
         for await (const _outputs of begin(journey, ctrl.signal, { dbFolder: sqliteFolderPath })) {
           // console.log("outputs", outputs);
         }
+
+        ctrl.abort();
       } catch (error) {
         console.error("%s cannot initialize journey object", ERR);
 

Original file line number	Diff line number	Diff line change
`@@ -1,8 +1,8 @@`
`1`	`1`	`{`
`2`	`2`	`"name": "@jerni/store-mongodb",`
`3`		`- "version": "0.4.4",`
	`3`	`+ "version": "0.4.5",`
`4`	`4`	`"exports": {`
`5`	`5`	`".": "./src/index.ts",`
`6`	`6`	`"./types": "./src/types.ts"`
`7`	`7`	`}`
`8`		`-}`
	`8`	`+}`
Original file line number	Diff line number	Diff line change
`@@ -112,7 +112,7 @@ export default function createJourney(config: JourneyConfig): JourneyInstance {`
`112`	`112`	`}`
`113`	`113`	`},`
`114`	`114`	`// biome-ignore lint/suspicious/noExplicitAny: because this is a placeholder, the client that uses jerni would override this type`
`115`		`- async getReader(model: any): Promise<AsyncDisposable> {`
	`115`	`+ async getReader(model: any): Promise<any> {`
`116`	`116`	`registerOnce();`
`117`	`117`	`const store = modelToStoreMap.get(model);`
`118`	`118`