spb/social-runtime-crawler

1import http from "node:http";2import fs from "node:fs";3import path from "node:path";4import { fileURLToPath } from "node:url";5import { createLogger, isPlatform, loadConfig } from "@src/shared";6import { PostgresStore } from "@src/storage";7import { Queries } from "./queries.ts";8import { JobRunner } from "./jobs.ts";910/**11 * Control-plane + research API (§53, §63, §82). Plain node:http, JSON, SSE.12 * Consumed by the Next.js console (apps/dashboard) through its authenticated proxy; mutations require SRC_API_TOKEN.13 */14const log = createLogger("api");15const cfg = loadConfig();16const here = path.dirname(fileURLToPath(import.meta.url));17const publicDir = path.join(here, "..", "public");18const API_TOKEN = process.env.SRC_API_TOKEN ?? "";19const PORT = Number(process.env.SRC_API_PORT ?? cfg.dashboardPort);2021if (!cfg.databaseUrl) {22  console.error("SRC_DATABASE_URL is required for the API");23  process.exit(1);24}25const store = await PostgresStore.connect(cfg.databaseUrl);26await store.migrate();27const queries = new Queries(store, cfg);28const jobs = new JobRunner(store, cfg, Number(process.env.SRC_MAX_JOBS ?? 2));2930function json(res: http.ServerResponse, data: unknown, status = 200) {31  res.writeHead(status, { "content-type": "application/json; charset=utf-8", "cache-control": "no-store" });32  res.end(JSON.stringify(data));33}34function readBody(req: http.IncomingMessage): Promise<string> {35  return new Promise((resolve, reject) => {36    let s = "";37    req.on("data", (c) => {38      s += c;39      if (s.length > 1e6) reject(new Error("body too large"));40    });41    req.on("end", () => resolve(s));42    req.on("error", reject);43  });44}45function authorized(req: http.IncomingMessage): boolean {46  if (!API_TOKEN) return true; // dev47  return req.headers["x-src-token"] === API_TOKEN;48}4950const server = http.createServer(async (req, res) => {51  const url = new URL(req.url ?? "/", "http://localhost");52  const parts = url.pathname.split("/").filter(Boolean);53  const method = req.method ?? "GET";54  try {55    if (parts[0] !== "api" && parts[0] !== "media") {56      // legacy static console (apps/api/public) — kept for local debugging57      const file = path.join(publicDir, parts.length ? parts.join("/") : "index.html");58      if (!file.startsWith(publicDir) || !fs.existsSync(file)) return json(res, { error: "not found" }, 404);59      res.writeHead(200, { "content-type": file.endsWith(".html") ? "text/html; charset=utf-8" : "application/octet-stream" });60      return fs.createReadStream(file).pipe(res);61    }62    if (parts[0] === "media") {63      const file = path.join(cfg.mediaDir, ...parts.slice(1).map(decodeURIComponent));64      if (!file.startsWith(cfg.mediaDir) || !fs.existsSync(file)) return json(res, { error: "not found" }, 404);65      res.writeHead(200, { "content-type": "image/jpeg", "cache-control": "public, max-age=86400" });66      return fs.createReadStream(file).pipe(res);67    }68    if (method !== "GET" && !authorized(req)) return json(res, { error: "unauthorized" }, 401);69    const [, r1, r2, r3] = parts;7071    if (r1 === "health") return json(res, { ok: true, jobs_running: jobs.activeCount, db: true, ts: new Date().toISOString() });72    if (r1 === "overview") return json(res, await queries.overview());73    if (r1 === "platforms" && !r2) return json(res, queries.platforms());74    if (r1 === "platforms" && r2) return isPlatform(r2) ? json(res, queries.platformDetail(r2)) : json(res, { error: "unknown platform" }, 404);75    if (r1 === "sessions" && !r2) return json(res, await queries.sessions(Number(url.searchParams.get("limit") ?? 100)));76    if (r1 === "sessions" && r2) {77      const id = r2;78      if (!r3) {79        const s = await queries.session(id);80        return s ? json(res, s) : json(res, { error: "not found" }, 404);81      }82      if (r3 === "events") return json(res, await store.sessionEvents(id, { types: url.searchParams.get("types")?.split(",").filter(Boolean), limit: Number(url.searchParams.get("limit") ?? 300) }));83      if (r3 === "actions") return json(res, await store.sessionActions(id));84      if (r3 === "entities") return json(res, await store.sessionEntities(id, Number(url.searchParams.get("limit") ?? 500)));85      if (r3 === "media") return json(res, await store.sessionMedia(id));86      if (r3 === "pages") return json(res, await queries.pages(id));87      if (r3 === "world") return json(res, await queries.world(id));88      if (r3 === "schemas") {89        const s = await queries.session(id);90        return json(res, await queries.schemas(s?.platform));91      }92      if (r3 === "relationships") return json(res, await store.sessionRelationships(id));93    }94    if (r1 === "entities" && !r2) return json(res, await queries.searchEntities({ q: url.searchParams.get("q") ?? undefined, type: url.searchParams.get("type") ?? undefined, platform: url.searchParams.get("platform") ?? undefined, sort: url.searchParams.get("sort") ?? undefined, limit: Math.min(200, Number(url.searchParams.get("limit") ?? 50)), offset: Number(url.searchParams.get("offset") ?? 0) }));95    if (r1 === "entities" && r2) {96      const e = await queries.entity(decodeURIComponent(r2));97      return e ? json(res, e) : json(res, { error: "not found" }, 404);98    }99    if (r1 === "schemas") return json(res, await queries.schemas(url.searchParams.get("platform") ?? undefined));100    if (r1 === "stats") return json(res, await store.entityStats());101102    if (r1 === "stream") {103      // Server-Sent Events: poll the observations table and push new rows (§53 live panels).104      res.writeHead(200, { "content-type": "text/event-stream", "cache-control": "no-store", connection: "keep-alive", "x-accel-buffering": "no" });105      let since = url.searchParams.get("since") ?? new Date(Date.now() - 60_000).toISOString();106      const session = url.searchParams.get("session") ?? undefined;107      const types = url.searchParams.get("types")?.split(",").filter(Boolean);108      let alive = true;109      req.on("close", () => (alive = false));110      res.write(`event: hello\ndata: ${JSON.stringify({ since })}\n\n`);111      while (alive) {112        try {113          const rows = await queries.liveEvents(since, session, 200);114          for (const row of rows) {115            since = new Date(row.ts).toISOString();116            if (types && !types.includes(row.event_type)) continue;117            res.write(`event: observation\ndata: ${JSON.stringify(row)}\n\n`);118          }119          res.write(`: ping ${Date.now()}\n\n`);120        } catch (err) {121          res.write(`event: error\ndata: ${JSON.stringify({ error: (err as Error).message })}\n\n`);122        }123        await new Promise((r) => setTimeout(r, 1500));124      }125      return res.end();126    }127128    if (r1 === "jobs" && method === "GET" && !r2) return json(res, await queries.jobs());129    if (r1 === "jobs" && method === "POST" && !r2) {130      const body = JSON.parse((await readBody(req)) || "{}");131      try {132        return json(res, await jobs.start(body), 201);133      } catch (err) {134        return json(res, { error: (err as Error).message }, 400);135      }136    }137    if (r1 === "jobs" && r2 && r3 === "log") {138      res.writeHead(200, { "content-type": "text/plain; charset=utf-8", "cache-control": "no-store" });139      return res.end(jobs.logTail(r2, Number(url.searchParams.get("lines") ?? 200)));140    }141    if (r1 === "jobs" && r2 && r3 === "stop" && method === "POST") return json(res, { stopped: jobs.stop(r2) });142143    return json(res, { error: "not found" }, 404);144  } catch (err) {145    log.error("request failed", { url: req.url, err: (err as Error).message });146    json(res, { error: (err as Error).message }, 500);147  }148});149150server.listen(PORT, "127.0.0.1", () => log.info(`api → http://127.0.0.1:${PORT}  (token ${API_TOKEN ? "on" : "off"})`));151