TypeScript 55.4%
Python 43.2%
SQL 1.2%
1import { readdir, stat } from "node:fs/promises";2import { join, resolve } from "node:path";3import type { FastifyInstance } from "fastify";4import YAML from "yaml";5import { z } from "zod";6import { extendSchema, importDocumentSchema, sourceSchema, type SensorEndpoint, type Tier } from "@websensor/core";7import { getConnector, listConnectors, NormalizeError } from "@websensor/connectors";8import { db, sql, textArray } from "@websensor/db";9import { cacheStats, invalidate } from "./cache";10import { config } from "./config";11import { engineStatus, factoryStatus, liveStats } from "./live";1213/**14 * Internal operations (spec §70–72, §85). Enabled only when WS_ADMIN_TOKEN is set; every request15 * must carry `X-WebSensor-Admin: <token>`. Never linked from public pages; `/ops` in the web app16 * asks for the token and stores it in sessionStorage.17 */18export async function registerAdminRoutes(app: FastifyInstance): Promise<void> {19 app.addHook("onRequest", async (req, reply) => {20 if (!req.url.startsWith("/api/v1/admin")) return;21 if (!config.adminToken) return reply.status(404).send({ error: "not_found" });22 const t = req.headers["x-websensor-admin"];23 if (typeof t !== "string" || t !== config.adminToken) return reply.status(401).send({ error: "admin_token_required" });24 });2526 app.get("/api/v1/admin/ops", async () => {27 const t0 = Date.now();28 await db.execute(sql`select 1`);29 const dbLatency = Date.now() - t0;30 const [queue, failing, slow, throughput, storage, llm, jobs, es, byStatus, recentErrors] = await Promise.all([31 db.execute<Record<string, unknown>>(sql`select count(*) filter (where next_check_at <= now())::int as due, count(*) filter (where next_check_at <= now() - interval '10 minutes')::int as overdue_10m, count(*) filter (where priority = 0)::int as p0, count(*) filter (where priority = 0 and next_check_at <= now())::int as p0_due, min(next_check_at) as oldest_due from sensors where enabled`).then((r) => r.rows[0]),32 db.execute<Record<string, unknown>>(sql`select split_part(split_part(s.url, '/', 3), ':', 1) as host, count(*)::int as failures, count(distinct s.id)::int as sensors, max(r.error) as last_error, max(r.started_at) as last_at from sensor_runs r join sensors s on s.id = r.sensor_id where r.started_at >= now() - interval '6 hours' and r.outcome in ('error','parse_error','rate_limited') group by 1 order by failures desc limit 20`).then((r) => r.rows),33 db.execute<Record<string, unknown>>(sql`select s.id, s.name, s.source_id, s.url, s.connector, s.avg_latency_ms, s.total_runs from sensors s where s.enabled and s.avg_latency_ms is not null order by s.avg_latency_ms desc limit 20`).then((r) => r.rows),34 db.execute<Record<string, unknown>>(sql`35 select to_char(date_trunc('hour', started_at at time zone 'UTC'), 'YYYY-MM-DD"T"HH24:00:00"Z"') as t, count(*)::int as checks, count(*) filter (where outcome = 'not_modified')::int as not_modified, count(*) filter (where outcome in ('error','parse_error','rate_limited'))::int as errors, count(*) filter (where outcome = 'event')::int as events, avg(duration_ms)::int as avg_ms36 from sensor_runs where started_at >= now() - interval '24 hours' group by 1 order by 1`).then((r) => r.rows),37 Promise.all([38 db.execute<Record<string, unknown>>(sql`select pg_size_pretty(pg_database_size(current_database())) as db_size, pg_database_size(current_database())::bigint as db_bytes, (select count(*) from snapshots)::int as snapshots, (select count(*) from snapshots where storage_key is not null)::int as snapshots_with_raw, (select coalesce(sum(content_length),0) from snapshots where storage_key is not null)::bigint as raw_bytes_uncompressed, (select count(*) from changes)::int as changes, (select count(*) from events)::int as events, (select count(*) from sensor_runs)::int as runs`).then((r) => r.rows[0]),39 blobDirSize(),40 ]).then(([d, blobs]) => ({ ...d, blobs })),41 db.execute<Record<string, unknown>>(sql`select model, count(*)::int as calls, sum(input_tokens)::bigint as input_tokens, sum(output_tokens)::bigint as output_tokens, count(*) filter (where not ok)::int as failures from llm_usage where at >= now() - interval '24 hours' group by model`).then((r) => r.rows),42 db.execute<Record<string, unknown>>(sql`select outcome, count(*)::int as n from sensor_runs where started_at >= now() - interval '1 hour' group by outcome order by n desc`).then((r) => r.rows),43 engineStatus(),44 db.execute<Record<string, unknown>>(sql`select status, health, count(*)::int as n from sensors group by status, health order by n desc`).then((r) => r.rows),45 db.execute<Record<string, unknown>>(sql`select r.started_at, s.id as sensor_id, s.source_id, s.connector, r.http_status, r.outcome, r.error from sensor_runs r join sensors s on s.id = r.sensor_id where r.outcome in ('error','parse_error') and r.started_at >= now() - interval '1 hour' order by r.started_at desc limit 30`).then((r) => r.rows),46 ]);47 return { db: { latency_ms: dbLatency }, queue, engine: es, workers: es ? { inflight: es.inflight, concurrency: es.concurrency, busy_hosts: es.busyHosts, circuit_open: es.circuitOpen } : null, outcomes_1h: jobs, top_failing_domains: failing, slowest_sensors: slow, throughput_24h: throughput, storage, llm_24h: llm, sensors_by_status: byStatus, recent_errors: recentErrors, cache: cacheStats(), live: liveStats(), connectors: listConnectors().map((c) => c.metadata()), generated_at: new Date().toISOString() };48 });4950 // ---- Sensor actions ------------------------------------------------------------------------------51 app.post<{ Params: { id: string } }>("/api/v1/admin/sensors/:id/run-now", async (req, reply) => {52 const r = await db.execute(sql`update sensors set next_check_at = now() - interval '1 second', enabled = true where id = ${req.params.id}`);53 if (!r.rowCount) return reply.status(404).send({ error: "not_found" });54 return { ok: true, scheduled: "now" };55 });56 app.post<{ Params: { id: string } }>("/api/v1/admin/sensors/:id/enable", async (req, reply) => {57 const r = await db.execute(sql`update sensors set enabled = true, status = 'ACTIVE', consecutive_errors = 0, next_check_at = now(), updated_at = now() where id = ${req.params.id}`);58 if (!r.rowCount) return reply.status(404).send({ error: "not_found" });59 return { ok: true };60 });61 app.post<{ Params: { id: string } }>("/api/v1/admin/sensors/:id/disable", async (req, reply) => {62 const r = await db.execute(sql`update sensors set enabled = false, status = 'DISABLED', updated_at = now() where id = ${req.params.id}`);63 if (!r.rowCount) return reply.status(404).send({ error: "not_found" });64 return { ok: true };65 });66 app.patch<{ Params: { id: string } }>("/api/v1/admin/sensors/:id", async (req, reply) => {67 const body = z.object({ tier: z.enum(["S", "A", "B", "C", "D"]).optional(), priority: z.number().int().min(0).max(3).optional(), base_interval_seconds: z.number().int().min(15).max(7 * 86400).nullable().optional(), config: z.record(z.string(), z.unknown()).optional(), name: z.string().min(1).max(120).optional() }).parse(req.body ?? {});68 const sets: ReturnType<typeof sql>[] = [sql`updated_at = now()`];69 if (body.tier) sets.push(sql`tier = ${body.tier}`);70 if (body.priority !== undefined) sets.push(sql`priority = ${body.priority}`);71 if (body.base_interval_seconds !== undefined) sets.push(sql`base_interval_seconds = ${body.base_interval_seconds}`);72 if (body.config) sets.push(sql`config = config || ${JSON.stringify(body.config)}::jsonb`);73 if (body.name) sets.push(sql`name = ${body.name}`);74 const r = await db.execute(sql`update sensors set ${sql.join(sets, sql`, `)} where id = ${req.params.id}`);75 if (!r.rowCount) return reply.status(404).send({ error: "not_found" });76 return { ok: true };77 });78 /** Dry-run a connector against a URL (or an existing sensor) without persisting anything: inspect response + normalized output. */79 app.post("/api/v1/admin/sensors/test", async (req, reply) => {80 const body = z.object({ sensor_id: z.string().optional(), url: z.string().url().optional(), connector: z.string().default("http"), type: z.string().default("HTML"), config: z.record(z.string(), z.unknown()).default({}) }).parse(req.body ?? {});81 let endpoint: SensorEndpoint;82 if (body.sensor_id) {83 const s = (await db.execute<Record<string, unknown>>(sql`select * from sensors where id = ${body.sensor_id}`)).rows[0];84 if (!s) return reply.status(404).send({ error: "not_found" });85 endpoint = { id: String(s.id), sourceId: String(s.source_id), name: String(s.name), url: String(s.url), type: String(s.type) as SensorEndpoint["type"], tier: String(s.tier) as Tier, connector: String(s.connector), config: (s.config as Record<string, unknown>) ?? {}, etag: null, lastModified: null, state: null };86 } else if (body.url) {87 endpoint = { id: "admin_test", sourceId: "admin", name: "test", url: body.url, type: body.type as SensorEndpoint["type"], tier: "C", connector: body.connector, config: body.config, etag: null, lastModified: null, state: null };88 } else return reply.status(400).send({ error: "sensor_id_or_url_required" });89 let connector;90 try {91 connector = getConnector(endpoint.connector);92 } catch (e) {93 return reply.status(400).send({ error: "unknown_connector", detail: (e as Error).message });94 }95 const t0 = Date.now();96 const obs = await connector.fetch(endpoint);97 const fetchMs = Date.now() - t0;98 if (obs.error) return { ok: false, stage: "fetch", error: obs.error, meta: obs.meta, fetch_ms: fetchMs };99 if (obs.meta.status >= 400) return { ok: false, stage: "fetch", http_status: obs.meta.status, meta: obs.meta, fetch_ms: fetchMs };100 try {101 const norm = await connector.normalize(endpoint, obs);102 return { ok: true, fetch_ms: fetchMs, meta: obs.meta, normalized: { mode: norm.mode, title: norm.title ?? null, extractionConfidence: norm.extractionConfidence, items: norm.items?.length ?? null, sample_items: norm.items?.slice(0, 5) ?? null, text_preview: norm.text?.slice(0, 1500) ?? null, json_preview: norm.json !== undefined ? JSON.stringify(norm.json).slice(0, 1500) : null, canonicalHash: norm.canonicalHash, publishedAt: norm.publishedAt ?? null } };103 } catch (e) {104 return { ok: false, stage: "normalize", error: e instanceof NormalizeError ? { code: e.code, message: e.message } : { code: "error", message: (e as Error).message }, meta: obs.meta, fetch_ms: fetchMs, body_preview: obs.body?.toString("utf8").slice(0, 800) ?? null };105 }106 });107108 // ---- Source actions ------------------------------------------------------------------------------109 app.post<{ Params: { id: string } }>("/api/v1/admin/sources/:id/enable", async (req) => {110 await db.execute(sql`update sources set enabled = true, updated_at = now() where id = ${req.params.id}`);111 invalidate("");112 return { ok: true };113 });114 app.post<{ Params: { id: string } }>("/api/v1/admin/sources/:id/disable", async (req) => {115 await db.execute(sql`update sources set enabled = false, updated_at = now() where id = ${req.params.id}`);116 invalidate("");117 return { ok: true };118 });119 app.post<{ Params: { id: string } }>("/api/v1/admin/sources/:id/run-now", async (req) => {120 const r = await db.execute(sql`update sensors set next_check_at = now() - interval '1 second' where source_id = ${req.params.id} and enabled`);121 return { ok: true, sensors: r.rowCount ?? 0 };122 });123 app.post("/api/v1/admin/sensors/bulk", async (req) => {124 const body = z.object({ ids: z.array(z.string()).min(1).max(2000), action: z.enum(["enable", "disable", "run-now"]) }).parse(req.body ?? {});125 const arr = sql.raw("array[" + body.ids.map((i) => "'" + i.replace(/'/g, "''") + "'").join(",") + "]::text[]");126 let r;127 if (body.action === "enable") r = await db.execute(sql`update sensors set enabled = true, status = 'ACTIVE', consecutive_errors = 0, next_check_at = now() where id = any(${arr})`);128 else if (body.action === "disable") r = await db.execute(sql`update sensors set enabled = false, status = 'DISABLED' where id = any(${arr})`);129 else r = await db.execute(sql`update sensors set next_check_at = now() - interval '1 second' where id = any(${arr}) and enabled`);130 return { ok: true, affected: r.rowCount ?? 0 };131 });132133 /**134 * Bulk import (spec §71): JSON body `{ sources: [...] }` or raw YAML (content-type text/yaml). Validates with the135 * registry schema first; `dry_run=1` returns the validation report without writing. Imported sources are stored136 * with `kind = 'registry'` and `config.seed = false` so the file-based sync never disables them.137 */138 app.post<{ Querystring: { dry_run?: string } }>("/api/v1/admin/sources/import", async (req, reply) => {139 let doc: unknown = req.body;140 if (typeof req.body === "string") {141 try {142 doc = YAML.parse(req.body);143 } catch (e) {144 return reply.status(400).send({ error: "yaml_parse_error", detail: (e as Error).message });145 }146 }147 const parsed = importDocumentSchema.safeParse(doc);148 if (!parsed.success) return reply.status(400).send({ error: "invalid_document", issues: parsed.error.issues });149 const report: { id: string | undefined; ok: boolean; extend: boolean; issues?: string[]; sensors?: number }[] = [];150 const existing = new Set((await db.execute<{ id: string }>(sql`select id from sources`)).rows.map((r) => r.id));151 const valid: { extend: boolean; data: z.infer<typeof sourceSchema> | z.infer<typeof extendSchema> }[] = [];152 for (const s of parsed.data.sources) {153 const isExtend = (s as { extend?: boolean })?.extend === true;154 const res = isExtend ? extendSchema.safeParse(s) : sourceSchema.safeParse(s);155 const id = (s as { id?: string })?.id;156 if (!res.success) {157 report.push({ id, ok: false, extend: isExtend, issues: res.error.issues.map((i) => `${i.path.join(".")}: ${i.message}`) });158 continue;159 }160 if (isExtend && !existing.has(res.data.id)) {161 report.push({ id, ok: false, extend: true, issues: ["extend: true but source does not exist"] });162 continue;163 }164 if (!isExtend && existing.has(res.data.id)) {165 report.push({ id, ok: false, extend: false, issues: ["source id already exists (use extend: true)"] });166 continue;167 }168 valid.push({ extend: isExtend, data: res.data });169 report.push({ id, ok: true, extend: isExtend, sensors: res.data.sensors.length });170 }171 if (req.query.dry_run === "1") return { dry_run: true, valid: valid.length, invalid: report.filter((r) => !r.ok).length, report };172 let sensors = 0;173 for (const v of valid) {174 if (!v.extend) {175 const s = v.data as z.infer<typeof sourceSchema>;176 await db.execute(sql`insert into sources (id, name, domain, homepage, description, categories, tier, importance_weight, discover, fallback, notes, enabled, llm_enabled, first_party, country, language, kind)177 values (${s.id}, ${s.name}, ${s.domain}, ${s.homepage ?? `https://${s.domain}`}, ${s.description ?? null}, ${sql.raw("'{" + s.categories.map((c) => '"' + c.replace(/"/g, "") + '"').join(",") + "}'::text[]")}, ${s.tier}, ${s.weight}, ${JSON.stringify(s.discover)}::jsonb, ${JSON.stringify(s.fallback)}::jsonb, ${s.notes ?? null}, ${s.enabled}, ${s.llm}, ${s.first_party ?? true}, ${s.country ?? null}, ${s.language ?? null}, 'registry')`);178 const entId = `org_${s.id}`;179 await db.execute(sql`insert into entities (id, name, type, domain, homepage, description, importance, categories) values (${entId}, ${s.name}, ${s.entity_type}, ${s.domain}, ${s.homepage ?? `https://${s.domain}`}, ${s.description ?? null}, 60, ${sql.raw("'{" + s.categories.map((c) => '"' + c.replace(/"/g, "") + '"').join(",") + "}'::text[]")}) on conflict (id) do nothing`);180 await db.execute(sql`insert into source_entities (source_id, entity_id) values (${s.id}, ${entId}) on conflict do nothing`);181 for (const a of new Set([s.name, s.domain, ...s.aliases])) await db.execute(sql`insert into entity_aliases (alias, entity_id) values (${a.toLowerCase()}, ${entId}) on conflict do nothing`);182 }183 const base = v.data;184 for (const sen of base.sensors) {185 const id = sen.id ?? `${base.id}_${sen.name.toLowerCase().replace(/[^a-z0-9]+/g, "-").replace(/^-|-$/g, "")}`;186 await db.execute(sql`insert into sensors (id, source_id, name, url, type, connector, tier, importance_weight, config, base_interval_seconds, status, priority)187 values (${id}, ${base.id}, ${sen.name}, ${sen.url}, ${sen.type}, ${sen.connector}, ${sen.tier ?? (v.extend ? "B" : (v.data as z.infer<typeof sourceSchema>).tier)}, ${sen.weight ?? 1}, ${JSON.stringify({ ...sen.config, seed: false, imported_at: new Date().toISOString() })}::jsonb, ${sen.interval ?? null}, 'PENDING', 2)188 on conflict (id) do update set name = excluded.name, url = excluded.url, type = excluded.type, connector = excluded.connector, config = excluded.config, enabled = true, updated_at = now()`);189 sensors++;190 }191 }192 invalidate("");193 return { dry_run: false, imported_sources: valid.filter((v) => !v.extend).length, extended_sources: valid.filter((v) => v.extend).length, sensors, invalid: report.filter((r) => !r.ok).length, report };194 });195196 // ---- Source Factory ---------------------------------------------------------------------------------197 app.get("/api/v1/admin/factory", async () => {198 const [seeds, cands, shadows, daily, sectors, recent, hb, ready] = await Promise.all([199 db.execute<Record<string, unknown>>(sql`select status, count(*)::int as n from factory_seeds group by status`).then((r) => Object.fromEntries(r.rows.map((x) => [String(x.status), Number(x.n)]))),200 db.execute<Record<string, unknown>>(sql`select status, count(*)::int as n from discovery_candidates group by status`).then((r) => Object.fromEntries(r.rows.map((x) => [String(x.status), Number(x.n)]))),201 db.execute<Record<string, unknown>>(sql`select count(*)::int as total, count(*) filter (where health = 'UP')::int as up, count(*) filter (where health <> 'UP')::int as unhealthy, count(*) filter (where total_runs >= 5)::int as ready, coalesce(sum(raw_changes),0)::int as raw_changes from sensors where status = 'SHADOW' and enabled`).then((r) => r.rows[0]),202 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),203 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 = 'discovering')::int as discovering, count(*) filter (where status in ('discovered','done'))::int as discovered, count(*) filter (where status = 'blocked')::int as blocked, count(*) filter (where status = 'error')::int as errors, coalesce(sum(shadow),0)::int as shadow, coalesce(sum(accepted),0)::int as accepted, coalesce(sum(rejected),0)::int as rejected from factory_seeds group by 1 order by seeds desc`).then((r) => r.rows),204 db.execute<Record<string, unknown>>(sql`select id, name, domain, sector, status, candidates, shadow, accepted, rejected, last_error, discovered_at, source_id from factory_seeds where status <> 'queued' order by updated_at desc limit 40`).then((r) => r.rows),205 factoryStatus(),206 db.execute<Record<string, unknown>>(sql`select kind_class, count(*)::int as n from discovery_candidates where status = 'shadow' group by 1 order by n desc`).then((r) => r.rows),207 ]);208 return { seeds, candidates: cands, shadow: shadows, shadow_by_kind: ready, daily, sectors, recent_seeds: recent, heartbeat: hb, generated_at: new Date().toISOString() };209 });210 app.get<{ Querystring: { status?: string; sector?: string; q?: string; limit?: string } }>("/api/v1/admin/factory/seeds", async (req) => {211 const lim = Math.min(1000, Number(req.query.limit ?? 200));212 const rows = await db.execute<Record<string, unknown>>(sql`select * from factory_seeds where 1=1 ${req.query.status ? sql`and status = ${req.query.status}` : sql``} ${req.query.sector ? sql`and sector = ${req.query.sector}` : sql``} ${req.query.q ? sql`and (name ilike ${"%" + req.query.q + "%"} or domain ilike ${"%" + req.query.q + "%"})` : sql``} order by importance desc, updated_at desc limit ${lim}`);213 return { items: rows.rows };214 });215 /** Bulk seeds `{ seeds: [{ name, domain, categories?, country?, tier?, importance?, aliases?, sector?, universe?, hints? }] }` — queued for the factory process. */216 app.post("/api/v1/admin/factory/seeds", async (req) => {217 const seedSchema = z.object({ id: z.string().regex(/^[a-z0-9][a-z0-9-]*$/).optional(), name: z.string().min(1), domain: z.string().min(3), categories: z.array(z.string()).default([]), country: z.string().regex(/^[A-Z]{2,3}$/).optional(), language: z.string().optional(), tier: z.enum(["S", "A", "B", "C", "D"]).default("B"), importance: z.number().int().min(1).max(3).default(2), aliases: z.array(z.string()).default([]), first_party: z.boolean().default(true), sector: z.string().optional(), universe: z.string().optional(), hints: z.record(z.string(), z.unknown()).default({}) });218 const body = z.object({ seeds: z.array(seedSchema).min(1).max(2000) }).parse(req.body ?? {});219 let n = 0;220 for (const s of body.seeds) {221 const domain = s.domain.toLowerCase().replace(/^https?:\/\//, "").replace(/^www\./, "").replace(/\/.*$/, "");222 const id = s.id ?? domain.replace(/[^a-z0-9]+/g, "-").replace(/^-|-$/g, "").slice(0, 60);223 await db.execute(sql`insert into factory_seeds (id, name, domain, categories, country, language, tier, importance, aliases, first_party, sector, universe, hints, status)224 values (${id}, ${s.name}, ${domain}, ${textArray(s.categories)}, ${s.country ?? null}, ${s.language ?? null}, ${s.tier}, ${s.importance}, ${textArray(s.aliases)}, ${s.first_party}, ${s.sector ?? null}, ${s.universe ?? null}, ${JSON.stringify(s.hints)}::jsonb, 'queued')225 on conflict (id) do update set name = excluded.name, hints = factory_seeds.hints || excluded.hints, status = 'queued', updated_at = now()`);226 n++;227 }228 invalidate("coverage");229 return { ok: true, queued: n };230 });231 app.post<{ Params: { id: string } }>("/api/v1/admin/factory/seeds/:id/requeue", async (req, reply) => {232 const r = await db.execute(sql`update factory_seeds set status = 'queued', updated_at = now() where id = ${req.params.id}`);233 if (!r.rowCount) return reply.status(404).send({ error: "not_found" });234 return { ok: true };235 });236 app.get<{ Querystring: { status?: string; seed?: string; source?: string; kind?: string; limit?: string } }>("/api/v1/admin/factory/candidates", async (req) => {237 const lim = Math.min(1000, Number(req.query.limit ?? 200));238 const rows = await db.execute<Record<string, unknown>>(sql`select c.*, s.health as shadow_health, s.total_runs as shadow_runs, s.raw_changes as shadow_changes from discovery_candidates c left join sensors s on s.id = c.shadow_sensor_id where 1=1 ${req.query.status ? sql`and c.status = ${req.query.status}` : sql``} ${req.query.seed ? sql`and c.seed_id = ${req.query.seed}` : sql``} ${req.query.source ? sql`and c.source_id = ${req.query.source}` : sql``} ${req.query.kind ? sql`and c.kind_class = ${req.query.kind}` : sql``} order by c.updated_at desc limit ${lim}`);239 return { items: rows.rows };240 });241 /** Manual decision on a shadow sensor: accept now (→ ACTIVE) or reject (→ DISABLED). */242 app.post<{ Params: { id: string; action: string } }>("/api/v1/admin/factory/shadow/:id/:action", async (req, reply) => {243 const s = (await db.execute<Record<string, unknown>>(sql`select id, config, status from sensors where id = ${req.params.id}`)).rows[0];244 if (!s) return reply.status(404).send({ error: "not_found" });245 const reason = `manual ${req.params.action} via admin API`;246 if (req.params.action === "accept") {247 const tp = Number(((s.config as Record<string, unknown>).targetPriority as number | undefined) ?? 2);248 await db.execute(sql`update sensors set status = 'ACTIVE', enabled = true, priority = ${tp}, config = (config - 'shadow') || jsonb_build_object('acceptedAt', now()::text, 'shadowReport', ${reason}::text), updated_at = now() where id = ${req.params.id}`);249 await db.execute(sql`update discovery_candidates set status = 'accepted', reason = ${reason}, decided_at = now(), updated_at = now() where shadow_sensor_id = ${req.params.id}`);250 } else if (req.params.action === "reject") {251 await db.execute(sql`update sensors set status = 'DISABLED', enabled = false, config = config || jsonb_build_object('rejectedAt', now()::text, 'shadowReport', ${reason}::text), updated_at = now() where id = ${req.params.id}`);252 await db.execute(sql`update discovery_candidates set status = 'rejected', reason = ${reason}, decided_at = now(), updated_at = now() where shadow_sensor_id = ${req.params.id}`);253 } else return reply.status(400).send({ error: "action must be accept or reject" });254 invalidate("");255 return { ok: true };256 });257 /** Promote a plain candidate (never shadowed) straight into shadow monitoring. */258 app.post<{ Params: { id: string } }>("/api/v1/admin/factory/candidates/:id/shadow", async (req, reply) => {259 const c = (await db.execute<Record<string, unknown>>(sql`select * from discovery_candidates where id = ${req.params.id}`)).rows[0];260 if (!c) return reply.status(404).send({ error: "not_found" });261 if (!c.connector || !c.type) return reply.status(400).send({ error: "candidate has no connector/type (legacy discovery row)" });262 const src = (await db.execute<Record<string, unknown>>(sql`select categories from sources where id = ${String(c.source_id)}`)).rows[0];263 const id = `${String(c.source_id)}_${String(c.name ?? c.kind_class ?? "page").toLowerCase().replace(/[^a-z0-9]+/g, "-").replace(/^-|-$/g, "")}-${String(c.id).slice(-5)}`.slice(0, 80);264 await db.execute(sql`insert into sensors (id, source_id, name, url, type, connector, tier, config, status, priority, validated_at)265 values (${id}, ${String(c.source_id)}, ${String(c.name ?? c.kind_class ?? "page")}, ${String(c.url)}, ${String(c.type)}, ${String(c.connector)}, ${String(c.tier ?? "C")}, ${JSON.stringify({ ...((c.config as Record<string, unknown>) ?? {}), factory: true, shadow: true, seed: false, kind: c.kind_class, candidate: c.id, seedId: c.seed_id, shadowSince: new Date().toISOString(), targetPriority: 2, categories: src?.categories ?? [] })}::jsonb, 'SHADOW', 3, now())266 on conflict (id) do nothing`);267 await db.execute(sql`update discovery_candidates set status = 'shadow', shadow_sensor_id = ${id}, updated_at = now() where id = ${req.params.id}`);268 return { ok: true, sensor_id: id };269 });270271 app.get("/api/v1/admin/failures", async () => {272 const rows = await db.execute<Record<string, unknown>>(sql`select r.started_at, r.outcome, r.http_status, r.error, r.duration_ms, s.id as sensor_id, s.name, s.source_id, s.url, s.connector from sensor_runs r join sensors s on s.id = r.sensor_id where r.outcome in ('error','parse_error','rate_limited') and r.started_at >= now() - interval '24 hours' order by r.started_at desc limit 200`);273 return { items: rows.rows };274 });275}276277async function blobDirSize(): Promise<{ files: number; bytes: number } | null> {278 const root = resolve(process.env.BLOB_STORE_DIR ?? "./data/blobs");279 let files = 0;280 let bytes = 0;281 const started = Date.now();282 const walk = async (dir: string): Promise<void> => {283 if (Date.now() - started > 4000) return; // bounded284 let entries: string[];285 try {286 entries = await readdir(dir);287 } catch {288 return;289 }290 for (const e of entries) {291 const p = join(dir, e);292 const st = await stat(p).catch(() => null);293 if (!st) continue;294 if (st.isDirectory()) await walk(p);295 else {296 files++;297 bytes += st.size;298 }299 }300 };301 await walk(root);302 return { files, bytes };303}304