5.0.33

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@push.rocks/smartdata",
-  "version": "5.0.27",
+  "version": "5.0.33",
   "private": false,
   "description": "do more with data",
   "main": "dist_ts/index.js",
@@ -30,8 +30,8 @@
     "@push.rocks/smartrx": "^3.0.6",
     "@push.rocks/smartstring": "^4.0.7",
     "@push.rocks/smarttime": "^4.0.5",
-    "@push.rocks/smartunique": "^3.0.3",
-    "@push.rocks/taskbuffer": "^3.1.3",
+    "@push.rocks/smartunique": "^3.0.6",
+    "@push.rocks/taskbuffer": "^3.1.5",
     "@tsclass/tsclass": "^4.0.42",
     "mongodb": "^5.7.0"
   },

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(),
@@ -75,6 +76,7 @@ tap.test('SmartdataDistributedCoordinator should update 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(),
@@ -104,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 });

ts/00_commitinfo_data.ts

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

ts/smartdata.classes.collection.ts

@@ -269,7 +269,7 @@ export class SmartdataCollection<T> {
    * if this.objectValidation is not set it passes.
    */
   private checkDoc(docArg: T): Promise<void> {
-    const done = plugins.smartq.defer<void>();
+    const done = plugins.smartpromise.defer<void>();
     let validationResult = true;
     if (this.objectValidation) {
       validationResult = this.objectValidation(docArg);

ts/smartdata.classes.distributedcoordinator.ts

@@ -2,6 +2,7 @@ import * as plugins from './smartdata.plugins.js';
 import { SmartdataDb } from './smartdata.classes.db.js';
 import { Manager, setDefaultManagerForDoc } from './smartdata.classes.collection.js';
 import { SmartDataDbDoc, svDb, unI } from './smartdata.classes.doc.js';
+import { SmartdataDbWatcher } from './smartdata.classes.watcher.js';
 
 @Manager()
 export class DistributedClass extends SmartDataDbDoc<DistributedClass, DistributedClass> {
@@ -11,7 +12,7 @@ export class DistributedClass extends SmartDataDbDoc<DistributedClass, Distribut
 
   @svDb()
   public data: {
-    status: 'bidding' | 'settled' | 'initializing' | 'stopped';
+    status: 'initializing' | 'bidding' | 'settled' | 'stopped';
     biddingShortcode?: string;
     biddingStartTime?: number;
     lastUpdated: number;
@@ -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;
       }
@@ -174,7 +179,7 @@ export class SmartdataDistributedCoordinator extends plugins.taskbuffer.distribu
       });
       console.log(`bidding for leadership...`);
       await plugins.smartdelay.delayFor(
-        plugins.smarttime.getMilliSecondsFromUnits({ seconds: 20 })
+        plugins.smarttime.getMilliSecondsFromUnits({ seconds: 5 })
       );
       await this.asyncExecutionStack.getExclusiveExecutionSlot(async () => {
         let biddingInstances = await DistributedClass.getInstances({});
@@ -182,7 +187,7 @@ export class SmartdataDistributedCoordinator extends plugins.taskbuffer.distribu
           (instanceArg) =>
             instanceArg.data.status === 'bidding' &&
             instanceArg.data.lastUpdated >=
-              Date.now() - plugins.smarttime.getMilliSecondsFromUnits({ seconds: 25 })
+              Date.now() - plugins.smarttime.getMilliSecondsFromUnits({ seconds: 10 })
         );
         console.log(`found ${biddingInstances.length} bidding instances...`);
         this.ownInstance.data.elected = true;
@@ -208,12 +213,30 @@ export class SmartdataDistributedCoordinator extends plugins.taskbuffer.distribu
    * the leading is implemented here
    */
   public async leadFunction() {
-    const watcher = await DistributedClass.watch({});
-    watcher.changeSubject.subscribe({
+    this.distributedWatcher = await DistributedClass.watch({});
+
+    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) {
@@ -234,11 +257,24 @@ export class SmartdataDistributedCoordinator extends plugins.taskbuffer.distribu
         console.error('instance need to be started first...');
         return;
       }
+      await this.ownInstance.updateFromDb();
       this.ownInstance.data.taskRequests.push(taskRequestArg);
       await this.ownInstance.save();
     });
+    await plugins.smartdelay.delayFor(10000);
+    const result = await this.asyncExecutionStack.getExclusiveExecutionSlot(async () => {
+      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(
     infoBasisArg: plugins.taskbuffer.distributedCoordination.IDistributedTaskRequest

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(() => {

ts/smartdata.plugins.ts

@@ -8,7 +8,6 @@ import * as lik from '@push.rocks/lik';
 import * as smartdelay from '@push.rocks/smartdelay';
 import * as smartlog from '@push.rocks/smartlog';
 import * as smartpromise from '@push.rocks/smartpromise';
-import * as smartq from '@push.rocks/smartpromise';
 import * as smartrx from '@push.rocks/smartrx';
 import * as smartstring from '@push.rocks/smartstring';
 import * as smarttime from '@push.rocks/smarttime';
@@ -21,7 +20,6 @@ export {
   smartdelay,
   smartpromise,
   smartlog,
-  smartq,
   smartrx,
   mongodb,
   smartstring,