1 files changed, 86 insertions, 0 deletions
diff --git a/lib/process/exec.ts b/lib/process/exec.ts
new file mode 100644
index 0000000..f8d572c
--- /dev/null
+++ b/lib/process/exec.ts
@@ -0,0 +1,86 @@
+import {
+    Either,
+    IEither,
+    type ITraceable,
+    LogLevel,
+    LogMetricTraceSupplier,
+    Metric,
+    TraceUtil,
+} from '@emprespresso/pengueno';
+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 = { streamTraceable?: Array<'stdout' | 'stderr'>; env?: Environment; clearEnv?: boolean };
+export const getStdout = (
+    cmd: ITraceable<Command, LogMetricTraceSupplier>,
+    options: Options = { streamTraceable: [] },
+): Promise<IEither<Error, string>> =>
+    cmd
+        .flatMap(TraceUtil.withFunctionTrace(getStdout))
+        .flatMap((tCmd) => tCmd.traceScope(() => `Command = ${tCmd.get()}`))
+        .map((tCmd) => {
+            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>(
+                new Promise<StdStreams>((res, rej) => {
+                    const proc = exec(_exec, { env });
+                    let stdout = '';
+                    proc.stdout?.on('data', (d: Buffer) => {
+                        const s = d.toString();
+                        stdout += s;
+                        if (options.streamTraceable?.includes('stdout')) {
+                            tCmd.trace.trace(s);
+                        }
+                    });
+                    const stderr = '';
+                    proc.stderr?.on('data', (d: Buffer) => {
+                        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}. ${stderr}`));
+                        }
+                    });
+                }),
+            );
+        })
+        .map(
+            TraceUtil.promiseify((tEitherStdStreams) =>
+                tEitherStdStreams.get().mapRight(({ stderr, stdout }) => {
+                    if (stderr) tEitherStdStreams.trace.traceScope(LogLevel.DEBUG).trace(`StdErr = ${stderr}`);
+                    return stdout;
+                }),
+            ),
+        )
+        .peek(TraceUtil.promiseify(TraceUtil.traceResultingEither(CmdMetric)))
+        .get();
+
+export const getStdoutMany = (
+    cmds: ITraceable<Array<Command>, LogMetricTraceSupplier>,
+    options: Options = { streamTraceable: [] },
+): Promise<IEither<Error, Array<string>>> =>
+    cmds
+        .coExtend((t) => t.get())
+        .reduce(
+            async (_result, tCmd) => {
+                const result = await _result;
+                return result.joinRightAsync(
+                    () => tCmd.map((cmd) => getStdout(cmd, options)).get(),
+                    (stdout, pre) => pre.concat(stdout),
+                );
+            },
+            Promise.resolve(Either.right<Error, Array<string>>([])),
+        );