fix relations for related remote objects

Author: tysonrm

src/domain/make-relations.js

@@ -5,7 +5,7 @@
  */
 
 import domainEvents from './domain-events'
-import { importRemoteCache } from './import-remotes'
+import { importModelCache, importRemoteCache } from './import-remotes'
 
 const { internalCacheRequest, internalCacheResponse, externalCacheRequest } =
   domainEvents
@@ -57,7 +57,7 @@ export const relationType = {
    * @param {*} rel
    * @returns
    */
-  oneToOne(model, ds, rel) {
+  oneToOne (model, ds, rel) {
     return this.manyToOne(model, ds, rel)
   },
 
@@ -87,14 +87,14 @@ export const relationType = {
     const rds = ds.factory.getRestrictedDataSource(model.modelName)
     const relRds = ds.factory.getRestrictedDataSource(rel.modelName, {
       isCached: true,
-      ephemeral: true,
+      ephemeral: true
     })
 
     // if relRds is in the same domain but is remote, this fails
     if (rds.namespace !== relRds.namespace) return null
 
     return rds[rel.name]({ args, model, ds: relRds, relation: rel })
-  },
+  }
 }
 
 /**
@@ -108,39 +108,39 @@ const updateForeignKeys = {
    * @param {import('./index').relations[x]} relation
    * @param {import('./model-factory').Datasource} ds
    */
