TypeScript 55.4%
Python 43.2%
SQL 1.2%
1import { closeDb, db, migrate, sql } from "@websensor/db";2import { closeDispatcher } from "@websensor/connectors";3import { config, factoryConfig, log } from "./config";4import { claimSeeds, evaluateShadows, factoryStats, processSeedSafe, seedFromCoverage, seedFromFiles, type ProcessOutcome } from "./factory";5import { closeRedis, getRedis } from "./redis";67/**8 * Source Factory process (`websensor-factory`): a slow, polite background worker separate from the poller.9 * N workers — each claims one queued seed at a time (systemic organizations first) and runs deep discovery +10 * shadow creation; a slow organization never stalls the others (per-organization deadline).11 * every 15 min — evaluate shadow sensors (accept / reject / defer)12 * every 24 h — re-seed from the coverage universes (new members become seeds automatically)13 * Heartbeat: Redis `ws:factory:status` (processed, queued, in-flight, last outcomes).14 */15async function heartbeat(extra: Record<string, unknown>): Promise<void> {16 try {17 await getRedis().set("ws:factory:status", JSON.stringify({ ...extra, at: new Date().toISOString(), pid: process.pid, version: config.version }), "EX", 120);18 } catch {19 // best effort20 }21}2223async function main(): Promise<void> {24 log.info({ env: config.env, version: config.version, concurrency: factoryConfig.concurrency, budget: factoryConfig.budget, deadlineMs: factoryConfig.deadlineMs }, "websensor factory starting");25 const applied = await migrate(config.databaseUrl);26 if (applied.length) log.info({ applied }, "migrations applied");27 if (!factoryConfig.enabled) {28 log.warn("WS_FACTORY=0 — factory idle");29 setInterval(() => void heartbeat({ idle: true }), 60_000);30 return;31 }32 let stopping = false;33 let processed = 0;34 let inflight = 0;35 let lastSeed = 0;36 let lastEval = 0;37 let maintenanceBusy = false;38 const recent: { seed: string; status: string; candidates: number; shadow: number; requests: number; ms: number }[] = [];39 const note = (o: ProcessOutcome): void => {40 processed++;41 recent.unshift({ seed: o.seedId, status: o.status, candidates: o.candidates, shadow: o.shadow, requests: o.requests, ms: o.durationMs });42 if (recent.length > 12) recent.pop();43 };4445 const seed = async (): Promise<void> => {46 try {47 const a = await seedFromFiles();48 const b = await seedFromCoverage({ mode: "hinted" });49 log.info({ files: a, coverage: { members: b.members, seeds: b.seeds, inserted: b.inserted, issues: b.issues.length } }, "factory: seeds refreshed");50 for (const i of b.issues) log.warn({ issue: i }, "coverage file issue");51 } catch (e) {52 log.error({ err: (e as Error).message }, "factory: seeding failed");53 }54 };5556 // Continuous worker pool: each worker claims one seed, processes it, and immediately claims the next.57 const worker = async (n: number): Promise<void> => {58 while (!stopping) {59 let seeds: Awaited<ReturnType<typeof claimSeeds>> = [];60 try {61 seeds = await claimSeeds(1);62 } catch (e) {63 log.warn({ worker: n, err: (e as Error).message }, "factory: claim failed");64 }65 if (!seeds.length) {66 await new Promise((r) => setTimeout(r, factoryConfig.tickSeconds * 1000 + Math.random() * 2000));67 continue;68 }69 inflight++;70 try {71 note(await processSeedSafe(seeds[0]!));72 } finally {73 inflight--;74 }75 }76 };7778 const maintenance = async (): Promise<void> => {79 if (stopping || maintenanceBusy) return;80 maintenanceBusy = true;81 try {82 if (Date.now() - lastSeed > 24 * 3600e3) {83 lastSeed = Date.now();84 await seed();85 }86 if (Date.now() - lastEval > 15 * 60e3) {87 lastEval = Date.now();88 await evaluateShadows().catch((e) => log.warn({ err: (e as Error).message }, "factory: shadow evaluation failed"));89 }90 const queued = Number((await db.execute<{ n: string }>(sql`select count(*)::text as n from factory_seeds where status = 'queued'`)).rows[0]?.n ?? 0);91 await heartbeat({ processed, queued, inflight, lastBatch: recent, concurrency: factoryConfig.concurrency });92 if (!inflight && !queued && processed % 50 === 0) {93 const st = await factoryStats();94 log.info({ seeds: st.seeds, candidates: st.candidates, shadow: st.shadow }, "factory: idle");95 }96 } catch (e) {97 log.error({ err: (e as Error).stack ?? (e as Error).message }, "factory maintenance failed");98 } finally {99 maintenanceBusy = false;100 }101 };102103 await heartbeat({ starting: true });104 await maintenance();105 const timer = setInterval(() => void maintenance(), 15_000);106 const workers = Array.from({ length: factoryConfig.concurrency }, (_, i) => worker(i));107108 const shutdown = async (signal: string): Promise<void> => {109 if (stopping) return;110 stopping = true;111 log.info({ signal, inflight }, "factory shutting down");112 clearInterval(timer);113 const deadline = Date.now() + 90_000;114 while (inflight > 0 && Date.now() < deadline) await new Promise((r) => setTimeout(r, 250));115 // Seeds still marked `discovering` are reclaimed automatically after 15 min (claim query; per-seed deadline is 2 min).116 await closeDispatcher();117 await closeRedis();118 await closeDb();119 process.exit(0);120 };121 process.on("SIGINT", () => void shutdown("SIGINT"));122 process.on("SIGTERM", () => void shutdown("SIGTERM"));123 process.on("unhandledRejection", (e) => log.error({ err: e instanceof Error ? e.stack : String(e) }, "unhandled rejection"));124 await Promise.all(workers);125}126127main().catch((e) => {128 log.fatal({ err: (e as Error).stack ?? String(e) }, "factory failed to start");129 process.exit(1);130});131