c0py/apps/api/src/worker/scan-worker.ts

import type { FastifyBaseLogger } from "fastify";
import { claimPendingScan, getScanTargetUrl, completeScan, failScan } from "../services/scans.js";
import type { PoolLike } from "../services/registries.js";
import { capturePage, buildEvidence } from "../capture/capture.js";
import type { Aud0Emitter } from "../audit/aud0.js";

export interface ScanWorkerDeps {
  db: PoolLike;
  logger: FastifyBaseLogger;
  aud0: Aud0Emitter;
  capture?: (url: string, timeoutMs: number) => Promise<{ evidence: unknown[]; summary: unknown }>;
  pollIntervalMs?: number;
  crawlTimeoutMs?: number;
  enabled?: boolean;
}

export function startScanWorker(deps: ScanWorkerDeps): {
  processOne: () => Promise<boolean>;
  stop: () => Promise<void>;
} {
  const interval = deps.pollIntervalMs ?? 2000;
  const crawlTimeoutMs = deps.crawlTimeoutMs ?? 30_000;
  const capture =
    deps.capture ??
    (async (url: string, timeoutMs: number) => {
      const page = await capturePage(url, { timeoutMs });
      return buildEvidence("scan", page);
    });

  let stopped = false;
  let timer: NodeJS.Timeout | null = null;

  async function processOne(): Promise<boolean> {
    let scan: Awaited<ReturnType<typeof claimPendingScan>> = null;
    try {
      scan = await claimPendingScan(deps.db);
    } catch (err) {
      deps.logger.error({ err }, "scan worker claim failed");
      return false;
    }
    if (!scan) return false;

    const started = Date.now();
    deps.logger.info({ scanId: scan.id, url: scan.registry_id }, "scan running");
    try {
      const url = await getScanTargetUrl(deps.db, scan.registry_id);
      if (!url) throw new Error("registry url not found");
      const out = await capture(url, crawlTimeoutMs);
      await completeScan(deps.db, scan.id, out);
      deps.aud0({
        event_type: "c0py.scan_completed",
        tenant_id: scan.tenant_id,
        actor_id: scan.owner_sub,
        app_id: "c0py",
        capability_id: "c0py.scan_run",
        resource_type: "scan",
        resource_id: scan.id,
        risk_level: "L2",
        payload: { summary: out.summary, durationMs: Date.now() - started },
      });
      deps.logger.info({ scanId: scan.id, durationMs: Date.now() - started }, "scan completed");
    } catch (err) {
      const message = err instanceof Error ? err.message : String(err);
      try {
        await failScan(deps.db, scan.id, message);
      } catch (err2) {
        deps.logger.error({ err: err2 }, "failScan write failed");
      }
      deps.aud0({
        event_type: "c0py.scan_failed",
        tenant_id: scan.tenant_id,
        actor_id: scan.owner_sub,
        app_id: "c0py",
        capability_id: "c0py.scan_run",
        resource_type: "scan",
        resource_id: scan.id,
        risk_level: "L2",
        payload: { error: message },
      });
      deps.logger.warn({ scanId: scan.id, err: message }, "scan failed");
    }
    return true;
  }

  async function loop(): Promise<void> {
    while (!stopped) {
      const didWork = await processOne();
      if (!didWork) {
        await new Promise<void>((resolve) => {
          timer = setTimeout(resolve, interval);
        });
      }
    }
  }

  if (deps.enabled !== false) {
    void loop();
  }

  return {
    processOne,
    stop: async () => {
      stopped = true;
      if (timer) clearTimeout(timer);
    },
  };
}