fix(core): update

package.json

@@ -22,18 +22,20 @@
   },
   "homepage": "https://gitlab.com/pushrocks/smartdata#README",
   "dependencies": {
-    "@pushrocks/lik": "^6.0.0",
+    "@pushrocks/lik": "^6.0.2",
     "@pushrocks/smartdelay": "^2.0.13",
     "@pushrocks/smartlog": "^3.0.1",
     "@pushrocks/smartmongo": "^2.0.7",
     "@pushrocks/smartpromise": "^3.1.7",
     "@pushrocks/smartrx": "^3.0.0",
     "@pushrocks/smartstring": "^4.0.2",
+    "@pushrocks/smarttime": "^4.0.1",
     "@pushrocks/smartunique": "^3.0.3",
-    "@tsclass/tsclass": "^4.0.19",
+    "@pushrocks/taskbuffer": "^3.0.10",
-    "@types/lodash": "^4.14.188",
+    "@tsclass/tsclass": "^4.0.28",
+    "@types/lodash": "^4.14.191",
     "lodash": "^4.17.21",
-    "mongodb": "^4.9.1"
+    "mongodb": "^4.13.0"
   },
   "devDependencies": {
     "@gitzone/tsbuild": "^2.1.65",
@@ -41,7 +43,7 @@
     "@gitzone/tstest": "^1.0.74",
     "@pushrocks/qenv": "^5.0.2",
     "@pushrocks/tapbundle": "^5.0.4",
-    "@types/node": "^18.7.16",
+    "@types/node": "^18.11.18",
     "@types/shortid": "0.0.29"
   },
   "files": [

test/test.ts

@@ -100,7 +100,7 @@ tap.test('should save the car to the db', async (toolsArg) => {
 });
 
 tap.test('expect to get instance of Car with shallow match', async () => {
-  const totalQueryCycles = totalCars / 6;
+  const totalQueryCycles = totalCars / 2;
   let counter = 0;
   do {
     const timeStart = Date.now();

ts/00_commitinfo_data.ts

@@ -3,6 +3,6 @@
  */
 export const commitinfo = {
   name: '@pushrocks/smartdata',
-  version: '5.0.10',
+  version: '5.0.11',
   description: 'do more with data'
 }

ts/index.ts

@@ -4,7 +4,13 @@ export * from './smartdata.classes.doc.js';
 export * from './smartdata.classes.easystore.js';
 export * from './smartdata.classes.cursor.js';
 
+import * as convenience from './smartadata.convenience.js';
+
+export {
+  convenience
+}
+
 // to be removed with the next breaking update
-import * as plugins from './smartdata.plugins.js';
+import type * as plugins from './smartdata.plugins.js';
 type IMongoDescriptor = plugins.tsclass.database.IMongoDescriptor;
 export type { IMongoDescriptor };

ts/smartadata.convenience.ts

@@ -0,0 +1,5 @@
+import * as plugins from './smartdata.plugins.js';
+
+export const getNewUniqueId = async (prefixArg?: string) => {
+  return plugins.smartunique.uni(prefixArg);
+}

ts/smartdata.classes.db.ts

@@ -16,6 +16,7 @@ export class SmartdataDb {
   mongoDbClient: plugins.mongodb.MongoClient;
   mongoDb: plugins.mongodb.Db;
   status: TConnectionStatus;
+  statusConnectedDeferred = plugins.smartpromise.defer();
   smartdataCollectionMap = new ObjectMap<SmartdataCollection<any>>();
 
   constructor(smartdataOptions: plugins.tsclass.database.IMongoDescriptor) {
@@ -51,6 +52,7 @@ export class SmartdataDb {
     });
     this.mongoDb = this.mongoDbClient.db(this.smartdataOptions.mongoDbName);
     this.status = 'connected';
+    this.statusConnectedDeferred.resolve();
     console.log(`Connected to database ${this.smartdataOptions.mongoDbName}`);
   }
 

ts/smartdata.classes.distributedcoordinator.ts

@@ -0,0 +1,201 @@
+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';
+
+@Manager()
+class DistributedClass extends SmartDataDbDoc<DistributedClass, DistributedClass> {
+  // INSTANCE
+  @unI()
+  public id: string;
+
+  @svDb()
+  public data: {
+    status: 'bidding' | 'settled' | 'initializing' | 'stopped';
+    biddingShortcode?: string;
+    lastUpdated: number;
+    elected: boolean;
+    /**
+     * used to store request
+     */
+    taskRequests: plugins.taskbuffer.distributedCoordination.IDistributedTaskRequest[],
+  
+    /**
+     * only used by the leader to convey consultation results
+     */
+    taskRequestResults: plugins.taskbuffer.distributedCoordination.IDistributedTaskRequestResult[];
+  }
+}
+
+/**
+ * This file implements a distributed coordinator according to the @pushrocks/taskbuffer standard.
+ * you should not set up this yourself. Instead, there is a factory on the SmartdataDb class
+ * that will take care of setting this up.
+ */
+export class SmartdataDistributedCoordinator extends plugins.taskbuffer.distributedCoordination
+  .AbstractDistributedCoordinator {
+  public readyPromise: Promise<any>;
+  public db: SmartdataDb;
+  private asyncExecutionStack = new plugins.lik.AsyncExecutionStack();
+  private ownInstance: DistributedClass;
+
+  constructor(dbArg?: SmartdataDb) {
+    super();
+    this.db = dbArg;
+    setDefaultManagerForDoc(this, DistributedClass);
+    this.readyPromise = this.db.statusConnectedDeferred.promise;
+    this.init();
+  }
+
+  // smartdata specific stuff
+  public async start() {
+    await this.init();
+  }
+
+  public async stop() {
+    if (this.ownInstance?.data.elected) {
+      this.ownInstance.data.elected = false;
+    } else {
+      console.log(`can't stop a distributed instance that has not been started yet.`);
+    }
+  }
+
+  public id = plugins.smartunique.uni('distributedInstance');
+
+  private startHeartbeat = async () => {
+    while (this.ownInstance.data.status !== 'stopped') {
+      await this.asyncExecutionStack.getExclusiveExecutionSlot(async () => {
+        await this.ownInstance.updateFromDb();
+        this.ownInstance.data.lastUpdated = Date.now();
+        await this.ownInstance.save();
+      });
+      await plugins.smartdelay.delayFor(10000);
+    }
+  };
+  public async init() {
+    await this.readyPromise;
+    if (!this.ownInstance) {
+      await this.asyncExecutionStack.getExclusiveExecutionSlot(async () => {
+        this.ownInstance = new DistributedClass();
+        this.ownInstance.id = this.id;
+        this.ownInstance.data = {
+          elected: false,
+          lastUpdated: Date.now(),
+          status: 'initializing',
+          taskRequests: [],
+          taskRequestResults: []
+        };
+        await this.ownInstance.save();
+      });
+    }
+
+    // lets enable the heartbeat
+    this.startHeartbeat();
+
+    // lets do a leader check
+    await this.checkAndMaybeLead();
+
+    return this.ownInstance;
+  }
+
+  // --> leader election
+  public async checkAndMaybeLead() {
+    const allInstances = await DistributedClass.getInstances({});
+    let leader = allInstances.find((instanceArg) => instanceArg.data.elected === true);
+    if (
+      leader &&
+      leader.data.lastUpdated >=
+        Date.now() - plugins.smarttime.getMilliSecondsFromUnits({ minutes: 1 })
+    ) {
+      await this.asyncExecutionStack.getExclusiveExecutionSlot(async () => {
+        await this.ownInstance.updateFromDb();
+        this.ownInstance.data.status = 'settled';
+        await this.ownInstance.save();
+      });
+      return;
+    } else {
+      await this.asyncExecutionStack.getExclusiveExecutionSlot(async () => {
+        this.ownInstance.data.status = 'bidding';
+        this.ownInstance.data.biddingShortcode = plugins.smartunique.shortId();
+        await this.ownInstance.save();
+      });
+      await plugins.smartdelay.delayFor(plugins.smarttime.getMilliSecondsFromUnits({ minutes: 2 }));
+      await this.asyncExecutionStack.getExclusiveExecutionSlot(async () => {
+        let biddingInstances = await DistributedClass.getInstances({});
+        biddingInstances = biddingInstances.filter(
+          (instanceArg) =>
+            !instanceArg.data.elected &&
+            instanceArg.data.lastUpdated >=
+              Date.now() - plugins.smarttime.getMilliSecondsFromUnits({ minutes: 1 })
+        );
+        this.ownInstance.data.elected = true;
+        for (const biddingInstance of biddingInstances) {
+          if (biddingInstance.data.biddingShortcode < this.ownInstance.data.biddingShortcode) {
+            this.ownInstance.data.elected = false;
+          }
+        }
+        await this.ownInstance.save();
+      });
+      if (this.ownInstance.data.elected) {
+        this.leadFunction();
+      }
+    }
+  }
+
+  /**
+   * when it has been determined
+   * that this instance is leading
+   * the leading is implemented here
+   */
+  public async leadFunction() {
+    const ownInstance = await this.init();
+    const watcher = await DistributedClass.watch({});
+    /**
+     * this function is started once per unique job request
+     */
+    const startResultTimer = async () => {
+
+    }
+    
+    watcher.changeSubject.subscribe({
+      next: async (distributedDoc) => {
+        distributedDoc
+      }
+    })
+    while (this.ownInstance.data.status !== 'stopped') {
+      await plugins.smartdelay.delayFor(1000);
+    }
+  };
+
+  // abstract implemented methods
+  public async fireDistributedTaskRequest(
+    taskRequestArg: plugins.taskbuffer.distributedCoordination.IDistributedTaskRequest
+  ): Promise<plugins.taskbuffer.distributedCoordination.IDistributedTaskRequestResult> {
+    const ownInstance = await this.init();
+    await this.asyncExecutionStack.getExclusiveExecutionSlot(async () => {
+      this.ownInstance.data.taskRequests.push(taskRequestArg);
+      await this.ownInstance.save();
+    });
+    return null;
+  }
+
+  public async updateDistributedTaskRequest(
+    infoBasisArg: plugins.taskbuffer.distributedCoordination.IDistributedTaskRequest
+  ): Promise<void> {
+    await this.asyncExecutionStack.getExclusiveExecutionSlot(async () => {
+      const existingInfoBasis = this.ownInstance.data.taskRequests.find(infoBasisItem => {
+        return infoBasisItem.taskName === infoBasisArg.taskName
+          && infoBasisItem.taskExecutionTime === infoBasisArg.taskExecutionTime;
+      });
+      Object.assign(existingInfoBasis, infoBasisArg);
+      await this.ownInstance.save();
+      plugins.smartdelay.delayFor(60000).then(() => {
+        this.asyncExecutionStack.getExclusiveExecutionSlot(async () => {
+          const indexToRemove = this.ownInstance.data.taskRequests.indexOf(existingInfoBasis);
+          this.ownInstance.data.taskRequests.splice(indexToRemove, indexToRemove);
+          await this.ownInstance.save();
+        })
+      })
+    });
+  }
+}

ts/smartdata.classes.doc.ts

@@ -255,6 +255,16 @@ export class SmartDataDbDoc<T extends TImplements, TImplements, TManager extends
     }
   }
 
+  /**
+   * updates an object from db
+   */
+  public async updateFromDb() {
+    const mongoDbNativeDoc = await this.collection.findOne(this.createIdentifiableObject());
+    for (const key of Object.keys(mongoDbNativeDoc)) {
+      this[key] = mongoDbNativeDoc[key];
+    }
+  }
+
   /**
    * creates a saveable object so the instance can be persisted as json in the database
    */

ts/smartdata.classes.easystore.ts

@@ -17,6 +17,15 @@ export class EasyStore<T> {
       @unI()
       public nameId: string;
 
+      @svDb()
+      public ephermal: {
+        activated: boolean;
+        timeout: number;
+      };
+
+      @svDb()
+      lastEdit: number;
+
       @svDb()
       public data: Partial<T>;
     }
@@ -90,4 +99,10 @@ export class EasyStore<T> {
     easyStore.data = {};
     await easyStore.save();
   }
+
+  public async cleanUpEphermal() {
+    while(await this.smartdataDbRef.statusConnectedDeferred.promise && this.smartdataDbRef.status === 'connected') {
+      
+    }
+  }
 }

ts/smartdata.plugins.ts

@@ -4,17 +4,21 @@ import * as tsclass from '@tsclass/tsclass';
 export { tsclass };
 
 // @pushrocks scope
-import * as smartlog from '@pushrocks/smartlog';
+import * as lik from '@pushrocks/lik';
-import * as lodash from 'lodash';
-import * as mongodb from 'mongodb';
 import * as smartdelay from '@pushrocks/smartdelay';
+import * as smartlog from '@pushrocks/smartlog';
 import * as smartpromise from '@pushrocks/smartpromise';
 import * as smartq from '@pushrocks/smartpromise';
 import * as smartrx from '@pushrocks/smartrx';
 import * as smartstring from '@pushrocks/smartstring';
+import * as smarttime from '@pushrocks/smarttime';
 import * as smartunique from '@pushrocks/smartunique';
+import * as taskbuffer from '@pushrocks/taskbuffer';
+import * as lodash from 'lodash';
+import * as mongodb from 'mongodb';
 
 export {
+  lik,
   smartdelay,
   smartpromise,
   smartlog,
@@ -23,5 +27,7 @@ export {
   smartrx,
   mongodb,
   smartstring,
+  smarttime,
   smartunique,
+  taskbuffer,
 };