Add events graph

package.json

@@ -38,13 +38,15 @@
   },
   "devDependencies": {
     "@types/lodash": "^4.14.178",
+    "@types/node": "^18.6.2",
     "axios": "^0.26.0",
     "barehttp": "^0.5.0",
     "code-concierge": "^1.1.0",
     "typedoc": "^0.22.13",
     "typedoc-plugin-markdown": "^3.11.14"
   },
   "dependencies": {
+    "callsites": "^3.1.0",
     "fp-ts": "^2.11.8",
     "hyperid": "^3.0.0",
     "io-ts": "^2.2.16",

src/context/context.ts → src/context/index.ts

@@ -43,8 +43,11 @@ function after(asyncId: number) {
   context.current = prevStates.get(asyncId);
 }
 
+let render = 0;
 function destroy(asyncId: number) {
   if (context.traces.has(asyncId)) {
+    console.log('render', render++);
+    console.log(context.traces.get(asyncId)?.events);
     context.traces.delete(asyncId);
     prevStates.delete(asyncId);
   }

src/engine/event.ts

@@ -3,8 +3,9 @@ import hyperid from 'hyperid';
 import { cloneDeep, isEqual } from 'lodash';
 
 import { deepFreeze } from './utils';
+import { transportAsyncStorage } from './typed-bus';
 
-import { context } from '../context/context';
+import { context } from '../context';
 
 const uuidGenerate = hyperid();
 const hookIdGenerate = hyperid();
@@ -16,20 +17,24 @@ class CustomSet<T> extends Set<T> {
 }
 export class Event<T = any> {
   uuid: string;
+  rootUUID?: string;
   hookId?: string;
   hookIdStale = false;
+  launchedFrom?: string;
 
   executionId?: string;
   timestamp: number;
   payload: T;
   orphanTransports?: CustomSet<string>;
   publishedTransports?: CustomSet<string>;
 
-  private constructor(payload: any, hook?: boolean) {
+  private constructor(payload: any, startHook?: boolean) {
     this.uuid = uuidGenerate();
-    this.hookId = this.getHook(hook);
+    this.hookId = this.getHook(startHook);
     this.timestamp = Date.now();
     this.payload = deepFreeze(payload);
+    this.rootUUID = context.current?.currentEvent?.rootUUID ?? context.current?.currentEvent?.uuid;
+    this.launchedFrom = transportAsyncStorage.getStore() as string;
 
     Object.defineProperty(this, 'uuid', { writable: false, configurable: false });
     Object.defineProperty(this, 'timestamp', { writable: false, configurable: false });
@@ -41,15 +46,15 @@ export class Event<T = any> {
     this.executionId = executionId;
   }
 
-  getHook(hook?: boolean) {
+  getHook(startHook?: boolean) {
     if (
       context.current?.currentEvent?.hookId &&
       context.current?.currentEvent?.hookIdStale === false
     ) {
       return context.current?.currentEvent?.hookId;
     }
 
-    return hook ? hookIdGenerate() : undefined;
+    return startHook ? hookIdGenerate() : undefined;
   }
 
   getUniqueStamp(transport?: string) {
@@ -97,7 +102,7 @@ export class Event<T = any> {
     this.publishedTransports.add(transport);
   }
 
-  static create<T>(payload: T, hook?: boolean) {
-    return new Event<T>(cloneDeep(payload), hook);
+  static create<T>(payload: T, startHook?: boolean) {
+    return new Event<T>(cloneDeep(payload), startHook);
   }
 }

src/engine/transport.ts

@@ -1,14 +1,17 @@
 import LRUCache from 'lru-cache';
 import * as iots from 'io-ts';
 import { isLeft } from 'fp-ts/lib/Either';
+import hyperid from 'hyperid';
 
 import { Event } from './event';
+import { transportAsyncStorage } from './typed-bus';
 
 import { EventBaseType } from '../validation/event-base-type';
 import { reporter } from '../validation/reporter';
 
 export type MatchEvent = (event: Event) => 'ok' | { [k: string]: string };
-export type ConsumerMethod = {
+export type ConsumerSignature = {
+  loc: string;
   contract: iots.Any;
   exec: (...args: any[]) => any;
   matchEvent: MatchEvent;
@@ -24,7 +27,7 @@ const cache = new LRUCache({ max: 10000, ttl: 10000 });
 export abstract class Transport {
   abstract name: string;
 
-  consumers: ConsumerMethod[] = [];
+  consumers: ConsumerSignature[] = [];
   ready = true;
   waitForReady = false;
   lastEvent: Event | undefined;
@@ -42,9 +45,17 @@ export abstract class Transport {
   } | null> {
     if (!this.ready) await this.waitForTransportReadiness();
 
+    console.log({ last: this.lastEvent, event });
+    if (this.lastEvent?.isAfter(event)) {
+      console.error(
+        `Next event to publish was produced before than the last published event or it's equal. Discarded`,
+      );
+      return null;
+    }
     this.lastEvent = event;
 
     if (typeof this._publish === 'function') {
+      this.lastEvent = event;
       return {
         ...(await this._publish(event)),
         transport: this.name,
@@ -55,33 +66,36 @@ export abstract class Transport {
 
     for (const consumer of this.consumers) {
       if (cache.get(event.getUniqueStamp(consumer.id))) {
+        this.lastEvent = event;
         // we dont publish this event into this consumer anymore
         return null;
       }
 
-      if (this.lastEvent?.isAfter(event)) {
-        console.error(
-          `Next event to publish was produced before than the last published event or it's equal. Discarded`,
-        );
-        return null;
-      }
-
       cache.set(event.getUniqueStamp(consumer.id), true);
 
       const okOrError = consumer.matchEvent(event);
       if (okOrError === 'ok') {
         publishedConsumers.push(
           new Promise<void>(async (resolve, reject) => {
+            let rejected = false;
             try {
-              await consumer.exec(event.payload);
+              await transportAsyncStorage.run(
+                {
+                  consumerId: consumer.id,
+                  consumerFunctionName: consumer.exec.name,
+                },
+                async () => await consumer.exec(event.payload),
+              );
+              // await
             } catch (reason: any) {
               reason.id = consumer.id;
               reason.execName =
                 consumer.exec.name ||
                 'Anonymous Function, please do not use arrow functions for the consumers';
+              rejected = true;
               reject(reason);
             } finally {
-              resolve();
+              if (!rejected) resolve();
             }
           }),
         );
@@ -130,7 +144,13 @@ export abstract class Transport {
   /**
    * Adds a consumer to the transport itself - internal method
    */
-  addConsumer(contract: iots.Any, fn: () => any, consumerId: string, hookId?: string): void {
+  addConsumer(
+    contract: iots.Any,
+    fn: () => any,
+    consumerId: string,
+    loc: string,
+    hookId?: string,
+  ): void {
     if (this.consumers.findIndex(({ exec }) => exec === fn) !== -1) {
       console.log(
         'Cant add a consumer to a transport that already has one with the same reference',
@@ -152,7 +172,7 @@ export abstract class Transport {
       return 'ok';
     };
 
-    this.consumers.push({ contract, exec: fn, matchEvent, id: consumerId });
+    this.consumers.push({ contract, exec: fn, matchEvent, loc, id: consumerId });
   }
 
   /**

src/engine/typed-bus.ts

@@ -1,16 +1,21 @@
 import * as iots from 'io-ts';
 import hyperid from 'hyperid';
+import callsites from 'callsites';
 
 import { Event } from './event';
 import { Transport } from './transport';
 import { EventsStore } from './events-store';
 import { DumpController } from './dump-controller';
 
 import { InternalTransport } from '../transports/internal-transport';
-import { context } from '../context/context';
+import { context } from '../context';
+
+import { AsyncLocalStorage } from 'async_hooks';
 
 const generateConsumerId = hyperid();
 
+export const transportAsyncStorage = new AsyncLocalStorage();
+
 type PublishOptions<T> = {
   onlySendTo?: string[];
   hook?: T extends iots.Any ? T : never;
@@ -33,6 +38,7 @@ export class TypedBusClass {
     eventData: any,
     options: PublishOptions<T> = {},
   ): Promise<{ result: R; hookId: string }> {
+    console.log('async storage store', transportAsyncStorage.getStore());
     const publishedTransports: string[] = [];
     const event = Event.create(eventData, typeof options.hook !== undefined);
 
@@ -42,13 +48,17 @@ export class TypedBusClass {
     if (options.hook) {
       hookPromise = new Promise<any>((resolve, reject) => {
         let consumerId = '';
-        const timoutRef = setTimeout(() => {
+        const timeoutRef = setTimeout(() => {
+          clearTimeout(timeoutRef);
           reject(new Error(`Timeout exceeded for a waiting hook ${options.hook!.name}`));
         }, options.hookTimeout || 10000);
 
         const resolver = (resultData: unknown) => {
           this.removeConsumer(consumerId);
-          clearTimeout(timoutRef);
+          clearTimeout(timeoutRef);
+
+          // prevent dangling closure living variable
+          (consumerId as any) = undefined;
 
           resolve({ result: resultData, hookId: context.current?.currentEvent?.hookId });
         };
@@ -88,6 +98,7 @@ export class TypedBusClass {
         } else {
           event.addPublishedTransport(result.value.transport);
         }
+
         result.value.publishedConsumers?.forEach((value) => {
           if (value.status == 'rejected') {
             console.error(`Error in consumer named "${value.reason.execName}"`, value.reason);
@@ -148,14 +159,30 @@ export class TypedBusClass {
     contract: iots.Any,
     exec: (...args: any[]) => any,
     options: { listenTo?: string[]; hookId?: string } = {},
+  ) {
+    return this._addConsumer(contract, exec, options);
+  }
+
+  private _addConsumer(
+    contract: iots.Any,
+    exec: (...args: any[]) => any,
+    options: { listenTo?: string[]; hookId?: string } = {},
+    skipStackLevel = 0,
   ) {
     const consumerId = generateConsumerId();
     const transportsList = options.listenTo?.length ? options.listenTo : ['internal'];
+    const site = callsites()[5 + skipStackLevel];
 
     let orphanConsumer = true;
     this.transports.forEach((transport) => {
       if (transportsList.includes(transport.name)) {
-        transport.addConsumer(contract, exec, consumerId, options.hookId);
+        transport.addConsumer(
+          contract,
+          exec,
+          consumerId,
+          site.getFileName()?.split('/').pop() + ':' + site.getLineNumber(),
+          options.hookId,
+        );
         orphanConsumer = false;
       }
 

src/engine/utils.ts

@@ -1,4 +1,4 @@
-export const deepFreeze = (objectToFreeze: any = {}, uuid?: string) => {
+export const deepFreeze = <T extends object>(objectToFreeze: T = {} as any, uuid?: string): T => {
   if (objectToFreeze instanceof Array) {
     objectToFreeze.forEach((item) => deepFreeze(item, uuid));
   } else {

src/example/playground.ts

@@ -9,7 +9,7 @@ class Dump extends DumpController {
     super(1, 'single');
   }
   dump(event: Event<any>): Promise<void> {
-    console.log({ dumpedOrphanEvent: event.toJSON() });
+    console.log({ dumpedOrphanEvent: event });
     return Promise.resolve();
   }
 
@@ -18,7 +18,7 @@ class Dump extends DumpController {
   }
 }
 
-TypedBus.setEventsDumpController(new Dump(), 'orphan');
+// TypedBus.setEventsDumpController(new Dump(), 'used');
 class ConsumerTest {
   @Consume(iots.type({ name: iots.string, age: iots.number }))
   async justConsumer(data: any) {
@@ -38,16 +38,18 @@ class ConsumerTest {
     await TypedBus.publish({ some: 'event', is: 'orphan' });
     await TypedBus.publish({ hookProp: 'value', hookValue: 123 });
     console.log(TypedBus.orphanEventsStore);
+    console.log(TypedBus.usedEventsStore);
   }
 }
 
 new ConsumerTest();
 
-setTimeout(() => {
+setTimeout(async () => {
   TypedBus.publish({ name: 'test', age: 4 });
+  console.log('published event');
 }, 1500);
 
 setTimeout(() => {
   console.log('Exit');
-  TypedBus.stopDumpers();
+  TypedBus.publish({ name: 'test', age: 2 });
 }, 3000);

src/graph/storage.ts

@@ -0,0 +1,34 @@
+/* eslint-disable max-classes-per-file */
+
+import LRUCache from 'lru-cache';
+
+import { Event } from '../engine/event';
+
+/**
+ *
+ * All events that have no `rootUUID` are considered root events.
+ * Root events are the first events of a chain of events.
+ * They are the events that are published by the user.
+ * All the events that have `launchedFrom` are considered child events.
+ * Child events are the events that are published by the consumers.
+ * Child events are grouped by same `launchedFrom` and `rootUUID`.
+ * Child events may considered as a chain of events.
+ * Build a tree that contains all the events and the structure of the chain of events.
+ */
+
+const cache = new LRUCache({ max: 1000000, ttl: 1000 });
+export class GraphStorage {
+  saveAndCommit(event: Event[]) {
+    if (event.rootUUID) {
+      const rootEvent = this.get(event.rootUUID);
+      if (rootEvent) {
+        rootEvent.addChild(event);
+        this.save(rootEvent);
+      } else {
+        this.save(event);
+      }
+    } else {
+      this.save(event);
+    }
+  }
+}

src/index.ts

@@ -1,5 +1,5 @@
 export { Event } from './engine/event';
 export { TypedBus } from './engine/instance';
-export { Transport, ConsumerMethod } from './engine/transport';
+export { Transport, ConsumerSignature as ConsumerMethod } from './engine/transport';
 export { DumpController } from './engine/dump-controller';
 export { Consume } from './decorators/consume';

yarn.lock

@@ -1166,6 +1166,11 @@
   resolved "https://registry.yarnpkg.com/@types/node/-/node-17.0.35.tgz#635b7586086d51fb40de0a2ec9d1014a5283ba4a"
   integrity sha512-vu1SrqBjbbZ3J6vwY17jBs8Sr/BKA+/a/WtjRG+whKg1iuLFOosq872EXS0eXWILdO36DHQQeku/ZcL6hz2fpg==
 
+"@types/node@^18.6.2":
+  version "18.6.2"
+  resolved "https://registry.yarnpkg.com/@types/node/-/node-18.6.2.tgz#ffc5f0f099d27887c8d9067b54e55090fcd54126"
+  integrity sha512-KcfkBq9H4PI6Vpu5B/KoPeuVDAbmi+2mDBqGPGUgoL7yXQtcWGu2vJWmmRkneWK3Rh0nIAX192Aa87AqKHYChQ==
+
 "@types/normalize-package-data@^2.4.0":
   version "2.4.1"
   resolved "https://registry.yarnpkg.com/@types/normalize-package-data/-/normalize-package-data-2.4.1.tgz#d3357479a0fdfdd5907fe67e17e0a85c906e1301"