spb/social-runtime-crawler

1import { spawn, type ChildProcess } from "node:child_process";2import fs from "node:fs";3import path from "node:path";4import { fileURLToPath } from "node:url";5import { createLogger, isPlatform, newId, type AgentMode, type AppConfig } from "@src/shared";6import type { PostgresStore } from "@src/storage";7import { hasAdapter, SUPPORTED_PLATFORMS } from "@src/connectors";89const log = createLogger("jobs");10const here = path.dirname(fileURLToPath(import.meta.url));11const repoRoot = path.resolve(here, "..", "..", "..");1213export interface JobRequest {14  platform: string;15  goal?: string;16  query?: string;17  mode?: AgentMode;18  seed?: string;19  minutes?: number;20  actions?: number;21  media?: number;22  headless?: boolean;23  account?: string;24}2526/**27 * Job runner (§63 control plane, single-node version): spawns the worker CLI as a child process,28 * streams its log to data/jobs/<id>.log and tracks status in crawl_jobs. One browser per job.29 */30export class JobRunner {31  private running = new Map<string, ChildProcess>();32  readonly logsDir: string;33  constructor(private readonly store: PostgresStore, private readonly cfg: AppConfig, private readonly maxConcurrent = 2) {34    this.logsDir = path.join(cfg.dataDir, "jobs");35    fs.mkdirSync(this.logsDir, { recursive: true });36  }3738  get activeCount() {39    return this.running.size;40  }4142  async start(req: JobRequest): Promise<{ job_id: string }> {43    if (!isPlatform(req.platform)) throw new Error("unknown platform");44    if (!hasAdapter(req.platform)) throw new Error(`no adapter for ${req.platform} yet (available: ${SUPPORTED_PLATFORMS.join(", ")})`);45    if (this.running.size >= this.maxConcurrent) throw new Error(`max ${this.maxConcurrent} concurrent jobs`);46    const job_id = newId("job");47    const mode = req.mode ?? "research";48    const minutes = clamp(req.minutes ?? 10, 1, 120);49    const actions = clamp(req.actions ?? 60, 1, 1000);50    const media = clamp(req.media ?? 1, 0, 4);51    const goal = req.goal?.trim() || (req.query ? `Discover public content about “${req.query}”` : `Observe the ${req.platform} feed`);52    const args = ["node_modules/tsx/dist/cli.mjs", "apps/worker/src/cli.ts", mode === "learn" ? "learn" : "crawl", req.platform, "--job", job_id, "--goal", goal, "--mode", mode, "--minutes", String(minutes), "--actions", String(actions), "--media", String(media), "--account", req.account?.trim() || "research-01"];53    if (req.query) args.push("--query", req.query);54    if (req.seed) args.push("--seed", req.seed);55    if (req.headless !== false) args.push("--headless");5657    await this.store.pool.query(`insert into crawl_jobs(job_id, platform, mode, goal, budget, status) values ($1,$2,$3,$4,$5,'queued')`, [job_id, req.platform, mode, goal, JSON.stringify({ minutes, actions, media, query: req.query ?? null, seed: req.seed ?? null, headless: req.headless !== false, account: req.account ?? "research-01" })]);5859    const logFile = path.join(this.logsDir, `${job_id}.log`);60    const out = fs.openSync(logFile, "a");61    const child = spawn(process.execPath, args, { cwd: repoRoot, env: { ...process.env, SRC_LOG_PRETTY: "1", SRC_LOG_LEVEL: "info" }, stdio: ["ignore", out, out] });62    this.running.set(job_id, child);63    await this.store.pool.query(`update crawl_jobs set status='running' where job_id=$1`, [job_id]);64    log.info("job started", { job_id, platform: req.platform, mode, pid: child.pid });65    child.on("exit", async (code, signal) => {66      this.running.delete(job_id);67      fs.closeSync(out);68      const status = signal ? "stopped" : code === 0 ? "done" : "failed";69      let result: unknown = null;70      try {71        const sid = (await this.store.pool.query(`select session_id from crawl_jobs where job_id=$1`, [job_id])).rows[0]?.session_id as string | undefined;72        if (sid) {73          const f = path.join(this.cfg.sessionsDir, sid, "summary.json");74          if (fs.existsSync(f)) result = JSON.parse(fs.readFileSync(f, "utf8"));75        }76      } catch {77        /* ignore */78      }79      await this.store.pool.query(`update crawl_jobs set status=$2, finished_at=now(), result=$3 where job_id=$1`, [job_id, status, result ? JSON.stringify(result) : null]).catch(() => {});80      log.info("job finished", { job_id, status, code, signal });81    });82    return { job_id };83  }8485  stop(job_id: string): boolean {86    const child = this.running.get(job_id);87    if (!child) return false;88    child.kill("SIGTERM");89    setTimeout(() => child.kill("SIGKILL"), 15_000).unref();90    return true;91  }9293  logTail(job_id: string, lines = 200): string {94    const f = path.join(this.logsDir, `${job_id}.log`);95    if (!fs.existsSync(f)) return "";96    const all = fs.readFileSync(f, "utf8").split("\n");97    return all.slice(-lines).join("\n");98  }99}100101function clamp(n: number, lo: number, hi: number) {102  return Math.max(lo, Math.min(hi, Number.isFinite(n) ? n : lo));103}104