triggershell/src/lib/runner/engine.ts

import fs from "node:fs";
import path from "node:path";
import { execa } from "execa";
import { eq } from "drizzle-orm";
import { getDb } from "../db/client";
import { runs, type RunStatus } from "../db/schema";
import { getConfig, getScript } from "../config/load";
import type { ScriptConfig } from "../config/schema";
import { buildInvocation, type Invocation } from "./build-args";
import { registerRun, unregisterRun } from "./registry";
import { emitRunMessage } from "./events";
import { logger } from "../logger";

const log = logger.child({ mod: "runner" });

export class ScriptNotFoundError extends Error {}

interface StartRunOptions {
  scriptId: string;
  variables: Record<string, unknown>;
  triggeredBy: string;
}

export async function startRun({
  scriptId,
  variables,
  triggeredBy,
}: StartRunOptions): Promise<string> {
  const script = getScript(scriptId);
  if (!script) throw new ScriptNotFoundError(`Unknown script '${scriptId}'`);

  const { logsDir } = getConfig();
  fs.mkdirSync(logsDir, { recursive: true });

  const invocation = buildInvocation(script, variables);
  const db = getDb();

  const runId = crypto.randomUUID();
  const logFilePath = path.join(logsDir, `${runId}.log`);

  db.insert(runs)
    .values({
      id: runId,
      scriptId: script.id,
      scriptName: script.name,
      status: "queued",
      variables: invocation.redactedVariables,
      resolvedCommand: invocation.redactedCommandLine,
      timeoutSeconds: script.timeoutSeconds,
      triggeredBy,
      logFilePath,
    })
    .run();

  log.info({ runId, scriptId: script.id, triggeredBy }, "run queued");

  // Fire and forget - the caller gets the runId immediately, progress streams over WS/polling.
  void executeRun(runId, script, invocation, logFilePath).catch((error) => {
    log.error(
      { runId, scriptId: script.id, err: error },
      "unhandled error executing run",
    );
  });

  return runId;
}

async function executeRun(
  runId: string,
  script: ScriptConfig,
  invocation: Invocation,
  logFilePath: string,
) {
  const db = getDb();
  const logStream = fs.createWriteStream(logFilePath, { flags: "a" });

  const setStatus = (
    status: RunStatus,
    extra: Partial<typeof runs.$inferInsert> = {},
  ) => {
    db.update(runs)
      .set({ status, ...extra })
      .where(eq(runs.id, runId))
      .run();
    emitRunMessage({
      type: "status",
      runId,
      status,
      exitCode: (extra.exitCode as number | null | undefined) ?? null,
      ts: Date.now(),
    });
  };

  const { configDir } = getConfig();
  const cwd = path.resolve(configDir, script.cwd);
  const controller = new AbortController();

  let seq = 0;
  const onChunk = (stream: "stdout" | "stderr") => (data: Buffer) => {
    const chunk = data.toString("utf-8");
    logStream.write(chunk);
    emitRunMessage({
      type: "output",
      runId,
      stream,
      chunk,
      seq: seq++,
      ts: Date.now(),
    });
  };

  registerRun({ runId, scriptId: script.id, controller });
  setStatus("running", { startedAt: new Date() });
  log.info(
    { runId, scriptId: script.id, command: script.command },
    "run started",
  );
  const startedAt = process.hrtime.bigint();

  try {
    const subprocess = execa(script.command, invocation.argv, {
      cwd,
      // Scripts run without a real TTY, so most CLI tools auto-disable color; these are the two
      // most widely honored override conventions (not universal - true TTY-detecting tools like
      // GNU coreutils' `--color=auto` still won't colorize without a pty). Real env/script-set
      // env still win, so this is only a default.
      env: {
        FORCE_COLOR: "1",
        CLICOLOR_FORCE: "1",
        ...process.env,
        ...invocation.env,
      },
      // `timeoutSeconds: 0` means no timeout - execa only enforces this when > 0.
      timeout: script.timeoutSeconds * 1000,
      cancelSignal: controller.signal,
      reject: false,
      shell: script.shell,
      input: invocation.stdin,
      buffer: false,
    });

    subprocess.stdout?.on("data", onChunk("stdout"));
    subprocess.stderr?.on("data", onChunk("stderr"));

    const result = await subprocess;

    let status: RunStatus;
    if (result.isCanceled) status = "cancelled";
    else if (result.timedOut) status = "timed_out";
    else if (result.failed) status = "failed";
    else status = "succeeded";

    setStatus(status, {
      exitCode: result.exitCode ?? null,
      endedAt: new Date(),
      errorMessage:
        status === "failed" || status === "timed_out"
          ? (result.shortMessage ?? null)
          : null,
    });
    const durationMs = Number(process.hrtime.bigint() - startedAt) / 1e6;
    log[status === "succeeded" ? "info" : "warn"](
      {
        runId,
        scriptId: script.id,
        status,
        exitCode: result.exitCode ?? null,
        durationMs: Math.round(durationMs),
      },
      "run finished",
    );
  } catch (error) {
    setStatus("failed", {
      endedAt: new Date(),
      errorMessage: (error as Error).message,
    });
    log.error(
      { runId, scriptId: script.id, err: error },
      "run failed to execute",
    );
  } finally {
    logStream.end();
    unregisterRun(runId);
  }
}

/** On boot, any DB row still `running`/`queued` has no live handle in this process - mark it interrupted
 * rather than pretending we can resume streaming its output. */
export function reconcileOrphanedRuns() {
  const db = getDb();
  const now = new Date();
  const orphaned = db
    .update(runs)
    .set({
      status: "interrupted",
      endedAt: now,
      errorMessage: "Server restarted while this run was in progress.",
    })
    .where(eq(runs.status, "running"))
    .run();
  const queued = db
    .update(runs)
    .set({
      status: "interrupted",
      endedAt: now,
      errorMessage: "Server restarted before this run could start.",
    })
    .where(eq(runs.status, "queued"))
    .run();

  const interruptedCount = orphaned.changes + queued.changes;
  if (interruptedCount > 0) {
    log.warn(
      { runningCount: orphaned.changes, queuedCount: queued.changes },
      "marked orphaned runs as interrupted after restart",
    );
  }

  return orphaned;
}