Nightly embeddings on a schedule

"""Every night: embed what changed on a docs site into an index on a volume, then post a summary to a webhook.

pip install openai requests
Run it once by hand:   RUNCRATE_API_KEY=... INDEX_DIR=./index python nightly_embed.py https://www.runcrate.ai/docs/sitemap.xml
Schedule it:            python nightly_embed.py --schedule <volume-id> <script-url> <sitemap-url>
On the schedule, RUNCRATE_API_KEY and WEBHOOK_URL come from your environment's secrets, never from the command.
"""
import hashlib, hmac, json, os, re, sys, time
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

import requests
from openai import OpenAI

EMBED = "Qwen/Qwen3-Embedding-0.6B"
INDEX_DIR = Path(os.environ.get("INDEX_DIR", "/workspace/docs-index"))  # /workspace is the volume on a Runcrate machine

def fetch(url: str) -> tuple[str, str | None]:
    for attempt in range(4):
        try:
            r = requests.get(url + ".md", timeout=30, headers={"User-Agent": "nightly-embed"})  # docs pages answer markdown at .md
            if r.status_code == 200:
                return url, r.text
            if r.status_code == 404:
                return url, None
        except requests.RequestException:
            pass
        time.sleep(2 ** attempt)
    return url, None

def notify(summary: dict) -> int:
    url = os.environ.get("WEBHOOK_URL")
    if not url:
        return 0
    text = f"Docs index: {summary['added']} added, {summary['changed']} changed, {summary['removed']} removed, {summary['pages']} pages, ${summary['cost_usd']:.5f}"
    body = json.dumps({"text": text, **summary}).encode()  # "text" is what Slack and Discord-compatible hooks show
    sig = hmac.new(os.environ.get("WEBHOOK_SECRET", "").encode(), body, hashlib.sha256).hexdigest()
    return requests.post(url, data=body, headers={"Content-Type": "application/json", "X-Signature-SHA256": sig}, timeout=15).status_code

def run(sitemap: str) -> None:
    client = OpenAI(base_url="https://api.runcrate.ai/v1", api_key=os.environ["RUNCRATE_API_KEY"], max_retries=4)
    t0 = time.time()
    INDEX_DIR.mkdir(parents=True, exist_ok=True)
    manifest_path, vectors_path = INDEX_DIR / "manifest.json", INDEX_DIR / "vectors.jsonl"
    manifest = json.loads(manifest_path.read_text()) if manifest_path.exists() else {}
    vectors = {}
    if vectors_path.exists():
        for line in vectors_path.read_text().splitlines():
            row = json.loads(line); vectors[row["url"]] = row

    urls = re.findall(r"<loc>([^<]+)</loc>", requests.get(sitemap, timeout=30, headers={"User-Agent": "nightly-embed"}).text)
    with ThreadPoolExecutor(max_workers=8) as pool:
        pages = {u: t for u, t in pool.map(fetch, urls) if t}
    hashes = {u: hashlib.sha256(t.encode()).hexdigest() for u, t in pages.items()}
    added = [u for u in pages if u not in manifest]
    changed = [u for u in pages if u in manifest and manifest[u] != hashes[u]]
    removed = [u for u in manifest if u not in pages]

    cost, todo = 0.0, added + changed
    for i in range(0, len(todo), 32):
        batch = todo[i:i + 32]
        # One vector per page, from its first 6,000 characters; split pages into sections for finer retrieval.
        raw = client.embeddings.with_raw_response.create(model=EMBED, input=[pages[u][:6000] for u in batch])
        cost += float(raw.headers.get("x-llm-cost", 0))
        for u, d in zip(batch, raw.parse().data):
            title = (re.search(r"^# (.+)$", pages[u], re.M) or re.search(r"title: (.+)", pages[u]) or [None, u])[1]
            vectors[u] = {"url": u, "title": title.strip(), "hash": hashes[u], "model": EMBED, "vector": d.embedding}
    for u in removed:
        vectors.pop(u, None)

    vectors_path.write_text("".join(json.dumps(v) + "\n" for v in vectors.values()))
    manifest_path.write_text(json.dumps(hashes, indent=0))
    summary = {"pages": len(pages), "added": len(added), "changed": len(changed), "removed": len(removed),
               "embedded": len(todo), "cost_usd": cost, "seconds": round(time.time() - t0, 1), "model": EMBED}
    summary["webhook"] = notify(summary)
    print(json.dumps(summary))
    Path("run.json").write_text(json.dumps(summary, indent=2))

