TypeScript 55.4%
Python 43.2%
SQL 1.2%
1import { coverageKey, newId, slugify, sourceSchema, type SourceSeed, type Tier } from "@websensor/core";2import { discoverOrganization, type CandidateKind, type DeepDiscoveryHints } from "@websensor/connectors";3import { db, factorySeeds, sensors, sql, type FactorySeed } from "@websensor/db";4import { factoryConfig, log } from "../config";5import { priorityFor, upsertSourceRecord } from "../registry";6import { scoreCandidate, selectForShadow } from "./score";78/**9 * Source Factory — one organization at a time:10 * seed → deep discovery → (create or attach the source) → score candidates → promotion policy →11 * shadow sensors (`status = SHADOW`: observed, evidence stored, nothing published) → later `evaluateShadows()`.12 */13export interface ProcessOutcome {14 seedId: string;15 sourceId: string | null;16 status: "discovered" | "blocked" | "error";17 candidates: number;18 shadow: number;19 requests: number;20 durationMs: number;21 notes: string[];22}2324interface SourceRow extends Record<string, unknown> {25 id: string;26 domain: string;27 categories: string[];28 tier: string;29 enabled: boolean;30}3132let sourceIndex: { at: number; byKey: Map<string, SourceRow>; ids: Set<string> } | null = null;33async function sourcesByDomain(force = false): Promise<{ byKey: Map<string, SourceRow>; ids: Set<string> }> {34 if (!force && sourceIndex && Date.now() - sourceIndex.at < 5 * 60_000) return sourceIndex;35 const rows = await db.execute<SourceRow>(sql`select id, domain, categories, tier, enabled from sources where kind = 'registry'`);36 const byKey = new Map<string, SourceRow>();37 const ids = new Set<string>();38 for (const r of rows.rows) {39 ids.add(r.id);40 const k = coverageKey(r.domain);41 if (!byKey.has(k)) byKey.set(k, r);42 }43 sourceIndex = { at: Date.now(), byKey, ids };44 return sourceIndex;45}4647export async function processSeed(seed: FactorySeed): Promise<ProcessOutcome> {48 const started = Date.now();49 const notes: string[] = [];50 const hints = (seed.hints ?? {}) as DeepDiscoveryHints & { posture?: boolean };51 const idx = await sourcesByDomain();52 const key = coverageKey(seed.domain);53 let source = idx.byKey.get(key) ?? null;5455 const disc = await discoverOrganization(seed.domain, {56 hints,57 budget: factoryConfig.budget,58 deadlineMs: factoryConfig.deadlineMs,59 concurrency: 4,60 sensorIdForLogs: `factory_${seed.id}`,61 probePages: true,62 probeSubdomains: true,63 probePosture: seed.importance >= 3 || hints.posture === true,64 probeOpenApi: true,65 });66 notes.push(...disc.notes);67 // The homepage may redirect to another registrable domain that we already monitor.68 if (!source && disc.canonicalDomain !== key) source = idx.byKey.get(disc.canonicalDomain) ?? null;6970 if (!disc.candidates.length) {71 const status = disc.blocked ? "blocked" : "discovered";72 await db.execute(sql`update factory_seeds set status = ${status}, candidates = 0, shadow = 0, source_id = ${source?.id ?? null}, discovered_at = now(), last_error = ${disc.blocked ? `homepage ${disc.homeStatus} (anti-bot)` : null}, updated_at = now() where id = ${seed.id}`);73 await bumpFactoryDaily({ seeds_processed: 1, requests: disc.requests, blocked: disc.blocked ? 1 : 0 });74 return { seedId: seed.id, sourceId: source?.id ?? null, status, candidates: 0, shadow: 0, requests: disc.requests, durationMs: Date.now() - started, notes };75 }7677 // ---- source: attach or create --------------------------------------------------------------------------78 let sourceId: string;79 if (source) {80 sourceId = source.id;81 if (!source.enabled) notes.push("source disabled in registry — candidates recorded, no shadow sensors");82 } else {83 sourceId = idx.ids.has(seed.id) ? `${seed.id}-${slugify(seed.domain.split(".")[0] ?? "x")}`.slice(0, 70) : seed.id;84 const seedDoc: SourceSeed = sourceSchema.parse({85 id: sourceId,86 name: seed.name,87 domain: seed.domain,88 homepage: seed.homepage ?? `https://${seed.domain}`,89 categories: seed.categories,90 tier: (seed.tier as Tier) ?? "B",91 weight: seed.weight,92 aliases: seed.aliases,93 country: seed.country ?? undefined,94 language: seed.language ?? undefined,95 first_party: seed.firstParty,96 notes: `Source Factory · ${seed.sector ?? "-"}/${seed.universe ?? "-"} · discovered ${new Date().toISOString().slice(0, 10)}`,97 llm: true,98 });99 await upsertSourceRecord(seedDoc, { origin: "factory", sector: seed.sector });100 idx.ids.add(sourceId);101 idx.byKey.set(key, { id: sourceId, domain: seed.domain, categories: seed.categories, tier: seed.tier, enabled: true });102 source = idx.byKey.get(key)!;103 }104105 // ---- existing sensors (this source + global URL uniqueness) ----------------------------------------------106 const existing = await db.execute<{ id: string; url: string; type: string; connector: string; config: Record<string, unknown>; status: string }>(sql`select id, url, type, connector, config, status from sensors where source_id = ${sourceId}`);107 const existingUrls = new Set(existing.rows.map((r) => r.url.replace(/\/$/, "")));108 const existingKinds = new Map<CandidateKind, number>();109 let existingTotal = 0;110 for (const r of existing.rows) {111 if (r.status === "DISABLED") continue;112 existingTotal++;113 const k = kindOfExisting(r);114 existingKinds.set(k, (existingKinds.get(k) ?? 0) + 1);115 }116 const urls = disc.candidates.map((c) => c.url);117 const global = urls.length ? await db.execute<{ url: string; source_id: string }>(sql`select url, source_id from sensors where url = any(${sql.raw("array[" + urls.map((u) => "'" + u.replace(/'/g, "''") + "'").join(",") + "]::text[]")}) and source_id <> ${sourceId}`) : { rows: [] as { url: string; source_id: string }[] };118 const ownedElsewhere = new Map(global.rows.map((r) => [r.url.replace(/\/$/, ""), r.source_id]));119120 // ---- score + select ---------------------------------------------------------------------------------------121 const scored = disc.candidates.map((c) => scoreCandidate(c, { importance: seed.importance, weight: seed.weight }));122 const fresh = scored.filter((s) => !existingUrls.has(s.cand.url.replace(/\/$/, "")) && !ownedElsewhere.has(s.cand.url.replace(/\/$/, "")));123 const { selected, skipped } = source.enabled ? selectForShadow(fresh, { kinds: existingKinds, total: existingTotal }) : { selected: [], skipped: fresh.map((s) => ({ s, reason: "source disabled" })) };124125 // ---- persist candidates ------------------------------------------------------------------------------------126 const day = new Date().toISOString().slice(0, 10);127 let shadowCreated = 0;128 for (const s of scored) {129 const k = s.cand.url.replace(/\/$/, "");130 let status = "candidate";131 let reason: string | null = null;132 if (existingUrls.has(k)) {133 status = "duplicate";134 reason = "already a sensor of this source";135 } else if (ownedElsewhere.has(k)) {136 status = "duplicate";137 reason = `already a sensor of ${ownedElsewhere.get(k)}`;138 } else if (selected.includes(s)) status = "shadow";139 else {140 const sk = skipped.find((x) => x.s === s);141 status = sk && /score/.test(sk.reason) ? "rejected" : "candidate";142 reason = sk?.reason ?? null;143 }144 const candId = newId("cand");145 let shadowSensorId: string | null = null;146 if (status === "shadow") {147 shadowSensorId = await createShadowSensor(sourceId, source, s.cand, candId, seed);148 if (shadowSensorId) shadowCreated++;149 else {150 status = "duplicate";151 reason = "sensor id/url collision";152 }153 }154 await db.execute(sql`155 insert into discovery_candidates (id, source_id, url, kind, evidence, score, status, seed_id, connector, type, name, kind_class, config, tier, score_value, reason, shadow_sensor_id, found_at, updated_at)156 values (${candId}, ${sourceId}, ${s.cand.url}, ${s.cand.type}, ${s.cand.evidence}, ${JSON.stringify({ value: s.cand.value, score: s.score, reasons: s.reasons, itemCount: s.cand.itemCount ?? null, itemsPerDay: s.cand.itemsPerDay ?? null, firstParty: s.cand.firstPartyConfidence, cost: s.cand.fetchCost, title: s.cand.title ?? null, fingerprint: s.cand.fingerprint ?? null })}::jsonb, ${status}, ${seed.id}, ${s.cand.connector}, ${s.cand.type}, ${s.cand.name}, ${s.cand.kind}, ${JSON.stringify(s.cand.config)}::jsonb, ${s.cand.suggestedTier}, ${s.score}, ${reason}, ${shadowSensorId}, now(), now())157 on conflict (source_id, url) do update set evidence = excluded.evidence, score = excluded.score, score_value = excluded.score_value, seed_id = excluded.seed_id, connector = excluded.connector, type = excluded.type, name = excluded.name, kind_class = excluded.kind_class, config = excluded.config, tier = excluded.tier, updated_at = now(),158 status = case when discovery_candidates.status in ('accepted','rejected','shadow') then discovery_candidates.status else excluded.status end,159 reason = case when discovery_candidates.status in ('accepted','rejected','shadow') then discovery_candidates.reason else excluded.reason end,160 shadow_sensor_id = coalesce(discovery_candidates.shadow_sensor_id, excluded.shadow_sensor_id)`);161 }162 for (const r of disc.rejected.slice(0, 40)) {163 await db.execute(sql`insert into discovery_candidates (id, source_id, url, kind, evidence, status, seed_id, reason, kind_class) values (${newId("cand")}, ${sourceId}, ${r.url}, 'HTML', ${r.reason}, 'rejected', ${seed.id}, ${r.reason}, 'other') on conflict (source_id, url) do nothing`);164 }165166 await db.execute(sql`update factory_seeds set status = 'discovered', source_id = ${sourceId}, candidates = ${scored.length}, shadow = shadow + ${shadowCreated}, discovered_at = now(), last_error = null, updated_at = now() where id = ${seed.id}`);167 await db.execute(sql`update sources set robots_checked_at = now(), updated_at = now() where id = ${sourceId}`);168 await bumpFactoryDaily({ seeds_processed: 1, requests: disc.requests, candidates: scored.length, shadow_created: shadowCreated, blocked: disc.blocked ? 1 : 0 });169 log.info({ seed: seed.id, source: sourceId, candidates: scored.length, shadow: shadowCreated, requests: disc.requests, ms: Date.now() - started, day }, "factory: organization discovered");170 return { seedId: seed.id, sourceId, status: "discovered", candidates: scored.length, shadow: shadowCreated, requests: disc.requests, durationMs: Date.now() - started, notes };171}172173function kindOfExisting(r: { url: string; type: string; connector: string; config: Record<string, unknown> }): CandidateKind {174 const k = (r.config as { kind?: string }).kind as CandidateKind | undefined;175 if (k) return k;176 if (r.connector === "statuspage" || r.connector === "statusjson") return "status";177 if (r.connector === "sitemap") return "sitemap";178 if (r.connector === "github") return "releases";179 if (r.connector === "edgar") return "filings";180 if (r.connector === "dns" || r.connector === "tls" || r.connector === "headers" || r.connector === "rdap") return "posture";181 if (r.connector === "openapi") return "api";182 if (r.connector === "rss") return /press/i.test(r.url) ? "press" : /blog/i.test(r.url) ? "blog" : /security|advisor/i.test(r.url) ? "security" : /changelog|release/i.test(r.url) ? "changelog" : "news";183 if (/pricing|plans/i.test(r.url)) return "pricing";184 if (/terms|privacy|legal/i.test(r.url)) return "legal";185 if (/career|jobs/i.test(r.url)) return "careers";186 if (/changelog|release-notes|whats-new/i.test(r.url)) return "changelog";187 if (/security|advisor/i.test(r.url)) return "security";188 return "other";189}190191async function createShadowSensor(sourceId: string, source: SourceRow, c: { url: string; type: string; connector: string; name: string; config: Record<string, unknown>; suggestedTier: Tier; kind: CandidateKind }, candId: string, seed: FactorySeed): Promise<string | null> {192 let id = `${sourceId}_${slugify(c.name)}`.slice(0, 80);193 const clash = (await db.execute<{ url: string }>(sql`select url from sensors where id = ${id}`)).rows[0];194 if (clash && clash.url.replace(/\/$/, "") !== c.url.replace(/\/$/, "")) id = `${id}-${newId("x").slice(-5)}`;195 const tier = c.suggestedTier;196 const priority = 3; // shadows never compete with production sensors197 const jitter = Math.floor(Math.random() * 30 * 60e3);198 try {199 await db200 .insert(sensors)201 .values({ id, sourceId, name: c.name, url: c.url, type: c.type, connector: c.connector, tier, importanceWeight: 1, config: { ...c.config, factory: true, shadow: true, seed: false, kind: c.kind, candidate: candId, seedId: seed.id, shadowSince: new Date().toISOString(), targetPriority: priorityFor(tier, source.categories) }, status: "SHADOW", priority, validatedAt: new Date(), nextCheckAt: new Date(Date.now() + jitter) })202 .onConflictDoNothing();203 } catch (e) {204 log.warn({ sensor: id, err: (e as Error).message }, "factory: shadow sensor insert failed");205 return null;206 }207 return id;208}209210export async function bumpFactoryDaily(delta: Partial<Record<"seeds_processed" | "requests" | "candidates" | "shadow_created" | "accepted" | "rejected" | "blocked", number>>): Promise<void> {211 const day = new Date().toISOString().slice(0, 10);212 const d = { seeds_processed: 0, requests: 0, candidates: 0, shadow_created: 0, accepted: 0, rejected: 0, blocked: 0, ...delta };213 await db.execute(sql`insert into factory_daily (day, seeds_processed, requests, candidates, shadow_created, accepted, rejected, blocked) values (${day}, ${d.seeds_processed}, ${d.requests}, ${d.candidates}, ${d.shadow_created}, ${d.accepted}, ${d.rejected}, ${d.blocked})214 on conflict (day) do update set seeds_processed = factory_daily.seeds_processed + ${d.seeds_processed}, requests = factory_daily.requests + ${d.requests}, candidates = factory_daily.candidates + ${d.candidates}, shadow_created = factory_daily.shadow_created + ${d.shadow_created}, accepted = factory_daily.accepted + ${d.accepted}, rejected = factory_daily.rejected + ${d.rejected}, blocked = factory_daily.blocked + ${d.blocked}`).catch(() => undefined);215}216217/** Claim up to `n` queued seeds (systemic organizations first). */218export async function claimSeeds(n: number, opts: { seedIds?: string[] } = {}): Promise<FactorySeed[]> {219 const claimed = await db.execute<Record<string, unknown>>(sql`220 update factory_seeds set status = 'discovering', attempts = attempts + 1, updated_at = now()221 where id in (222 select id from factory_seeds223 where ${opts.seedIds ? sql`id = any(${sql.raw("array[" + opts.seedIds.map((i) => "'" + i.replace(/'/g, "''") + "'").join(",") + "]::text[]")})` : sql`(status = 'queued' or (status = 'error' and attempts < 3 and updated_at < now() - interval '6 hours') or (status = 'discovering' and updated_at < now() - interval '15 minutes'))`}224 order by importance desc, created_at asc limit ${n} for update skip locked)225 returning *`);226 return claimed.rows.map(normalizeSeed);227}228229/** Run a seed with error capture (status → error, retried up to 3 times by the claim query). */230export async function processSeedSafe(seed: FactorySeed): Promise<ProcessOutcome> {231 try {232 return await processSeed(seed);233 } catch (e) {234 log.error({ seed: seed.id, err: (e as Error).stack ?? (e as Error).message }, "factory: seed failed");235 await db.execute(sql`update factory_seeds set status = 'error', last_error = ${(e as Error).message.slice(0, 500)}, updated_at = now() where id = ${seed.id}`).catch(() => undefined);236 return { seedId: seed.id, sourceId: null, status: "error", candidates: 0, shadow: 0, requests: 0, durationMs: 0, notes: [(e as Error).message] };237 }238}239240/** One batch: claim `n` seeds and process them with bounded concurrency (CLI / tests; the process uses a continuous pool). */241export async function runFactoryBatch(n = factoryConfig.concurrency, opts: { seedIds?: string[] } = {}): Promise<ProcessOutcome[]> {242 const seeds = await claimSeeds(n, opts);243 const out: ProcessOutcome[] = [];244 let i = 0;245 await Promise.all(246 Array.from({ length: Math.min(factoryConfig.concurrency, seeds.length) }, async () => {247 while (i < seeds.length) {248 const seed = seeds[i++]!;249 out.push(await processSeedSafe(seed));250 }251 }),252 );253 return out;254}255256function normalizeSeed(r: Record<string, unknown>): FactorySeed {257 const d = (v: unknown): Date | null => (v ? new Date(v as string) : null);258 return {259 id: r.id as string,260 name: r.name as string,261 domain: r.domain as string,262 homepage: (r.homepage as string | null) ?? null,263 categories: (r.categories as string[]) ?? [],264 country: (r.country as string | null) ?? null,265 language: (r.language as string | null) ?? null,266 tier: (r.tier as string) ?? "B",267 weight: Number(r.weight ?? 1),268 importance: Number(r.importance ?? 2),269 aliases: (r.aliases as string[]) ?? [],270 firstParty: Boolean(r.first_party ?? true),271 sector: (r.sector as string | null) ?? null,272 universe: (r.universe as string | null) ?? null,273 hints: (r.hints as Record<string, unknown>) ?? {},274 status: r.status as string,275 attempts: Number(r.attempts ?? 0),276 sourceId: (r.source_id as string | null) ?? null,277 candidates: Number(r.candidates ?? 0),278 shadow: Number(r.shadow ?? 0),279 accepted: Number(r.accepted ?? 0),280 rejected: Number(r.rejected ?? 0),281 lastError: (r.last_error as string | null) ?? null,282 discoveredAt: d(r.discovered_at),283 createdAt: d(r.created_at) ?? new Date(),284 updatedAt: d(r.updated_at) ?? new Date(),285 };286}287288export async function factoryStats(): Promise<Record<string, unknown>> {289 const [seeds, cands, shadows, daily, sectors] = await Promise.all([290 db.execute<Record<string, unknown>>(sql`select status, count(*)::int as n from factory_seeds group by status`).then((r) => r.rows),291 db.execute<Record<string, unknown>>(sql`select status, count(*)::int as n from discovery_candidates group by status`).then((r) => r.rows),292 db.execute<Record<string, unknown>>(sql`select count(*)::int as shadow, count(*) filter (where total_runs >= ${factoryConfig.shadowMinChecks})::int as ready, count(*) filter (where health <> 'UP')::int as unhealthy from sensors where status = 'SHADOW'`).then((r) => r.rows[0]),293 db.execute<Record<string, unknown>>(sql`select day::text as day, seeds_processed, requests, candidates, shadow_created, accepted, rejected, blocked from factory_daily order by day desc limit 30`).then((r) => r.rows),294 db.execute<Record<string, unknown>>(sql`select coalesce(sector, '-') as sector, count(*)::int as seeds, count(*) filter (where status = 'queued')::int as queued, count(*) filter (where status in ('discovered','done'))::int as discovered, count(*) filter (where status = 'blocked')::int as blocked, sum(shadow)::int as shadow, sum(accepted)::int as accepted, sum(rejected)::int as rejected from factory_seeds group by 1 order by seeds desc`).then((r) => r.rows),295 ]);296 return { seeds: Object.fromEntries(seeds.map((r) => [r.status, r.n])), candidates: Object.fromEntries(cands.map((r) => [r.status, r.n])), shadow: shadows, daily, sectors, config: { concurrency: factoryConfig.concurrency, budget: factoryConfig.budget, minScore: factoryConfig.minScore, shadowMinChecks: factoryConfig.shadowMinChecks, shadowMinHours: factoryConfig.shadowMinHours, shadowMaxHours: factoryConfig.shadowMaxHours, caps: factoryConfig.caps } };297}298299export { factorySeeds };300