pi-car-companion/server/src/jobs.ts

import { randomUUID } from "node:crypto";
import { getServers } from "node:dns";
import { hostname, networkInterfaces } from "node:os";
import type { CompanionDatabase } from "./database.js";
import { collectSystemStatus } from "./status.js";
import { DISABLED_ADB_CONFIG, type AdbConfig } from "./adb.js";

const MAX_OUTPUT_BYTES = 32 * 1024;
const SECRET_KEY = /authorization|cookie|csrf|password|secret|token|credential/i;
const SECRET_VALUE = /(bearer\s+\S+|(?:password|secret|token|authorization|cookie)=\S+)/gi;

export type JobStatus = "queued" | "running" | "succeeded" | "failed" | "timed_out" | "interrupted";

export type JobDescriptor = {
  id: string;
  title: string;
  description: string;
  timeoutMs: number;
  riskLevel: "low" | "medium" | "high";
  confirmationRequired: boolean;
};

export type JobDefinition = JobDescriptor & {
  run: (signal: AbortSignal) => Promise<unknown>;
};

export type JobRun = {
  id: string;
  jobId: string;
  status: JobStatus;
  output: unknown | null;
  errorCode: string | null;
  createdAt: string;
  startedAt: string | null;
  finishedAt: string | null;
};

function networkDiagnostics() {
  let localHostname: { available: true; value: string } | { available: false; reason: string };
  let dnsServers: { available: true; value: string[] } | { available: false; reason: string };
  let interfaces:
    | { available: true; value: Array<{ name: string; address: string; family: string; internal: boolean; netmask: string }> }
    | { available: false; reason: string };
  try {
    localHostname = { available: true, value: hostname() };
  } catch {
    localHostname = { available: false, reason: "The host name is not exposed by this system" };
  }
  try {
    dnsServers = { available: true, value: getServers() };
  } catch {
    dnsServers = { available: false, reason: "Resolver configuration is not exposed by this system" };
  }
  try {
    interfaces = {
      available: true,
      value: Object.entries(networkInterfaces()).flatMap(([name, addresses]) =>
        (addresses ?? []).map(({ address, family, internal, netmask }) => ({ name, address, family, internal, netmask }))
      )
    };
  } catch {
    interfaces = { available: false, reason: "Network interfaces are not exposed by this system" };
  }
  return {
    hostname: localHostname,
    dnsServers,
    interfaces
  };
}

function supportBundle(database: CompanionDatabase, generatedAt: string, status: unknown) {
  const tableCounts = Object.fromEntries(
    ["users", "sessions", "audit_events", "job_runs"].map((table) => {
      const row = database.prepare(`SELECT COUNT(*) AS count FROM ${table}`).get() as { count: number };
      return [table, row.count];
    })
  );
  const jobStates = database
    .prepare("SELECT status, COUNT(*) AS count FROM job_runs GROUP BY status ORDER BY status")
    .all() as Array<{ status: JobStatus; count: number }>;

  return {
    manifest: {
      format: "pi-car-companion-support-bundle",
      version: 1,
      generatedAt,
      entries: [
        { name: "system-status", description: "Current host and companion health", classification: "diagnostic" },
        { name: "network", description: "Local interface and resolver state", classification: "sensitive-local" },
        { name: "database-summary", description: "Record counts only; no account or session values", classification: "diagnostic" }
      ],
      exclusions: ["password hashes", "session identifiers", "CSRF values", "cookies", "environment variables", "file contents"]
    },
    diagnostics: {
      systemStatus: status,
      network: networkDiagnostics(),
      database: { tableCounts, jobStates }
    }
  };
}

function defaultDefinitions(database: CompanionDatabase, adbConfig: AdbConfig): readonly JobDefinition[] {
  const safeJob = (definition: Omit<JobDefinition, "riskLevel" | "confirmationRequired">): JobDefinition => ({
    ...definition,
    riskLevel: "low",
    confirmationRequired: false
  });
  return [
    safeJob({
      id: "refresh-health",
      title: "Refresh system health",
      description: "Collect a fresh system health snapshot.",
      timeoutMs: 15_000,
      run: async () => collectSystemStatus(adbConfig)
    }),
    safeJob({
      id: "network-diagnostics",
      title: "Network diagnostics",
      description: "Inspect local interfaces and resolver configuration without contacting the internet.",
      timeoutMs: 10_000,
      run: async () => networkDiagnostics()
    }),
    safeJob({
      id: "support-bundle",
      title: "Generate support bundle",
      description: "Collect a bounded, sanitized diagnostic manifest without secrets or file contents.",
      timeoutMs: 20_000,
      run: async () => {
        const generatedAt = new Date().toISOString();
        return supportBundle(database, generatedAt, await collectSystemStatus(adbConfig));
      }
    })
  ];
}

function sanitize(value: unknown, key = ""): unknown {
  if (SECRET_KEY.test(key)) return "[REDACTED]";
  if (typeof value === "string") return value.replace(SECRET_VALUE, "[REDACTED]");
  if (Array.isArray(value)) return value.map((item) => sanitize(item));
  if (value && typeof value === "object") {
    return Object.fromEntries(Object.entries(value).map(([childKey, child]) => [childKey, sanitize(child, childKey)]));
  }
  return value;
}