def schedule(volume_id: str, script_url: str, sitemap: str) -> None:
    """Create the nightly schedule through the API. The command fetches this script, so the volume holds only the index."""
    res = requests.post(
        "https://www.runcrate.ai/api/v1/schedules",
        headers={"Authorization": f"Bearer {os.environ['RUNCRATE_API_KEY']}", "Content-Type": "application/json"},
        json={
            "kind": "job", "name": "nightly-embed", "cron": "0 3 * * *", "timezone": "Europe/London",
            "spec": {
                "command": f"curl -fsSL '{script_url}' -o /tmp/nightly_embed.py && pip install -q openai requests && python3 /tmp/nightly_embed.py '{sitemap}'",
                "gpu_type": os.environ.get("NIGHTLY_GPU", "A4000"),  # the job only calls the API: take the lowest-priced GPU
                "max_runtime_minutes": 15,
                "storage_id": volume_id,
            },
        },
        timeout=30,
    )
    res.raise_for_status()
    s = res.json().get("data", res.json())
    print(f"schedule {s['id']} next run {s.get('next_run_at')}")

if __name__ == "__main__":
    if sys.argv[1] == "--schedule":
        schedule(*sys.argv[2:5])
    else:
        run(sys.argv[1])

// Every night: embed what changed on a docs site into an index on a volume, then post a summary to a webhook.
//
// npm install openai
// Run it once by hand:  RUNCRATE_API_KEY=... INDEX_DIR=./index npx tsx nightly-embed-schedule.ts https://www.runcrate.ai/docs/sitemap.xml
// Schedule it:           npx tsx nightly-embed-schedule.ts --schedule <volume-id> <script-url> <sitemap-url>
// On the schedule, RUNCRATE_API_KEY and WEBHOOK_URL come from your environment's secrets, never from the command.
import { createHash, createHmac } from "node:crypto";
import { existsSync, mkdirSync, readFileSync, writeFileSync } from "node:fs";
import { join } from "node:path";
import OpenAI from "openai";

const EMBED = "Qwen/Qwen3-Embedding-0.6B";
const INDEX_DIR = process.env.INDEX_DIR ?? "/workspace/docs-index"; // /workspace is the volume on a Runcrate machine
const sleep = (ms: number) => new Promise<void>((r) => setTimeout(r, ms));

async function fetchPage(url: string): Promise<[string, string | null]> {
  for (let attempt = 0; attempt < 4; attempt++) {
    try {
      const r = await fetch(`${url}.md`, { headers: { "User-Agent": "nightly-embed" }, signal: AbortSignal.timeout(30_000) }); // docs pages answer markdown at .md
      if (r.status === 200) return [url, await r.text()];
      if (r.status === 404) return [url, null];
    } catch {
      // network error: retry
    }
    await sleep(2 ** attempt * 1000);
  }
  return [url, null];
}

type Summary = { pages: number; added: number; changed: number; removed: number; embedded: number; cost_usd: number; seconds: number; model: string; webhook?: number };

async function notify(summary: Summary): Promise<number> {
  const url = process.env.WEBHOOK_URL;
  if (!url) return 0;
  const text = `Docs index: ${summary.added} added, ${summary.changed} changed, ${summary.removed} removed, ${summary.pages} pages, $${summary.cost_usd.toFixed(5)}`;
  const body = JSON.stringify({ text, ...summary }); // "text" is what Slack and Discord-compatible hooks show
  const sig = createHmac("sha256", process.env.WEBHOOK_SECRET ?? "").update(body).digest("hex");
  const res = await fetch(url, { method: "POST", body, headers: { "Content-Type": "application/json", "X-Signature-SHA256": sig } });
  return res.status;
}

