add jobs data + executor spawn async

Author: tacxou

src/_common/tasks/executor.task.ts

@@ -10,35 +10,32 @@ export async function executorTask(
   options?: ExecutorConfigInterface,
 ): Promise<ExecutorResponseInterface> {
   return new Promise((resolve, reject) => {
-    const child = spawn(join(command), [], {
+    const jobDataArg = JSON.stringify(job.data);
+    const escapedJobDataArg = `'${jobDataArg.replace(/'/g, "'\\''")}'`;
+
+    const child = spawn(join(command), [escapedJobDataArg], {
       shell: options?.shell ?? true,
     });
 
-    let output = '';
-    let error = '';
+    let outputChunk = '';
+    let errorChunk = '';
 
-    // Ecoute pour la sortie standard (stdout)
     child.stdout.on('data', (data) => {
       console.log(`stdout: ${data}`);
-      output += data;
+      outputChunk += data;
     });
 
-    // Ecoute pour la sortie d'erreur (stderr)
     child.stderr.on('data', (data) => {
       console.error(`stderr: ${data}`);
-      error += data;
+      errorChunk += data;
     });
 
-    // Envoyer les données de job au processus enfant
-    child.stdin.write(JSON.stringify(job.data));
-    child.stdin.end();
-
     child.on('close', (code) => {
       console.log(`Le processus enfant s'est terminé avec le code ${code}`);
       resolve({
         status: code,
-        output: output,
-        error: error,
+        output: outputChunk?.toString(),
+        error: errorChunk?.toString(),
       });
     });
 
@@ -47,13 +44,4 @@ export async function executorTask(
       reject(spawnError);
     });
   });
-  /**const out = spawnSync(join(command), [], {
-    input: JSON.stringify(job.data),
-    shell: options?.shell ?? true,
-  });
-  return Promise.resolve({
-    status: out.status,
-    output: out.stdout?.toString(),
-    error: out.stderr?.toString(),
-  });**/
 }

src/backend-runner/_dto/backend-result-info.dto.ts

@@ -0,0 +1,15 @@
+import { IsEnum, IsObject, IsOptional, IsString } from 'class-validator';
+import { BackendCodesEnum } from '../_interfaces/backend-codes.enum';
+
+export class BackendResultInfoDto {
+  @IsEnum(BackendCodesEnum)
+  public status: number;
+
+  @IsString()
+  @IsOptional()
+  public message?: string;
+
+  @IsObject()
+  @IsOptional()
+  public data?: object;
+}

src/backend-runner/_executors/catch-all.executor.ts

@@ -4,8 +4,11 @@ import { ExecutorExecuteResponseInterface, ExecutorInterface } from '../executor
 import { join } from 'path';
 import { Job } from 'bullmq';
 import { BackendConfigV1Dto } from '../_dto/backend-config-v1.dto';
-import { BackendResultInterface } from '../_interfaces/backend-result.interface';
+import { BackendResultInfoInterface, BackendResultInterface } from '../_interfaces/backend-result.interface';
 import { BackendCodesEnum } from '../_interfaces/backend-codes.enum';
+import { validateOrReject } from 'class-validator';
+import { BackendResultInfoDto } from '../_dto/backend-result-info.dto';
+import { plainToInstance } from 'class-transformer';
 
 export class CatchAllExecutor implements ExecutorInterface {
   public constructor(public service: BackendRunnerService) {}
@@ -46,29 +49,70 @@ export class CatchAllExecutor implements ExecutorInterface {
     try {
       if (process.status !== 0) {
         this.service.logger.error(`Error executing backend ${backend.name}`);
+        const error = this.extractLastJsonImproved(process.error);
+        const errorSchema = plainToInstance(BackendResultInfoDto, error);
+        await validateOrReject(errorSchema);
+
         return {
           backend: backend.name,
           status: process.status,
-          error: JSON.parse(process.output),
+          error,
         };
       }
 
       this.service.logger.log(`Backend ${backend.name} executed successfully`);
+      const output = this.extractLastJsonImproved(process.output);
+      const outputSchema = plainToInstance(BackendResultInfoDto, output);
+      await validateOrReject(outputSchema);
+
       return {
         backend: backend.name,
         status: process.status,
-        output: JSON.parse(process.output),
+        output,
       };
     } catch (e) {
       this.service.logger.error(`Error parsing JSON output from backend ${backend.name}`);
+
       return {
         backend: backend.name,
         status: BackendCodesEnum.INVALID_JSON_RESPONSE,
-        error: {
-          status: BackendCodesEnum.INVALID_JSON_RESPONSE,
-          message: process.error,
-        },
+        message: `Invalid JSON response from backend: ${process.error || process.output}`,
       };
     }
   }
+
+  private extractLastJsonImproved(stdout: string): BackendResultInfoInterface {
+    const jsonCandidates = [];
+    let braceCount = 0,
+      inString = false,
+      escape = false,
+      currentJson = '';
+
+    for (const char of stdout) {
+      if (escape) {
+        escape = false;
+      } else if (char === '\\') {
+        escape = true;
+      } else if (char === '"') {
+        inString = !inString;
+      } else if (!inString && char === '{') {
+        braceCount++;
+        if (braceCount === 1) {
+          currentJson = char;
+          continue;
+        }
+      } else if (!inString && char === '}') {
+        braceCount--;
+      }
+
+      if (braceCount > 0) currentJson += char;
+      if (braceCount === 0 && currentJson !== '') {
+        currentJson += char;
+        jsonCandidates.push(currentJson);
+        currentJson = '';
+      }
+    }
+
+    return JSON.parse(jsonCandidates[jsonCandidates.length - 1]);
+  }
 }

src/backend-runner/_interfaces/backend-result.interface.ts

@@ -1,6 +1,7 @@
 export interface BackendResultInterface {
   backend: string;
   status: number;
+  message?: string;
   output?: BackendResultInfoInterface;
   error?: BackendResultInfoInterface;
 }
@@ -9,4 +10,4 @@ export interface BackendResultInfoInterface {
   status: number;
   message?: string;
   data?: object;
-}
+}