fix(external-jobs): replace '.cjs' with '.js' for runtime import

Author: Gerald Baulig

cfg/config.json

@@ -101,7 +101,7 @@
         "messageObject": "io.restorecommerce.job.JobList"
       },
       "deleteJobs": {
-        "messageObject": "io.restorecommerce.job.Deleted"
+        "messageObject": "io.restorecommerce.resourcebase.DeleteRequest"
       },
       "jobsCreated": {
         "messageObject": "io.restorecommerce.job.JobList"
@@ -110,7 +110,7 @@
         "messageObject": "io.restorecommerce.job.JobList"
       },
       "jobsDeleted": {
-        "messageObject": "io.restorecommerce.resourcebase.DeleteRequest"
+        "messageObject": "io.restorecommerce.resourcebase.Deleted"
       },
       "jobDone": {
         "messageObject": "io.restorecommerce.job.JobDone"
@@ -225,8 +225,8 @@
     }
   },
   "externalJobs": {
-    "sourcePath": "./lib/external-jobs",
-    "importPath": "./external-jobs"
+    "sourcePath": "./lib/external-jobs/",
+    "importPath": "./external-jobs/"
   },
   "queue": [
     {

test/cfg/config_test.json renamed to cfg/config_test.json

@@ -81,7 +81,7 @@ export const unmarshallProtobufAny = (data: any, logger: Logger): any => {
         unmarshalled = JSON.parse(decoded);
       }
     }
-  } catch (error) {
+  } catch (error: any) {
     logger.error('Error unmarshalling job payload', {
       data, code: error.code,
       message: error.message, stack: error.stack
@@ -119,13 +119,13 @@ export async function checkAccessRequest(ctx: GQLClientContext, resource: Resour
 
   let result: DecisionResponse | PolicySetRQResponse;
   try {
-    result = await accessRequest(subject, resource, action, ctx, { operation, roleScopingEntityURN: cfg?.get('authorization:urns:roleScopingEntityURN') });
-  } catch (err) {
+    result = await accessRequest(subject, resource, action, ctx, { operation, /* roleScopingEntityURN: cfg?.get('authorization:urns:roleScopingEntityURN') */ });
+  } catch (err: any) {
     return {
       decision: Response_Decision.DENY,
       operation_status: {
-        code: err.code || 500,
-        message: err.details || err.message,
+        code: Number.isInteger(err.code) ? err.code : 500,
+        message: err.details ?? err.message,
       }
     };
   }

src/schedulingService.ts

@@ -1,4 +1,3 @@
-import path from 'path';
 import * as _ from 'lodash-es';
 import * as chassis from '@restorecommerce/chassis-srv';
 import { Events, Topic, registerProtoMeta } from '@restorecommerce/kafka-client';
@@ -339,7 +338,7 @@ export class Worker {
     const externalJobDir = process.env.EXTERNAL_JOBS_DIR ?? cfg.get('externalJobs:sourcePath') ?? './lib/external-jobs';
     let externalJobFiles: Array<string>;
     try {
-      externalJobFiles = fs.readdirSync(process.env.EXTERNAL_JOBS_DIR ?? './lib/external-jobs');
+      externalJobFiles = fs.readdirSync(externalJobDir);
     } catch (err: any) {
       if (err.message.includes('no such file or directory')) {
         this.logger.info('No files for external job processors found');
@@ -350,23 +349,22 @@ export class Worker {
     if (externalJobFiles?.length > 0) {
       externalJobFiles.forEach(async (externalFile) => {
         if (externalFile.endsWith('.js') || externalFile.endsWith('.cjs')) {
-          const import_dir = process.env.EXTERNAL_JOBS_REQUIRE_DIR ?? cfg.get('externalJobs:importPath') ?? './external-jobs/';
-
+          const import_path = [
+            process.env.EXTERNAL_JOBS_REQUIRE_DIR ?? cfg.get('externalJobs:importPath') ?? './external-jobs/',
+            externalFile
+          ].join('').replace(/\.cjs$/, '.js');
           try {
-            const fileImport = await import(path.join(import_dir, externalFile));
+            const fileImport = await import(import_path);
+            this.logger?.info('imported:', fileImport);
             // check for double default
             if (fileImport?.default?.default) {
-              (async () => (await import(import_dir + externalFile)).default.default(cfg, logger, events, runWorker))().catch(err => {
-                this.logger.error(`Error scheduling external job ${externalFile}`, { err: err.message });
-              });
+              await fileImport.default.default(cfg, logger, events, runWorker);
             } else {
-              (async () => (await import(import_dir + externalFile)).default(cfg, logger, events, runWorker))().catch(err => {
-                this.logger.error(`Error scheduling external job ${externalFile}`, { err: err.message });
-              });
+              await fileImport.default(cfg, logger, events, runWorker);
             }
           }
           catch (err: any) {
-            this.logger.error(`Error scheduling external job ${externalFile}`, { err });
+            this.logger?.error(`Error scheduling external job ${import_path}`, { err });
           }
         }
       });

src/utilts.ts

@@ -150,7 +150,7 @@ const startGrpcMockServer = async (methodWithOutput: MethodWithOutput[]) => {
       oneofs: true
     });
     await mockServer.start();
-    logger.info('Mock ACS Server started on port 50061');
+    logger.info(`Mock ACS Server started on ${mockServer.serverAddress}`);
   } catch (err) {
     logger.error('Error starting mock ACS server', err);
   }
@@ -183,7 +183,7 @@ const startIDSGrpcMockServer = async (methodWithOutput: MethodWithOutput[]) => {
       oneofs: true
     });
     await mockServerIDS.start();
-    logger.info('Mock IDS Server started on port 50051');
+    logger.info(`Mock IDS Server started on ${mockServerIDS.serverAddress}`);
   } catch (err) {
     logger.error('Error starting mock IDS server', err);
   }
@@ -204,7 +204,6 @@ describe(`testing scheduling-srv ${testSuffix}: gRPC`, () => {
   let worker: Worker;
   let jobEvents: Topic;
   let grpcSchedulingSrv: SchedulingServiceClient;
-  let cfg: any;
 
   before(async function (): Promise<any> {
     this.timeout(40000);
@@ -285,7 +284,7 @@ describe(`testing scheduling-srv ${testSuffix}: gRPC`, () => {
     payloadShouldBeEmpty(await grpcSchedulingSrv.read(JobReadRequest.fromPartial({ subject }), {}), false);
   });
   beforeEach(async () => {
-    for (let event of ['jobsCreated', 'jobsDeleted']) {
+    for (const event of ['jobsCreated', 'jobsDeleted']) {
       await jobEvents.on(event, () => { });
     }
   });

src/worker.ts

@@ -129,7 +129,7 @@ const startGrpcMockServer = async (methodWithOutput: MethodWithOutput[]) => {
       oneofs: true
     });
     await mockServer.start();
-    logger.info('Mock ACS Server started on port 50061');
+    logger.info(`Mock ACS Server started on ${mockServer.serverAddress}`);
   } catch (err) {
     logger.error('Error starting mock ACS server', err);
   }
@@ -162,7 +162,7 @@ const startIDSGrpcMockServer = async (methodWithOutput: MethodWithOutput[]) => {
       oneofs: true
     });
     await mockServerIDS.start();
-    logger.info('Mock IDS Server started on port 50051');
+    logger.info(`Mock IDS Server started on ${mockServerIDS.serverAddress}`);
   } catch (err) {
     logger.error('Error starting mock IDS server', err);
   }

test/grpc_acs.spec.ts

@@ -6,7 +6,7 @@ import { PolicySetRQResponse } from '@restorecommerce/acs-client';
 import { Effect } from '@restorecommerce/rc-grpc-clients/dist/generated-server/io/restorecommerce/rule.js';
 import { createServiceConfig } from '@restorecommerce/service-config';
 
-export const cfg = createServiceConfig(process.cwd() + '/test');
+export const cfg = createServiceConfig(process.cwd());
 
 export function validateScheduledJob(job: any, expectedSchedule: string, logger: Logger): void {
   should.exist(job.data);