5.0.41

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@push.rocks/smartdata",
-  "version": "5.0.30",
+  "version": "5.0.41",
   "private": false,
   "description": "do more with data",
   "main": "dist_ts/index.js",
@@ -22,27 +22,26 @@
   },
   "homepage": "https://gitlab.com/pushrocks/smartdata#README",
   "dependencies": {
-    "@push.rocks/lik": "^6.0.5",
+    "@push.rocks/lik": "^6.0.14",
     "@push.rocks/smartdelay": "^3.0.1",
     "@push.rocks/smartlog": "^3.0.2",
     "@push.rocks/smartmongo": "^2.0.10",
     "@push.rocks/smartpromise": "^4.0.2",
-    "@push.rocks/smartrx": "^3.0.6",
+    "@push.rocks/smartrx": "^3.0.7",
-    "@push.rocks/smartstring": "^4.0.7",
+    "@push.rocks/smartstring": "^4.0.15",
-    "@push.rocks/smarttime": "^4.0.5",
+    "@push.rocks/smarttime": "^4.0.6",
-    "@push.rocks/smartunique": "^3.0.6",
+    "@push.rocks/smartunique": "^3.0.8",
-    "@push.rocks/taskbuffer": "^3.1.5",
+    "@push.rocks/taskbuffer": "^3.1.7",
-    "@tsclass/tsclass": "^4.0.42",
+    "@tsclass/tsclass": "^4.0.52",
-    "mongodb": "^5.7.0"
+    "mongodb": "^6.5.0"
   },
   "devDependencies": {
     "@gitzone/tsbuild": "^2.1.66",
     "@gitzone/tsrun": "^1.2.44",
     "@gitzone/tstest": "^1.0.77",
-    "@push.rocks/qenv": "^6.0.2",
+    "@push.rocks/qenv": "^6.0.5",
-    "@push.rocks/tapbundle": "^5.0.15",
+    "@push.rocks/tapbundle": "^5.0.22",
-    "@types/node": "^20.5.0",
+    "@types/node": "^20.11.30"
-    "@types/shortid": "0.0.29"
   },
   "files": [
     "ts/**/*",

test/test.distributedcoordinator.ts

@@ -55,6 +55,7 @@ tap.test('SmartdataDistributedCoordinator should handle distributed task request
 
     const mockTaskRequest: taskbuffer.distributedCoordination.IDistributedTaskRequest = {
         submitterId: "mockSubmitter12345",    // Some unique mock submitter ID
+        requestResponseId: 'uni879873462hjhfkjhsdf', // Some unique ID for the request-response
         taskName: "SampleTask",
         taskVersion: "1.0.0",                  // Assuming it's a version string
         taskExecutionTime: Date.now(),
@@ -105,7 +106,7 @@ tap.test('should elect only one leader amongst multiple instances', async (tools
 
 tap.test('should clean up', async () => {
     await smartmongoInstance.stopAndDumpToDir(`.nogit/testdata/`);
-    setTimeout(() => process.exit(0), 1000);
+    setTimeout(() => process.exit(), 2000);
 })
 
 tap.start({ throwOnError: true });

test/test.ts

@@ -72,6 +72,11 @@ class Car extends smartdata.SmartDataDbDoc<Car, Car> {
   }
 }
 
+tap.test('should create a new id', async () => {
+  const newid = await Car.getNewId();
+  console.log(newid);
+})
+
 tap.test('should save the car to the db', async (toolsArg) => {
   const myCar = new Car('red', 'Volvo');
   await myCar.save();

ts/00_commitinfo_data.ts

@@ -3,6 +3,6 @@
  */
 export const commitinfo = {
   name: '@push.rocks/smartdata',
-  version: '5.0.30',
+  version: '5.0.41',
   description: 'do more with data'
 }

ts/smartdata.classes.collection.ts

@@ -26,7 +26,8 @@ const collectionFactory = new CollectionFactory();
  */
 export function Collection(dbArg: SmartdataDb | TDelayed<SmartdataDb>) {
   return function classDecorator<T extends { new (...args: any[]): {} }>(constructor: T) {
-    return class extends constructor {
+    const decoratedClass = class extends constructor {
+      public static className = constructor.name;
       public static get collection() {
         if (!(dbArg instanceof SmartdataDb)) {
           dbArg = dbArg();
@@ -40,6 +41,7 @@ export function Collection(dbArg: SmartdataDb | TDelayed<SmartdataDb>) {
         return collectionFactory.getCollection(constructor.name, dbArg);
       }
     };
+    return decoratedClass;
   };
 }
 
@@ -56,9 +58,10 @@ export const setDefaultManagerForDoc = <T>(managerArg: IManager, dbDocArg: T): T
  * This is a decorator that will tell the decorated class what dbTable to use
  * @param dbArg
  */
-export function Manager<TManager extends IManager>(managerArg?: TManager | TDelayed<TManager>) {
+export function managed<TManager extends IManager>(managerArg?: TManager | TDelayed<TManager>) {
   return function classDecorator<T extends { new (...args: any[]): any }>(constructor: T) {
-    return class extends constructor {
+    const decoratedClass = class extends constructor {
+      public static className = constructor.name;
       public static get collection() {
         let dbArg: SmartdataDb;
         if (!managerArg) {
@@ -106,9 +109,15 @@ export function Manager<TManager extends IManager>(managerArg?: TManager | TDela
         return manager;
       }
     };
+    return decoratedClass;
   };
 }
 
+/**
+ * @dpecrecated use @managed instead
+ */
+export const Manager = managed;
+
 export class SmartdataCollection<T> {
   /**
    * the collection that is used

ts/smartdata.classes.distributedcoordinator.ts

@@ -1,9 +1,10 @@
 import * as plugins from './smartdata.plugins.js';
 import { SmartdataDb } from './smartdata.classes.db.js';
-import { Manager, setDefaultManagerForDoc } from './smartdata.classes.collection.js';
+import { managed, setDefaultManagerForDoc } from './smartdata.classes.collection.js';
 import { SmartDataDbDoc, svDb, unI } from './smartdata.classes.doc.js';
+import { SmartdataDbWatcher } from './smartdata.classes.watcher.js';
 
-@Manager()
+@managed()
 export class DistributedClass extends SmartDataDbDoc<DistributedClass, DistributedClass> {
   // INSTANCE
   @unI()
@@ -39,6 +40,7 @@ export class SmartdataDistributedCoordinator extends plugins.taskbuffer.distribu
   public db: SmartdataDb;
   private asyncExecutionStack = new plugins.lik.AsyncExecutionStack();
   public ownInstance: DistributedClass;
+  public distributedWatcher: SmartdataDbWatcher<DistributedClass>;
 
   constructor(dbArg: SmartdataDb) {
     super();
@@ -54,6 +56,9 @@ export class SmartdataDistributedCoordinator extends plugins.taskbuffer.distribu
 
   public async stop() {
     await this.asyncExecutionStack.getExclusiveExecutionSlot(async () => {
+      if (this.distributedWatcher) {
+        await this.distributedWatcher.close();
+      }
       if (this.ownInstance?.data.elected) {
         this.ownInstance.data.elected = false;
       }
@@ -208,12 +213,30 @@ export class SmartdataDistributedCoordinator extends plugins.taskbuffer.distribu
    * the leading is implemented here
    */
   public async leadFunction() {
-    const watcher = await DistributedClass.watch({});
+    this.distributedWatcher = await DistributedClass.watch({});
-    watcher.changeSubject.subscribe({
+
+    const currentTaskRequests: Array<{
+      taskName: string;
+      taskExecutionTime: number;
+      /**
+       * all instances that requested this task
+       */
+      requestingDistibutedInstanceIds: string[];
+      responseTimeout: plugins.smartdelay.Timeout<any>;
+    }> = [];
+
+    this.distributedWatcher.changeSubject.subscribe({
       next: async (distributedDoc) => {
+        if (!distributedDoc) {
+          console.log(`registered deletion of instance...`);
+          return;
+        }
+        console.log(distributedDoc);
+        console.log(`registered change for ${distributedDoc.id}`);
         distributedDoc;
       },
     });
+
     while (this.ownInstance.data.status !== 'stopped' && this.ownInstance.data.elected) {
       const allInstances = await DistributedClass.getInstances({});
       for (const instance of allInstances) {
@@ -239,8 +262,18 @@ export class SmartdataDistributedCoordinator extends plugins.taskbuffer.distribu
       await this.ownInstance.save();
     });
     await plugins.smartdelay.delayFor(10000);
-
+    const result = await this.asyncExecutionStack.getExclusiveExecutionSlot(async () => {
-    return null;
+      await this.ownInstance.updateFromDb();
+      const taskRequestResult = this.ownInstance.data.taskRequestResults.find((resultItem) => {
+        return resultItem.requestResponseId === taskRequestArg.requestResponseId;
+      });
+      return taskRequestResult;
+    });
+    if (!result) {
+      console.warn('no result found for task request...');
+      return null;
+    }
+    return result;
   }
 
   public async updateDistributedTaskRequest(

ts/smartdata.classes.doc.ts

@@ -127,6 +127,13 @@ export class SmartDataDbDoc<T extends TImplements, TImplements, TManager extends
     }
   }
 
+  /**
+   * get a unique id prefixed with the class name
+   */
+  public static async getNewId<T = any>(this: plugins.tsclass.typeFest.Class<T>, lengthArg: number = 20) {
+    return `${(this as any).className}:${plugins.smartunique.shortId(lengthArg)}`;
+  }
+
   /**
    * get cursor
    * @returns

ts/smartdata.classes.easystore.ts

@@ -37,7 +37,16 @@ export class EasyStore<T> {
     this.nameId = nameIdArg;
   }
 
-  private async getEasyStore() {
+  private easyStorePromise: Promise<InstanceType<typeof this.easyStoreClass>>;
+  private async getEasyStore(): Promise<InstanceType<typeof this.easyStoreClass>> {
+    if (this.easyStorePromise) {
+      return this.easyStorePromise;
+    };
+
+    // first run from here
+    const deferred = plugins.smartpromise.defer<InstanceType<typeof this.easyStoreClass>>();
+    this.easyStorePromise = deferred.promise;
+
     let easyStore = await this.easyStoreClass.getInstance({
       nameId: this.nameId,
     });
@@ -48,7 +57,8 @@ export class EasyStore<T> {
       easyStore.data = {};
       await easyStore.save();
     }
-    return easyStore;
+    deferred.resolve(easyStore);
+    return this.easyStorePromise;
   }
 
   /**
@@ -70,7 +80,7 @@ export class EasyStore<T> {
   /**
    * writes a specific key to the keyValueStore
    */
-  public async writeKey(keyArg: keyof T, valueArg: any) {
+  public async writeKey<TKey extends keyof T>(keyArg: TKey, valueArg: T[TKey]) {
     const easyStore = await this.getEasyStore();
     easyStore.data[keyArg] = valueArg;
     await easyStore.save();

ts/smartdata.classes.watcher.ts

@@ -17,9 +17,13 @@ export class SmartdataDbWatcher<T = any> {
     smartdataDbDocArg: typeof SmartDataDbDoc
   ) {
     this.changeStream = changeStreamArg;
-    this.changeStream.on('change', async (item: T) => {
+    this.changeStream.on('change', async (item: any) => {
+      if (!item.fullDocument) {
+        this.changeSubject.next(null);
+        return;
+      }
       this.changeSubject.next(
-        smartdataDbDocArg.createInstanceFromMongoDbNativeDoc(item) as any as T
+        smartdataDbDocArg.createInstanceFromMongoDbNativeDoc(item.fullDocument) as any as T
       );
     });
     plugins.smartdelay.delayFor(0).then(() => {