TypeScript 55.4%
Python 43.2%
SQL 1.2%
1import { db, sensors, sources, sql, type Sensor, type Source } from "@websensor/db";2import { config, log } from "./config";3import { m } from "./metrics";4import { runSensor } from "./pipeline";5import { publishEngineStatus } from "./redis";67/**8 * Scheduler: claims due sensors with `FOR UPDATE SKIP LOCKED` (safe with several engine9 * processes), enforces global + per-host concurrency, and runs the pipeline. A claim moves10 * `next_check_at` 10 minutes ahead as a lease; the pipeline then sets the real next time.11 */12export class Scheduler {13 private inflight = 0;14 private perHost = new Map<string, number>();15 private stopped = false;16 private sourceCache = new Map<string, Source>();17 private sourceCacheAt = 0;18 private timer: NodeJS.Timeout | null = null;1920 async start(): Promise<void> {21 log.info({ concurrency: config.fetchConcurrency, perHost: config.perHostConcurrency }, "scheduler started");22 const tick = async (): Promise<void> => {23 if (this.stopped) return;24 try {25 await this.tick();26 } catch (e) {27 log.error({ err: (e as Error).message }, "scheduler tick failed");28 }29 this.timer = setTimeout(tick, this.inflight >= config.fetchConcurrency ? 500 : 800);30 };31 void tick();32 }3334 async stop(): Promise<void> {35 this.stopped = true;36 if (this.timer) clearTimeout(this.timer);37 const deadline = Date.now() + 30_000;38 while (this.inflight > 0 && Date.now() < deadline) await new Promise((r) => setTimeout(r, 200));39 }4041 private async refreshSources(): Promise<void> {42 if (Date.now() - this.sourceCacheAt < 60_000 && this.sourceCache.size) return;43 const rows = await db.select().from(sources);44 this.sourceCache = new Map(rows.map((r) => [r.id, r]));45 this.sourceCacheAt = Date.now();46 }4748 private lastStatusAt = 0;49 private hostFailures = new Map<string, { count: number; until: number }>();5051 /** Snapshot for the ops dashboard. */52 status(): Record<string, unknown> {53 const busyHosts = [...this.perHost.entries()].filter(([, n]) => n > 0).map(([h, n]) => ({ host: h, inflight: n }));54 const tripped = [...this.hostFailures.entries()].filter(([, f]) => f.until > Date.now()).map(([h, f]) => ({ host: h, failures: f.count, until: new Date(f.until).toISOString() }));55 return { inflight: this.inflight, concurrency: config.fetchConcurrency, perHostConcurrency: config.perHostConcurrency, busyHosts, circuitOpen: tripped, sources: this.sourceCache.size };56 }5758 /** Domain circuit breaker (spec §86): after 5 consecutive failures on a host, pause it for 10 minutes. */59 noteHostOutcome(host: string, failed: boolean): void {60 if (!failed) {61 this.hostFailures.delete(host);62 return;63 }64 const f = this.hostFailures.get(host) ?? { count: 0, until: 0 };65 f.count++;66 if (f.count >= 5) f.until = Date.now() + 10 * 60e3;67 this.hostFailures.set(host, f);68 }6970 private async tick(): Promise<void> {71 await this.refreshSources();72 const due = await db.execute<{ n: string }>(sql`select count(*)::text as n from sensors where enabled and next_check_at <= now()`);73 const dueN = Number(due.rows[0]?.n ?? 0);74 m.queueDue.set(dueN);75 if (Date.now() - this.lastStatusAt > 10_000) {76 this.lastStatusAt = Date.now();77 void publishEngineStatus({ ...this.status(), due: dueN, version: config.version });78 }79 const slots = config.fetchConcurrency - this.inflight;80 if (slots <= 0) return;81 // Priority-aware claim: critical sensors (priority 0) go first among what is due; within a priority the most82 // overdue first. We claim more candidates than free slots because many sensors share a few big hosts83 // (github.com, data.sec.gov…) and the per-host limit would otherwise leave slots idle.84 const claimed = await db.execute<Sensor>(sql`85 update sensors set next_check_at = now() + interval '10 minutes'86 where id in (select id from sensors where enabled and next_check_at <= now() order by priority asc, next_check_at asc limit ${slots * 4} for update skip locked)87 returning *`);88 let started = 0;89 for (const row of claimed.rows) {90 const sensor = normalizeRow(row as unknown as Record<string, unknown>);91 const host = safeHost(sensor.url);92 const hostLimit = config.perHostOverrides[host] ?? config.perHostConcurrency;93 const hostBusy = started >= slots || (this.perHost.get(host) ?? 0) >= hostLimit;94 if (hostBusy) {95 // give back the lease quickly (5–20 s, jittered so a big host drains smoothly)96 await db.update(sensors).set({ nextCheckAt: new Date(Date.now() + 5000 + Math.random() * 15000) }).where(sql`id = ${sensor.id}`);97 continue;98 }99 started++;100 const trip = this.hostFailures.get(host);101 if (trip && trip.until > Date.now()) {102 await db.update(sensors).set({ nextCheckAt: new Date(trip.until + Math.random() * 30_000) }).where(sql`id = ${sensor.id}`);103 continue;104 }105 const source = this.sourceCache.get(sensor.sourceId);106 if (!source || !source.enabled) {107 await db.update(sensors).set({ nextCheckAt: new Date(Date.now() + 3600e3) }).where(sql`id = ${sensor.id}`);108 continue;109 }110 this.inflight++;111 this.perHost.set(host, (this.perHost.get(host) ?? 0) + 1);112 m.inflight.set(this.inflight);113 void runSensor(sensor, source)114 .then((outcome) => this.noteHostOutcome(host, outcome === "error" || outcome === "rate_limited"))115 .catch(async (e) => {116 log.error({ sensor: sensor.id, err: (e as Error).stack ?? (e as Error).message }, "pipeline crashed");117 this.noteHostOutcome(host, true);118 await db.update(sensors).set({ nextCheckAt: new Date(Date.now() + 15 * 60e3), lastError: `pipeline: ${(e as Error).message}`.slice(0, 500), consecutiveErrors: sensor.consecutiveErrors + 1, health: "DEGRADED" }).where(sql`id = ${sensor.id}`).catch(() => undefined);119 })120 .finally(() => {121 this.inflight--;122 this.perHost.set(host, Math.max(0, (this.perHost.get(host) ?? 1) - 1));123 m.inflight.set(this.inflight);124 });125 }126 }127}128129/** Raw `returning *` rows come back snake_case; map to the Drizzle camelCase shape. */130function normalizeRow(r: Record<string, unknown>): Sensor {131 const d = (v: unknown): Date | null => (v ? new Date(v as string) : null);132 return {133 id: r.id as string,134 sourceId: r.source_id as string,135 name: r.name as string,136 url: r.url as string,137 type: r.type as string,138 connector: r.connector as string,139 tier: r.tier as string,140 importanceWeight: Number(r.importance_weight),141 config: (r.config ?? {}) as Record<string, unknown>,142 baseIntervalSeconds: (r.base_interval_seconds as number | null) ?? null,143 enabled: Boolean(r.enabled),144 health: r.health as string,145 nextCheckAt: d(r.next_check_at) ?? new Date(),146 lastCheckAt: d(r.last_check_at),147 lastChangeAt: d(r.last_change_at),148 lastEventAt: d(r.last_event_at),149 lastStatus: (r.last_status as number | null) ?? null,150 lastError: (r.last_error as string | null) ?? null,151 etag: (r.etag as string | null) ?? null,152 lastModified: (r.last_modified as string | null) ?? null,153 state: (r.state as Record<string, unknown> | null) ?? null,154 lastSnapshotId: (r.last_snapshot_id as string | null) ?? null,155 consecutiveErrors: Number(r.consecutive_errors ?? 0),156 totalRuns: Number(r.total_runs ?? 0),157 totalNotModified: Number(r.total_not_modified ?? 0),158 rawChanges: Number(r.raw_changes ?? 0),159 meaningfulChanges: Number(r.meaningful_changes ?? 0),160 avgLatencyMs: (r.avg_latency_ms as number | null) ?? null,161 status: (r.status as string) ?? "ACTIVE",162 validatedAt: d(r.validated_at),163 priority: Number(r.priority ?? 2),164 createdAt: d(r.created_at) ?? new Date(),165 updatedAt: d(r.updated_at) ?? new Date(),166 };167}168169function safeHost(u: string): string {170 try {171 return new URL(u).hostname;172 } catch {173 return u;174 }175}176177/** Roll up connector health from the last 24 h of runs. */178export async function rollupConnectorHealth(): Promise<void> {179 await db.execute(sql`180 insert into connector_health (connector, status, runs_24h, errors_24h, success_rate, avg_latency_ms, changes_24h, events_24h, last_success_at, last_error_at, last_error, http_codes, updated_at)181 select s.connector,182 case when count(r.id) = 0 then 'UP'183 when sum(case when r.outcome in ('error','parse_error') then 1 else 0 end)::float / count(r.id) > 0.5 then 'ERROR'184 when bool_or(r.outcome = 'rate_limited') and max(r.started_at) filter (where r.outcome = 'rate_limited') > now() - interval '1 hour' then 'RATE_LIMITED'185 when sum(case when r.outcome in ('error','parse_error') then 1 else 0 end)::float / count(r.id) > 0.15 then 'DEGRADED'186 else 'UP' end,187 count(r.id)::int,188 sum(case when r.outcome in ('error','parse_error','rate_limited') then 1 else 0 end)::int,189 case when count(r.id) = 0 then null else 1 - sum(case when r.outcome in ('error','parse_error','rate_limited') then 1 else 0 end)::float / count(r.id) end,190 avg(r.duration_ms)::int,191 (select count(*) from changes c join sensors s2 on s2.id = c.sensor_id where s2.connector = s.connector and c.detected_at >= now() - interval '24 hours')::int,192 (select count(*) from events e join sensors s3 on s3.id = e.sensor_id where s3.connector = s.connector and e.detected_at >= now() - interval '24 hours')::int,193 max(r.started_at) filter (where r.outcome not in ('error','parse_error','rate_limited')),194 max(r.started_at) filter (where r.outcome in ('error','parse_error','rate_limited')),195 (array_agg(r.error order by r.started_at desc) filter (where r.error is not null))[1],196 coalesce((select jsonb_object_agg(code, n) from (select coalesce(r2.http_status, 0)::text as code, count(*) as n from sensor_runs r2 join sensors s4 on s4.id = r2.sensor_id where s4.connector = s.connector and r2.started_at >= now() - interval '24 hours' group by 1) x), '{}'::jsonb),197 now()198 from sensors s left join sensor_runs r on r.sensor_id = s.id and r.started_at >= now() - interval '24 hours'199 group by s.connector200 on conflict (connector) do update set status = excluded.status, runs_24h = excluded.runs_24h, errors_24h = excluded.errors_24h, success_rate = excluded.success_rate, avg_latency_ms = excluded.avg_latency_ms, changes_24h = excluded.changes_24h, events_24h = excluded.events_24h, last_success_at = excluded.last_success_at, last_error_at = excluded.last_error_at, last_error = excluded.last_error, http_codes = excluded.http_codes, updated_at = now()`);201}202203/** Retention: raw fetch logs are short-lived; snapshots/events are kept. */204export async function pruneOldRuns(): Promise<void> {205 await db.execute(sql`delete from sensor_runs where started_at < now() - interval '14 days' and outcome in ('unchanged','not_modified')`);206 await db.execute(sql`delete from sensor_runs where started_at < now() - interval '60 days'`);207}208