function boundedJson(value: unknown): string {
  const json = JSON.stringify(sanitize(value));
  if (Buffer.byteLength(json) <= MAX_OUTPUT_BYTES) return json;
  return JSON.stringify({ truncated: true, message: "Job output exceeded the 32 KiB storage limit." });
}

function rowToRun(row: Record<string, unknown>): JobRun {
  return {
    id: String(row.id),
    jobId: String(row.job_id),
    status: row.status as JobStatus,
    output: typeof row.output_json === "string" ? JSON.parse(row.output_json) : null,
    errorCode: typeof row.error_code === "string" ? row.error_code : null,
    createdAt: String(row.created_at),
    startedAt: typeof row.started_at === "string" ? row.started_at : null,
    finishedAt: typeof row.finished_at === "string" ? row.finished_at : null
  };
}

export class JobRunner {
  readonly #jobs: Map<string, JobDefinition>;
  readonly #listeners = new Set<(run: JobRun) => void>();
  readonly #controllers = new Set<AbortController>();
  #closed = false;

  constructor(
    private readonly database: CompanionDatabase,
    jobs?: readonly JobDefinition[],
    adbConfig: AdbConfig = DISABLED_ADB_CONFIG
  ) {
    this.#jobs = new Map((jobs ?? defaultDefinitions(database, adbConfig)).map((job) => [job.id, job]));
  }

  listJobs(): JobDescriptor[] {
    return [...this.#jobs.values()].map((job) => ({
      id: job.id,
      title: job.title,
      description: job.description,
      timeoutMs: job.timeoutMs,
      riskLevel: job.riskLevel,
      confirmationRequired: job.confirmationRequired
    }));
  }

  has(jobId: string): boolean {
    return this.#jobs.has(jobId);
  }

  onRun(listener: (run: JobRun) => void): () => void {
    this.#listeners.add(listener);
    return () => this.#listeners.delete(listener);
  }

  close(): void {
    this.#closed = true;
    for (const controller of this.#controllers) controller.abort();
    this.database
      .prepare(
        `UPDATE job_runs SET status = 'interrupted', error_code = 'PROCESS_STOPPED', finished_at = ?
         WHERE status IN ('queued', 'running')`
      )
      .run(new Date().toISOString());
    this.#listeners.clear();
  }

  enqueue(jobId: string, actorUserId: number): JobRun {
    if (this.#closed) throw new Error("RUNNER_CLOSED");
    const job = this.#jobs.get(jobId);
    if (!job) throw new Error("UNKNOWN_JOB");
    const id = randomUUID();
    const createdAt = new Date().toISOString();
    this.database
      .prepare(
        `INSERT INTO job_runs (id, job_id, actor_user_id, status, created_at)
         VALUES (?, ?, ?, 'queued', ?)`
      )
      .run(id, jobId, actorUserId, createdAt);
    this.#emit(id);
    queueMicrotask(() => void this.#execute(id, job));
    return this.getRun(id)!;
  }

  getRun(id: string): JobRun | null {
    const row = this.database.prepare("SELECT * FROM job_runs WHERE id = ?").get(id) as Record<string, unknown> | undefined;
    return row ? rowToRun(row) : null;
  }

  listRuns(limit = 50): JobRun[] {
    const rows = this.database
      .prepare("SELECT * FROM job_runs ORDER BY created_at DESC LIMIT ?")
      .all(Math.min(Math.max(limit, 1), 100)) as Record<string, unknown>[];
    return rows.map(rowToRun);
  }

  async #execute(id: string, job: JobDefinition): Promise<void> {
    if (this.#closed) return;
    const startedAt = new Date().toISOString();
    this.database.prepare("UPDATE job_runs SET status = 'running', started_at = ? WHERE id = ?").run(startedAt, id);
    this.#emit(id);
    const controller = new AbortController();
    this.#controllers.add(controller);
    let timer: NodeJS.Timeout | undefined;
    try {
      const timeout = new Promise<never>((_resolve, reject) => {
        timer = setTimeout(() => {
          controller.abort();
          reject(new Error("JOB_TIMEOUT"));
        }, job.timeoutMs);
        timer.unref();
      });
      const output = await Promise.race([job.run(controller.signal), timeout]);
      if (this.#closed) return;
      this.database
        .prepare("UPDATE job_runs SET status = 'succeeded', output_json = ?, finished_at = ? WHERE id = ?")
        .run(boundedJson(output), new Date().toISOString(), id);
      this.#emit(id);
    } catch (error) {
      if (this.#closed) return;
      const timedOut = error instanceof Error && error.message === "JOB_TIMEOUT";
      this.database
        .prepare("UPDATE job_runs SET status = ?, error_code = ?, finished_at = ? WHERE id = ?")
        .run(timedOut ? "timed_out" : "failed", timedOut ? "JOB_TIMEOUT" : "JOB_FAILED", new Date().toISOString(), id);
      this.#emit(id);
    } finally {
      if (timer) clearTimeout(timer);
      this.#controllers.delete(controller);
    }
  }

  #emit(id: string): void {
    const run = this.getRun(id);
    if (!run) return;
    for (const listener of this.#listeners) {
      try {
        listener(run);
      } catch {
        // A disconnected event consumer must not change the persisted job result.
      }
    }
  }
}