triggershell/src/cli/commands/run.ts

import path from "node:path";
import { eq } from "drizzle-orm";
import { WebSocket } from "ws";
import { ConfigError, loadConfig } from "../../lib/config/load";
import { defaultValuesForScript } from "../../lib/config/defaults";
import { getDb, migrateOnBoot } from "../../lib/db/client";
import { runs, type RunStatus } from "../../lib/db/schema";
import { startRun, reconcileOrphanedRuns } from "../../lib/runner/engine";
import { cancelRun } from "../../lib/runner/registry";
import { runEvents } from "../../lib/runner/events";
import { readLogTail } from "../../lib/runner/log-file";
import { buildVariableSchema } from "../../lib/validation/variable-schema";
import type { ServerMessage } from "../../lib/ws/protocol";
import { loadDotenv } from "../lib/env-file";
import { isServerReachable } from "../lib/network";
import { resolveConfigPath } from "../lib/paths";
import { coerceVariables, parseVarFlags } from "../lib/variables";

export interface RunOptions {
  config?: string;
  var: string[];
  host?: string;
  port?: number;
  token?: string;
  local?: boolean;
  remote?: boolean;
  wait: boolean;
}

const NON_TERMINAL: RunStatus[] = ["queued", "running"];

function exitCodeFor(status: RunStatus): number {
  return status === "succeeded" ? 0 : 1;
}

export async function runCommand(scriptId: string, opts: RunOptions): Promise<void> {
  const configPath = resolveConfigPath(opts.config);
  loadDotenv(path.join(path.dirname(configPath), ".env"));

  let loaded;
  try {
    loaded = loadConfig(configPath);
  } catch (error) {
    if (error instanceof ConfigError) {
      console.error(`Config error: ${error.message}`);
      for (const issue of error.issues) console.error(`  - ${issue}`);
      process.exitCode = 1;
      return;
    }
    throw error;
  }

  const { config } = loaded;
  const script = config.scripts.find((s) => s.id === scriptId);
  if (!script) {
    console.error(`No script '${scriptId}' configured.`);
    if (config.scripts.length > 0) {
      console.error(
        `Available scripts: ${config.scripts.map((s) => s.id).join(", ")}`,
      );
    }
    process.exitCode = 1;
    return;
  }

  let variables: Record<string, unknown>;
  try {
    const grouped = parseVarFlags(opts.var);
    const raw = {
      ...defaultValuesForScript(script.variables),
      ...coerceVariables(script.variables, grouped),
    };
    const parsed = buildVariableSchema(script).safeParse(raw);
    if (!parsed.success) {
      console.error("Validation failed:");
      for (const [field, issues] of Object.entries(
        parsed.error.flatten().fieldErrors,
      )) {
        console.error(`  ${field}: ${(issues ?? []).join(", ")}`);
      }
      process.exitCode = 1;
      return;
    }
    variables = parsed.data;
  } catch (error) {
    console.error((error as Error).message);
    process.exitCode = 1;
    return;
  }

  const host = opts.host ?? config.server.host;
  const port = opts.port ?? config.server.port;
  const url = `http://${host}:${port}`;
  const token = opts.token ?? process.env.TRIGGERSHELL_API_TOKEN;

  if (opts.local && opts.remote) {
    console.error("--local and --remote can't be used together.");
    process.exitCode = 1;
    return;
  }

  let useRemote: boolean;
  if (opts.remote) {
    if (!(await isServerReachable(url))) {
      console.error(`No triggershell server reachable at ${url}.`);
      process.exitCode = 1;
      return;
    }
    useRemote = true;
  } else if (opts.local) {
    useRemote = false;
  } else {
    useRemote = await isServerReachable(url);
  }

  if (useRemote) {
    if (config.auth.enabled && !token) {
      console.error(
        `${url} requires auth - pass --token or set TRIGGERSHELL_API_TOKEN (generate one with \`triggershell users add-token\`).`,
      );
      process.exitCode = 1;
      return;
    }
    await runRemote(url, scriptId, variables, token, opts.wait);
  } else {
    await runLocal(configPath, script.id, variables, opts.wait);
  }
}