async function run(sitemap: string) {
  const client = new OpenAI({ baseURL: "https://api.runcrate.ai/v1", apiKey: process.env.RUNCRATE_API_KEY, maxRetries: 4 });
  const t0 = Date.now();
  mkdirSync(INDEX_DIR, { recursive: true });
  const manifestPath = join(INDEX_DIR, "manifest.json");
  const vectorsPath = join(INDEX_DIR, "vectors.jsonl");
  const manifest: Record<string, string> = existsSync(manifestPath) ? JSON.parse(readFileSync(manifestPath, "utf8")) : {};
  type Row = { url: string; title: string; hash: string; model: string; vector: number[] };
  const vectors = new Map<string, Row>();
  if (existsSync(vectorsPath)) {
    for (const line of readFileSync(vectorsPath, "utf8").split("\n").filter(Boolean)) {
      const row: Row = JSON.parse(line);
      vectors.set(row.url, row);
    }
  }

  const xml = await (await fetch(sitemap, { headers: { "User-Agent": "nightly-embed" } })).text();
  const urls = [...xml.matchAll(/<loc>([^<]+)<\/loc>/g)].map((m) => m[1]);
  const pages = new Map<string, string>();
  for (let i = 0; i < urls.length; i += 8) {
    for (const [u, t] of await Promise.all(urls.slice(i, i + 8).map(fetchPage))) if (t) pages.set(u, t);
  }
  const hashes: Record<string, string> = {};
  for (const [u, t] of pages) hashes[u] = createHash("sha256").update(t).digest("hex");
  const added = [...pages.keys()].filter((u) => !(u in manifest));
  const changed = [...pages.keys()].filter((u) => u in manifest && manifest[u] !== hashes[u]);
  const removed = Object.keys(manifest).filter((u) => !pages.has(u));

  let cost = 0;
  const todo = [...added, ...changed];
  for (let i = 0; i < todo.length; i += 32) {
    const batch = todo.slice(i, i + 32);
    // One vector per page, from its first 6,000 characters; split pages into sections for finer retrieval.
    const { data, response } = await client.embeddings.create({ model: EMBED, input: batch.map((u) => (pages.get(u) ?? "").slice(0, 6000)) }).withResponse();
    cost += Number(response.headers.get("x-llm-cost") ?? 0);
    batch.forEach((u, j) => {
      const page = pages.get(u) ?? "";
      const title = (page.match(/^# (.+)$/m)?.[1] ?? page.match(/title: (.+)/)?.[1] ?? u).trim();
      vectors.set(u, { url: u, title, hash: hashes[u], model: EMBED, vector: data.data[j].embedding });
    });
  }
  for (const u of removed) vectors.delete(u);

  writeFileSync(vectorsPath, [...vectors.values()].map((v) => `${JSON.stringify(v)}\n`).join(""));
  writeFileSync(manifestPath, JSON.stringify(hashes));
  const summary: Summary = {
    pages: pages.size, added: added.length, changed: changed.length, removed: removed.length,
    embedded: todo.length, cost_usd: cost, seconds: Math.round((Date.now() - t0) / 100) / 10, model: EMBED,
  };
  summary.webhook = await notify(summary);
  console.log(JSON.stringify(summary));
  writeFileSync("run.json", JSON.stringify(summary, null, 2));
}

/** Create the nightly schedule through the API. The command fetches this script, so the volume holds only the index. */
async function schedule(volumeId: string, scriptUrl: string, sitemap: string) {
  const res = await fetch("https://www.runcrate.ai/api/v1/schedules", {
    method: "POST",
    headers: { Authorization: `Bearer ${process.env.RUNCRATE_API_KEY}`, "Content-Type": "application/json" },
    body: JSON.stringify({
      kind: "job",
      name: "nightly-embed",
      cron: "0 3 * * *",
      timezone: "Europe/London",
      spec: {
        // The job runs in a Node image; /workspace (the volume) is mounted into it.
        image: "node:22",
        command: `curl -fsSL '${scriptUrl}' -o /tmp/nightly.ts && npm i -s --prefix /tmp openai && cd /tmp && npx -y tsx nightly.ts '${sitemap}'`,
        gpu_type: process.env.NIGHTLY_GPU ?? "A4000", // the job only calls the API: take the lowest-priced GPU
        max_runtime_minutes: 15,
        storage_id: volumeId,
      },
    }),
  });
  if (!res.ok) throw new Error(`schedule create answered ${res.status}: ${await res.text()}`);
  const body = (await res.json()) as { id?: string; next_run_at?: string; data?: { id: string; next_run_at?: string } };
  const s = body.data ?? body;
  console.log(`schedule ${s.id} next run ${s.next_run_at}`);
}

const args = process.argv.slice(2);
if (args[0] === "--schedule") await schedule(args[1], args[2], args[3]);
else await run(args[0]);