Uses correct subpackage for server in ci and streams stdout and stderr during ci builds

4 files changed, 49 insertions, 12 deletions

diff --git a/.ci/ci.cjs b/.ci/ci.cjs
index 323c561..8d183d4 100755
--- a/.ci/ci.cjs
+++ b/.ci/ci.cjs
@@ -477,7 +477,15 @@ var Optional = class _Optional extends _Tagged3 {
 };
 
 // ../model/job/index.ts
-var isJob = (j) => !!(isObject(j) && "arguments" in j && isObject(j.arguments) && "type" in j && typeof j.type === "string" && j);
+var JobTypes = [
+  "fetch_code",
+  "ci_pipeline",
+  "build_docker_image.js",
+  "ansible_playbook.js",
+  "checkout_ci.js"
+];
+var isJobType = (j) => typeof j === "string" && JobTypes.includes(j);
+var isJob = (j) => !!(isObject(j) && "arguments" in j && isObject(j.arguments) && "type" in j && isJobType(j.type) && j);
 
 // ../model/pipeline/builder.ts
 var BasePipelineBuilder = class {
@@ -564,7 +572,7 @@ var getPipeline = () => {
   gitHookPipeline.addStage({
     parallelJobs: [baseCiPackageBuild]
   });
-  const subPackages = ["worker", "hooks"].map((_package) => ({
+  const subPackages = ["worker", "server"].map((_package) => ({
     type: "build_docker_image.js",
     arguments: {
       ...commonBuildArgs,
diff --git a/.ci/ci.ts b/.ci/ci.ts
index e2e375b..f634b21 100644
--- a/.ci/ci.ts
+++ b/.ci/ci.ts
@@ -5,6 +5,7 @@ import {
     BuildDockerImageJob,
     DefaultGitHookPipelineBuilder,
     FetchCodeJob,
+    Job,
 } from '@emprespresso/ci_model';
 
 const REGISTRY = 'oci.liz.coffee';
@@ -37,7 +38,7 @@ const getPipeline = () => {
         parallelJobs: [baseCiPackageBuild],
     });
 
-    const subPackages = ['worker', 'hooks'].map((_package) => ({
+    const subPackages = ['worker', 'server'].map((_package) => (<Job>{
         type: 'build_docker_image.js',
         arguments: {
             ...commonBuildArgs,
diff --git a/u/process/exec.ts b/u/process/exec.ts
index a2cdbca..7721f61 100644
--- a/u/process/exec.ts
+++ b/u/process/exec.ts
@@ -7,19 +7,17 @@ import {
     Metric,
     TraceUtil,
 } from '@emprespresso/pengueno';
-import { promisify } from 'node:util';
-import { exec as execCallback } from 'node:child_process';
-const exec = promisify(execCallback);
+import { exec } from 'node:child_process';
 
 export type Command = string[] | string;
 export type StdStreams = { stdout: string; stderr: string };
 
 export const CmdMetric = Metric.fromName('Exec').asResult();
 type Environment = Record<string, string>;
-type Options = { env?: Environment; clearEnv?: boolean };
+type Options = { streamTraceable?: Array<'stdout' | 'stderr'>; env?: Environment; clearEnv?: boolean };
 export const getStdout = (
     cmd: ITraceable<Command, LogMetricTraceSupplier>,
-    options: Options = {},
+    options: Options = { streamTraceable: [] },
 ): Promise<IEither<Error, string>> =>
     cmd
         .flatMap(TraceUtil.withFunctionTrace(getStdout))
@@ -28,7 +26,36 @@ export const getStdout = (
             const cmd = tCmd.get();
             const _exec = typeof cmd === 'string' ? cmd : cmd.join(' ');
             const env = options.clearEnv ? options.env : { ...process.env, ...options.env };
-            return Either.fromFailableAsync<Error, StdStreams>(exec(_exec, { env }));
+            return Either.fromFailableAsync<Error, StdStreams>(
+                new Promise<StdStreams>((res, rej) => {
+                    const proc = exec(_exec, { env });
+                    let stdout = '';
+                    proc.stdout?.on('data', (d) => {
+                        const s = d.toString();
+                        stdout += s;
+                        if (options.streamTraceable?.includes('stdout')) {
+                            tCmd.trace.trace(s);
+                        }
+                    });
+                    let stderr = '';
+                    proc.stderr?.on('data', (d) => {
+                        const s = d.toString();
+                        stdout += s;
+                        if (options.streamTraceable?.includes('stderr')) {
+                            tCmd.trace.trace(s);
+                        }
+                    });
+
+                    proc.on('exit', (code) => {
+                        const streams = { stdout, stderr };
+                        if (code === 0) {
+                            res(streams);
+                        } else {
+                            rej(new Error(`exited with non-zero code: ${code}`));
+                        }
+                    });
+                }),
+            );
         })
         .map(
             TraceUtil.promiseify((tEitherStdStreams) =>
@@ -43,7 +70,7 @@ export const getStdout = (
 
 export const getStdoutMany = (
     cmds: ITraceable<Array<Command>, LogMetricTraceSupplier>,
-    options: Options = {},
+    options: Options = { streamTraceable: [] },
 ): Promise<IEither<Error, Array<string>>> =>
     cmds
         .coExtend((t) => t.get())
diff --git a/worker/executor.ts b/worker/executor.ts
index fe3a688..2640f39 100644
--- a/worker/executor.ts
+++ b/worker/executor.ts
@@ -25,10 +25,11 @@ export const executeJob = (tJob: ITraceable<Job, LogMetricTraceSupplier>) => {
                     tJob.trace.traceScope(LogLevel.ERROR).trace(JSON.stringify(badEntries));
                     return new Error('invalid job arguments');
                 })
-                .flatMapAsync((args) => getStdout(tJob.move(tJob.get().type), { env: args })),
+                .flatMapAsync((args) =>
+                    getStdout(tJob.move(tJob.get().type), { env: args, streamTraceable: ['stdout', 'stderr'] }),
+                ),
         )
         .flatMapAsync(TraceUtil.promiseify(TraceUtil.traceResultingEither(metric)))
-        .peek(TraceUtil.promiseify((t) => t.traceScope(() => LogLevel.DEBUG).trace.trace(JSON.stringify(t.get()))))
         .get();
 };
 // -- </job.exectuor> --