-  async [relationType.manyToOne.name](fromModel, toModels, relation, ds) {
+  async [relationType.manyToOne.name] (fromModel, toModels, relation, ds) {
     return fromModel.update(
       { [relation.foreignKey]: toModels[0].getId() },
       false
     )
   },
 
-  async [relationType.oneToOne.name](fromModel, toModels, relation, ds) {
+  async [relationType.oneToOne.name] (fromModel, toModels, relation, ds) {
     return this[relationType.manyToOne.name](fromModel, toModels, relation, ds)
   },
 
-  async [relationType.oneToMany.name](fromModel, toModels, relation, ds) {
+  async [relationType.oneToMany.name] (fromModel, toModels, relation, ds) {
     return Promise.all(
       toModels.map(async m => {
         const model = await ds.find(m.id || m.getId())
         return ds.save(m.id, {
           ...model,
-          [relation.foreignKey]: fromModel.getId(),
+          [relation.foreignKey]: fromModel.getId()
         })
       })
     )
   },
 
-  async [relationType.containsMany.name](fromModel, toModels, relation, ds) {
+  async [relationType.containsMany.name] (fromModel, toModels, relation, ds) {
     toModels.map(model =>
       model.update({ [relation.foreignKey]: fromModel.getId() })
     )
   },
 
-  async [relationType.custom.name](fromModel, toModels, relation, ds) {
+  async [relationType.custom.name] (fromModel, toModels, relation, ds) {
     const customFn = fromModel[`${relation}UpdateForeignKeys`]
     if (customFn === 'function') customFn(toModels, relation, ds)
-  },
+  }
 }
 
 /**
@@ -151,9 +151,13 @@ const updateForeignKeys = {
  * @param {import('./datasource').default} ds
  * @returns
  */
-async function createNewModels(args, fromModel, relation, ds) {
+async function createNewModels (args, fromModel, relation, ds) {
   if (args.length > 0) {
-    const { UseCaseService, importRemoteCache } = require('.')
+    const {
+      UseCaseService,
+      importRemoteCache,
+      default: ModelFactory
+    } = require('.')
     const service = UseCaseService(relation.modelName.toUpperCase())
     const newModels = await Promise.all(
       args.map(arg => service.createModel(arg))
@@ -173,7 +177,7 @@ async function createNewModels(args, fromModel, relation, ds) {
  * @param {import("./event-broker").EventBroker} broker
  * @returns {Promise<import(".").Event>} source model
  */
-export function requireRemoteObject(model, relation, broker, ...args) {
+export function requireRemoteObject (model, relation, broker, ...args) {
   const request = internalCacheRequest(relation.modelName)
   const response = internalCacheResponse(relation.modelName)
 
@@ -193,7 +197,7 @@ export function requireRemoteObject(model, relation, broker, ...args) {
     modelId: id,
     relation,
     model,
-    args,
+    args
   }
 
   return new Promise(async function (resolve) {
@@ -209,7 +213,7 @@ export function requireRemoteObject(model, relation, broker, ...args) {
  * @param {import('.').relations[x]} relation
  * @returns {boolean}
  */
-function isRelatedModelLocal(relation) {
+function isRelatedModelLocal (relation) {
   return require('.')
     .default.getModelSpecs()
     .filter(spec => !spec.isCached)
@@ -222,7 +226,7 @@ function isRelatedModelLocal(relation) {
  * @param {import("./index").relations} relations
  * @param {import("./datasource").default} datasource
  */
-export default function makeRelations(relations, datasource, broker) {
+export default function makeRelations (relations, datasource, broker) {
   if (Object.getOwnPropertyNames(relations).length < 1) return
 
   return Object.keys(relations)
@@ -231,9 +235,8 @@ export default function makeRelations(relations, datasource, broker) {
 
       const rel = {
         ...relations[relation],
-        modelName: relModelName.toUpperCase(),
-        domain: require('.').default.getModelSpec(relModelName).domain,
-        name: relation,
+        modelName: relModelName,
+        name: relation
       }
 
       try {
@@ -245,15 +248,13 @@ export default function makeRelations(relations, datasource, broker) {
 
         return {
           // the relation function
-          async [relation](...args) {
+          async [relation] (...args) {
+            const local = isRelatedModelLocal(relModelName)
+            if (!local) await importModelCache(relModelName)
             // Get existing (or create temp) datasource of related object
-            const local = isRelatedModelLocal(rel)
             const createNew = args?.length > 0
 
-            if (!local) {
-              // the ds is for a remote object, fetch the code for it.
-              await importRemoteCache(rel.modelName)
-            } else if (createNew && rel.type !== 'custom') {
+            if (createNew && rel.type !== 'custom') {
               // fetch the local ds and create the models
               const ds = datasource.factory.getDataSource(rel.modelName)
               return await createNewModels(args, this, rel, ds)
@@ -262,7 +263,7 @@ export default function makeRelations(relations, datasource, broker) {
             // if the object is remote, we now have its code
             const ds = datasource.factory.getDataSource(
               rel.modelName,
-              rel.domain
+              require('.').default.getModelSpec(relModelName).domain
             )
 
             const models = await relationType[rel.type](this, ds, rel, args)
@@ -283,7 +284,7 @@ export default function makeRelations(relations, datasource, broker) {
             }
 
             return models
-          },
+          }
         }
       } catch (error) {
         console.error({ fn: makeRelations.name, error })

src/domain/thread-pool.js

@@ -262,7 +262,7 @@ export class ThreadPool extends EventEmitter {
 
           const errorFn = AsyncResource.bind(error => {
             pool.jobTime(job.stopTimer())
-            console.error({ fn: this.run.name, error })
+            console.error({ fn: this.run.name, msg: 'dead thread', error })
             unsubscribe('exit', exitFn)
             unsubscribe('message', messageFn)
             pool.incrementErrorCount()
@@ -303,6 +303,7 @@ export class ThreadPool extends EventEmitter {
         }
         console.log('aegis up', msg)
         pool.connectEventChannel(worker, eventChannel)
+        pool.threads.push(thread)
         resolve(thread)
       })
     })
@@ -321,7 +322,7 @@ export class ThreadPool extends EventEmitter {
       // call thread.run
       this.waitingJobs.shift()(thread)
     // return to pool
-    else this.freeThreads.push(thread)
+    else this.checkin(thread)
   }
 
   /**
@@ -335,11 +336,8 @@ export class ThreadPool extends EventEmitter {
       file: this.file,
       workerData: this.workerData
     })
-
-    if (thread) {
-      this.threads.push(thread)
-      return thread
-    }
+    
+    if (thread) return thread
 
     throw new Error('error creating thread')
   }
@@ -519,6 +517,24 @@ export class ThreadPool extends EventEmitter {
     )
   }
 
+  checkout () {
+    if (this.freeThreads.length > 0) {
+      const thread = this.freeThreads.shift()
+      const threadsInUse = this.threads.length - this.freeThreads.length
+      console.debug(`thread checked-out, total in-use now ${threadsInUse}`)
+      return thread
+    } 
+  }
+
+  checkin (thread) {
+    if (thread) {
+      this.freeThreads.push(thread)
+      const threadsInUse = this.threads.length - this.freeThreads.length
+      console.debug(`thread checked-in, total in-use now ${threadsInUse}`)
+    }
+    return this
+  }
+
   /**
    * Spin up a new thread if needed and available.
    */
@@ -553,7 +569,7 @@ export class ThreadPool extends EventEmitter {
         ...options
       })
 
-      let thread = this.freeThreads.shift()
+      let thread = this.checkout()
 
       if (!thread) thread = await this.allocate()