spb/fetcha

1import { db, providerConfigs, providerHealth, domainProfiles, featureFlags, eq } from "@fetcha/db";2import { ProviderRegistry, type ProviderId } from "@fetcha/providers";3import { CircuitBreaker, FetchExecutor, RoutingEngine, type DomainKnowledge } from "@fetcha/routing";4import { getBrowserPool, type BrowserPool } from "@fetcha/browser";5import { newId } from "@fetcha/core";6import { config } from "../config";78export interface Engine {9  registry: ProviderRegistry;10  circuit: CircuitBreaker;11  routing: RoutingEngine;12  executor: FetchExecutor;13  browser: BrowserPool;14  /** Global kill-switch from the `browser_enabled` feature flag (default on). */15  browserEnabled: boolean;16  reload(): Promise<void>;17  probeAll(): Promise<void>;18  knowledge(domain: string): Promise<DomainKnowledge | null>;19}2021let _engine: Engine | null = null;2223export async function getEngine(): Promise<Engine> {24  if (_engine) return _engine;25  const circuit = new CircuitBreaker();26  let registry = new ProviderRegistry();27  let routing = new RoutingEngine(registry, circuit);28  let executor = new FetchExecutor(routing, circuit);29  const browser = getBrowserPool({ log: { info: (m) => console.log("[browser]", m), warn: (m) => console.warn("[browser]", m) } });30  let browserEnabled = true;3132  const engine: Engine = {33    get registry() {34      return registry;35    },36    circuit,37    browser,38    get browserEnabled() {39      return browserEnabled && browser.enabled;40    },41    get routing() {42      return routing;43    },44    get executor() {45      return executor;46    },47    async reload() {48      // Prices and enabled flags come from the DB; credentials always from the environment.49      const rows = await db.select().from(providerConfigs).catch(() => []);50      const prices: Partial<Record<ProviderId, Record<string, number>>> = {};51      const disabled: ProviderId[] = [];52      for (const r of rows) {53        prices[r.id as ProviderId] = r.pricePerGbUsd;54        if (!r.enabled) disabled.push(r.id as ProviderId);55      }56      registry = new ProviderRegistry({ prices, disabled });57      routing = new RoutingEngine(registry, circuit);58      executor = new FetchExecutor(routing, circuit);59      const [flag] = await db.select().from(featureFlags).where(eq(featureFlags.key, "browser_enabled")).limit(1).catch(() => []);60      browserEnabled = flag ? flag.enabled : true;61    },62    async probeAll() {63      const results = await Promise.all(64        registry.all().flatMap((p) => p.networks.map((n) => p.health(n).catch((e) => ({ provider: p.id, network: n, status: "down" as const, latencyMs: null, detail: (e as Error).message, checkedAt: new Date() })))),65      );66      for (const h of results) {67        await db68          .insert(providerHealth)69          .values({ id: newId("evt"), provider: h.provider, network: h.network, status: h.status, latencyMs: h.latencyMs, detail: h.detail ?? null, checkedAt: h.checkedAt })70          .catch(() => {});71      }72    },73    async knowledge(domain: string) {74      if (!domain) return null;75      const [row] = await db.select().from(domainProfiles).where(eq(domainProfiles.domain, domain)).limit(1).catch(() => []);76      if (!row) return null;77      return { domain, routeStats: row.routeStats ?? {}, policy: row.policy ?? null, browserRequiredRate: row.requests ? row.browserRequired / row.requests : 0, browserSamples: row.requests };78    },79  };80  await engine.reload();81  _engine = engine;82  return engine;83}8485export function startHealthLoop(engine: Engine, log: { info: (m: string) => void; warn: (m: string) => void }): () => void {86  let stopped = false;87  let ticks = 0;88  const run = async () => {89    if (stopped) return;90    try {91      await engine.reload();92      await engine.probeAll();93      log.info("provider health probe complete");94    } catch (e) {95      log.warn(`provider health probe failed: ${(e as Error).message}`);96    }97    // Retention prune roughly every 6 hours (every Nth probe), plus once shortly after boot.98    const every = Math.max(1, Math.round((6 * 60 * 60_000) / config.healthProbeIntervalMs));99    if (ticks % every === 0) {100      try {101        const { pruneRetention } = await import("./retention");102        await pruneRetention(log);103      } catch (e) {104        log.warn(`retention prune failed: ${(e as Error).message}`);105      }106    }107    ticks++;108  };109  const t = setTimeout(run, 5_000);110  const interval = setInterval(run, config.healthProbeIntervalMs);111  return () => {112    stopped = true;113    clearTimeout(t);114    clearInterval(interval);115  };116}117