async function runLocal(
  configPath: string,
  scriptId: string,
  variables: Record<string, unknown>,
  wait: boolean,
): Promise<void> {
  process.env.TRIGGERSHELL_CONFIG_PATH = configPath;
  migrateOnBoot();
  reconcileOrphanedRuns();

  const runId = await startRun({ scriptId, variables, triggeredBy: "cli" });
  console.log(`Started run ${runId}`);
  if (!wait) return;

  // Read-then-register, same as the WS subscribe handler (src/lib/ws/server.ts): the run may
  // already have produced output - or even finished - between `startRun` returning and this line,
  // so we snapshot the log file and current status first, synchronously, before attaching a live
  // listener for anything after that point. Both this read and the listener attach below are
  // synchronous (better-sqlite3 and fs are sync here), so there's no gap either could fall through.
  const row = getDb().select().from(runs).where(eq(runs.id, runId)).get();
  if (row) {
    const { text } = readLogTail(row.logFilePath);
    if (text) process.stdout.write(text);
    if (!NON_TERMINAL.includes(row.status)) {
      process.exitCode = exitCodeFor(row.status);
      return;
    }
  }

  const finalStatus = await new Promise<RunStatus>((resolve) => {
    function onMessage(message: ServerMessage) {
      if (message.runId !== runId) return;
      if (message.type === "output") {
        process.stdout.write(message.chunk);
      } else if (message.type === "status" && !NON_TERMINAL.includes(message.status)) {
        cleanup();
        resolve(message.status);
      }
    }
    function onSigint() {
      cancelRun(runId);
    }
    function cleanup() {
      runEvents.off("message", onMessage);
      process.off("SIGINT", onSigint);
    }
    process.on("SIGINT", onSigint);
    runEvents.on("message", onMessage);
  });

  process.exitCode = exitCodeFor(finalStatus);
}

async function runRemote(
  url: string,
  scriptId: string,
  variables: Record<string, unknown>,
  token: string | undefined,
  wait: boolean,
): Promise<void> {
  const headers: Record<string, string> = { "Content-Type": "application/json" };
  if (token) headers.Authorization = `Bearer ${token}`;

  const response = await fetch(`${url}/api/scripts/${scriptId}/runs`, {
    method: "POST",
    headers,
    body: JSON.stringify({ variables }),
  });

  if (!response.ok) {
    const body = await response.json().catch(() => ({}) as Record<string, unknown>);
    if (response.status === 401) {
      console.error(
        "Unauthorized - pass --token or set TRIGGERSHELL_API_TOKEN (see `triggershell users add-token`).",
      );
    } else if (body.fieldErrors) {
      console.error("Validation failed:");
      for (const [field, issues] of Object.entries(
        body.fieldErrors as Record<string, string[]>,
      )) {
        console.error(`  ${field}: ${issues.join(", ")}`);
      }
    } else {
      console.error((body.error as string) ?? `Request failed (${response.status})`);
    }
    process.exitCode = 1;
    return;
  }

  const { runId } = (await response.json()) as { runId: string };
  console.log(`Started run ${runId}`);
  if (!wait) return;

  const wsUrl = `${url.replace(/^http/, "ws")}/ws/runs${token ? `?token=${encodeURIComponent(token)}` : ""}`;

  const finalStatus = await new Promise<RunStatus>((resolve, reject) => {
    const ws = new WebSocket(wsUrl);

    function onSigint() {
      try {
        ws.send(JSON.stringify({ type: "cancel", runId }));
      } catch {
        // socket may already be closing - nothing more we can do
      }
    }
    process.on("SIGINT", onSigint);

    function cleanup() {
      process.off("SIGINT", onSigint);
      ws.close();
    }

    ws.on("open", () => {
      ws.send(JSON.stringify({ type: "subscribe", runId, afterBytes: 0 }));
    });
    ws.on("message", (raw) => {
      let message: ServerMessage;
      try {
        message = JSON.parse(raw.toString());
      } catch {
        return;
      }
      if (message.runId !== runId) return;
      if (message.type === "output") {
        process.stdout.write(message.chunk);
      } else if (message.type === "status" && !NON_TERMINAL.includes(message.status)) {
        cleanup();
        resolve(message.status);
      } else if (message.type === "error") {
        cleanup();
        reject(new Error(message.message));
      }
    });
    ws.on("error", (error) => {
      process.off("SIGINT", onSigint);
      reject(error);
    });
  });

  process.exitCode = exitCodeFor(finalStatus);
}