gadget/gadget-drone/src/services/subprocess.ts

// src/services/subprocess.ts
// Copyright (C) 2026 Rob Colbert <rob.colbert@openplatform.us>
// Licensed under the Apache License, Version 2.0

import assert from "node:assert";

import path from "node:path";
import fs from "node:fs";

import { ChildProcess, spawn } from "node:child_process";

import { GadgetId, IProject } from "@gadget/api";

import WorkspaceService from "./workspace.js";

import { GadgetService } from "../lib/service.ts";

export interface SubProcess {
  project: IProject;
  createdAt: Date;
  updatedAt: Date;
  process: ChildProcess;
  pid: number;
  stdoutFileName: string;
  stderrFileName: string;
}

export interface SubProcessSummary {
  pid: number;
  project: IProject;
  createdAt: Date;
  updatedAt: Date;
  stdoutFileName: string;
  stderrFileName: string;
}

type SubProcessMap = Map<number, SubProcess>;

const GRACEFUL_SHUTDOWN_MS = 5_000;

class SubProcessService extends GadgetService {
  private processMap: SubProcessMap = new Map<number, SubProcess>();

  get name(): string {
    return "SubprocessService";
  }
  get slug(): string {
    return "svc:subprocess";
  }

  async start(): Promise<void> {
    this.log.info("started");
  }

  async stop(): Promise<void> {
    await this.killAll();
    this.log.info("stopped");
  }

  async create(
    project: IProject,
    cmd: string,
    args?: string[],
  ): Promise<SubProcess> {
    const NOW = new Date();
    assert(
      WorkspaceService.gadgetDir,
      "Gadget workspace directory not initialized",
    );

    const logDir = path.join(WorkspaceService.gadgetDir, "subprocess-logs");
    await fs.promises.mkdir(logDir, { recursive: true });

    const projectDir = WorkspaceService.getProjectDirectory(project.slug);
    const child = spawn(cmd, args, { cwd: projectDir });
    assert(child.pid, "subprocess did not define a process id");

    const pid = child.pid;
    const stdoutFileName = path.join(logDir, `subproc-${pid}.stdout.log`);
    const stderrFileName = path.join(logDir, `subproc-${pid}.stderr.log`);

    const stdoutFile = fs.createWriteStream(stdoutFileName, "utf-8");
    const stderrFile = fs.createWriteStream(stderrFileName, "utf-8");

    const sp: SubProcess = {
      project,
      createdAt: NOW,
      updatedAt: NOW,
      process: child,
      pid,
      stdoutFileName,
      stderrFileName,
    };

    child.stdout.on("data", (data: any) => {
      sp.updatedAt = new Date();
      stdoutFile.write(data);
    });
    child.stderr.on("data", (data: any) => {
      sp.updatedAt = new Date();
      stderrFile.write(data);
    });

    child.on("exit", () => {
      stdoutFile.close();
      stderrFile.close();
    });

    this.processMap.set(pid, sp);

    return sp;
  }

  ps(): SubProcess[] {
    return Array.from(this.processMap.values());
  }

  summarize(sp: SubProcess): SubProcessSummary {
    return {
      pid: sp.pid,
      project: sp.project,
      createdAt: sp.createdAt,
      updatedAt: sp.updatedAt,
      stdoutFileName: sp.stdoutFileName,
      stderrFileName: sp.stderrFileName,
    };
  }

  async killPid(pid: number): Promise<boolean> {
    const sp = this.processMap.get(pid);
    if (!sp) {
      throw new Error("process not found");
    }
    return this.killSubProcess(sp);
  }

  async killProject(projectId: GadgetId): Promise<void> {
    for (const sp of this.processMap.values()) {
      if (sp.project._id !== projectId) {
        continue;
      }
      await this.killSubProcess(sp);
    }
  }

  async killSubProcess(sp: SubProcess): Promise<boolean> {
    await this.terminateProcess(sp);
    await this.removeLogFiles(sp);
    this.processMap.delete(sp.pid);
    this.log.info("subprocess killed", { pid: sp.pid });
    return true;
  }

  async haltSubProcess(sp: SubProcess): Promise<boolean> {
    await this.terminateProcess(sp);
    this.processMap.delete(sp.pid);
    this.log.info("subprocess halted, logs retained", { pid: sp.pid });
    return true;
  }

  async killAll(): Promise<boolean> {
    const entries = Array.from(this.processMap.entries());
    for (const [pid] of entries) {
      const sp = this.processMap.get(pid);
      if (sp) {
        await this.killSubProcess(sp);
      }
    }
    return true;
  }

  async haltPid(pid: number): Promise<boolean> {
    const sp = this.processMap.get(pid);
    if (!sp) {
      throw new Error("process not found");
    }
    return this.haltSubProcess(sp);
  }

  async getLog(
    pid: number,
    stream: "stdout" | "stderr",
  ): Promise<string> {
    const sp = this.processMap.get(pid);
    if (!sp) {
      throw new Error("process not found");
    }

    const filePath =
      stream === "stdout" ? sp.stdoutFileName : sp.stderrFileName;
    try {
      return await fs.promises.readFile(filePath, "utf-8");
    } catch {
      return "";
    }
  }

  private async terminateProcess(sp: SubProcess): Promise<void> {
    const pid = sp.pid;
    this.log.info("terminating subprocess", { pid });

    const exited = new Promise<void>((resolve) => {
      sp.process.on("exit", () => resolve());
    });

    sp.process.stdout?.removeAllListeners();
    sp.process.stderr?.removeAllListeners();

    sp.process.kill("SIGINT");

    const timeout = new Promise<never>((_, reject) =>
      setTimeout(() => reject(new Error("graceful shutdown timeout")), GRACEFUL_SHUTDOWN_MS),
    );

    try {
      await Promise.race([exited, timeout]);
    } catch {
      this.log.warn("subprocess did not exit gracefully, sending SIGKILL", {
        pid,
      });
      sp.process.kill("SIGKILL");
      try {
        await Promise.race([
          exited,
          new Promise<never>((_, reject) =>
            setTimeout(() => reject(new Error("SIGKILL timeout")), GRACEFUL_SHUTDOWN_MS),
          ),
        ]);
      } catch {
        this.log.warn("subprocess did not respond to SIGKILL", { pid });
      }
    }
  }

  private async removeLogFiles(sp: SubProcess): Promise<void> {
    await fs.promises.rm(sp.stdoutFileName, { force: true });
    await fs.promises.rm(sp.stderrFileName, { force: true });
  }
}

export default new SubProcessService();