jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344import { safeParse, array, object, string } from "@atcute/lexicons/validations";
import { manifestSchema, snapshotResponse } from "./snapshot";
import pageSource from "./page.html";import ui from "./ui.js";
/** the page names its script by content hash so an html/js cache skew on deploy cannot happen. */function contentHash(text: string): string { let h = 2166136261; for (let i = 0; i < text.length; i++) h = Math.imul(h ^ text.charCodeAt(i), 16777619); return (h >>> 0).toString(36);}const UI_VERSION = contentHash(ui);const page = pageSource.replace('src="ui.js"', `src="ui.js?v=${UI_VERSION}"`);
interface Env extends Bindings { readonly STRATA_INGEST_TOKEN: string;}
const MAX_ARCHIVE_OBJECT_BYTES = 200 << 20;const ARCHIVE_KEY = /^(events\/ns=[a-z0-9._-]+\/seg=\d+\.parquet|agg\/seg=\d+\.parquet|manifest\/seg=\d+\.json)$/;
/** the drain uploads one object per (segment, partition). app.bsky raw rows are the only private thing: * everything else — tail namespaces, marker rows, per-segment aggregates — is public by decision (2026-08-29). */function bucketFor(key: string, env: Env): R2Bucket { return key.startsWith("events/ns=app.bsky/") ? env.PRIVATE : env.PUBLIC;}
async function putArchive( request: Request, key: string, env: Env,): Promise<Response> { if (!authorized(request, env)) return json({ error: "unauthorized" }, 401); if (!ARCHIVE_KEY.test(key)) return json( { error: "key must be events/ns=<ns>/seg=<n>.parquet, agg/seg=<n>.parquet, or manifest/seg=<n>.json", }, 400, ); const length = Number(request.headers.get("content-length") ?? "0"); if ( !Number.isFinite(length) || length <= 0 || length > MAX_ARCHIVE_OBJECT_BYTES ) return json({ error: "content-length required, ≤ 200 MiB" }, 413); if (request.body === null) return json({ error: "empty body" }, 400); const bucket = bucketFor(key, env); try { const object = await bucket.put(key, request.body, { httpMetadata: { contentType: key.endsWith(".json") ? "application/json" : "application/vnd.apache.parquet", }, }); return json({ key, size: object.size, etag: object.httpEtag, bucket: bucket === env.PRIVATE ? "private" : "public", }); } catch (err) { // an R2 hiccup surfaces here as a 503 with the message, not a bare 1101; the drain retries on 5xx return json( { error: err instanceof Error ? err.message : "put failed" }, 503, ); }}
/** what this worker answers with: plain JSON built from our own row types (type aliases, so arrays of them fit). */type JsonValue = | string | number | boolean | null | undefined | readonly JsonValue[] | { readonly [key: string]: JsonValue };
function json( body: JsonValue, status = 200, headers: Readonly<Record<string, string>> = {},): Response { return new Response(JSON.stringify(body), { status, headers: { "content-type": "application/json", "access-control-allow-origin": "*", ...headers, }, });}
function timingSafeEqual(a: string, b: string): boolean { const enc = new TextEncoder(); const ab = enc.encode(a); const bb = enc.encode(b); if (ab.byteLength !== bb.byteLength) return false; return crypto.subtle.timingSafeEqual(ab, bb);}
function authorized(request: Request, env: Env): boolean { const header = request.headers.get("authorization") ?? ""; const token = header.startsWith("Bearer ") ? header.slice("Bearer ".length) : ""; return token.length > 0 && timingSafeEqual(token, env.STRATA_INGEST_TOKEN);}
const PUBLIC_BASE = "https://pub-735e1688181b45e49c925eed81fe2da7.r2.dev";
/** which drained segments hold rows for a namespace, as public parquet URLs the page can hand to duckdb-wasm. * app.bsky is private by policy, so it is refused here. */async function partitions(url: URL, env: Env): Promise<Response> { const ns = url.searchParams.get("ns"); if (ns === null || !/^[a-z0-9-]+\.[a-z0-9-]+$/i.test(ns)) return json({ error: "ns must be an authority like fm.teal" }, 400); if (ns === "app.bsky") return json( { error: "app.bsky rows are not public; use the aggregates" }, 403, ); const listed = await env.PUBLIC.list({ prefix: `events/ns=${ns}/`, limit: 1000, }); const urls = listed.objects.map((o) => `${PUBLIC_BASE}/${o.key}`); const bytes = listed.objects.reduce((a, o) => a + o.size, 0); return json({ ns, files: urls, bytes, truncated: listed.truncated }, 200, { "cache-control": "public, max-age=120", });}
async function publishSnapshot( request: Request, env: Env, key: string,): Promise<Response> { if (!authorized(request, env)) return json({ error: "unauthorized" }, 401); if (key === "checkpoint" && request.method === "GET") { const saved = await env.PRIVATE.get("snapshots/checkpoint"); return saved ? new Response(saved.body) : json({ error: "not found" }, 404); } if (request.method === "DELETE" && /^objects\/[a-f0-9]{64}$/.test(key)) { const current = await env.PRIVATE.get("snapshots/manifest"); if (current) { const parsed = safeParse(manifestSchema, await current.json()); if (!parsed.ok) return json({ error: "invalid current manifest" }, 500); if (parsed.value.objects.some((o) => `objects/${o.digest}` === key)) return json({ error: "object in use" }, 409); } await env.PRIVATE.delete(`snapshots/${key}`); return json({ deleted: key }); } if (request.method !== "PUT") return json({ error: "method not allowed" }, 405); if (key === "batch") { const length = Number(request.headers.get("content-length") ?? "0"); if (!Number.isSafeInteger(length) || length <= 0 || length > 16 << 20) return json({ error: "invalid content-length" }, 413); const parsed = safeParse( array(object({ digest: string(), body: string() })), await request.json(), ); if (!parsed.ok || parsed.value.length > 64) return json({ error: "invalid batch" }, 400); for (const item of parsed.value) { const digest = [ ...new Uint8Array( await crypto.subtle.digest( "SHA-256", new TextEncoder().encode(item.body), ), ), ] .map((b) => b.toString(16).padStart(2, "0")) .join(""); if (digest !== item.digest) return json({ error: "digest mismatch" }, 400); } await Promise.all( parsed.value.map((item) => env.PRIVATE.put(`snapshots/objects/${item.digest}`, item.body, { httpMetadata: { contentType: "application/json" }, }), ), ); return json({ saved: parsed.value.length }); }
if ( key !== "manifest" && key !== "checkpoint" && !/^objects\/[a-f0-9]{64}$/.test(key) ) return json({ error: "invalid key" }, 400); const length = Number(request.headers.get("content-length") ?? "0"); if ( !Number.isSafeInteger(length) || length <= 0 || length > (key === "checkpoint" ? 128 << 20 : 16 << 20) ) return json({ error: "invalid content-length" }, 413); if (key === "checkpoint") { await env.PRIVATE.put("snapshots/checkpoint", request.body); return json({ saved: key }); } const body = await request.arrayBuffer(); if (body.byteLength !== length) return json({ error: "length mismatch" }, 400); if (key.startsWith("objects/")) { const digest = [ ...new Uint8Array(await crypto.subtle.digest("SHA-256", body)), ] .map((b) => b.toString(16).padStart(2, "0")) .join(""); if (key !== `objects/${digest}`) return json({ error: "digest mismatch" }, 400); } else { const parsed = safeParse( manifestSchema, JSON.parse(new TextDecoder().decode(body)), ); if ( !parsed.ok || parsed.value.objects.some((o) => !/^[a-f0-9]{64}$/.test(o.digest)) ) return json({ error: "invalid manifest" }, 400); for (const name of [ "progress", "segments", "checksums", "collections", "namespaces", "heat/namespaces", "heat/collections", ]) { const digest = parsed.value.objects.find((o) => o.name === name)?.digest; if (!digest || !(await env.PRIVATE.head(`snapshots/objects/${digest}`))) return json({ error: `missing ${name}` }, 400); } const previous = await env.PRIVATE.get("snapshots/manifest"); if (previous) await env.PRIVATE.put("snapshots/previous-manifest", previous.body); } await env.PRIVATE.put(`snapshots/${key}`, body, { httpMetadata: { contentType: "application/json" }, }); return json({ saved: key });}
export default { async fetch( request: Request, env: Env, ctx: ExecutionContext, ): Promise<Response> { const url = new URL(request.url); if (url.pathname.startsWith("/snapshot/")) return publishSnapshot( request, env, url.pathname.slice("/snapshot/".length), ); const preview = url.pathname.startsWith("/snapshot-preview/api/"); const apiPath = preview ? url.pathname.slice("/snapshot-preview".length) : url.pathname; if ( request.method === "GET" && [ "/api/progress", "/api/checksums", "/api/collections", "/api/namespaces", "/api/heat", ].includes(apiPath) ) { const cache = caches.default; const cacheUrl = new URL(request.url); cacheUrl.searchParams.sort(); const cacheKey = new Request(cacheUrl); const cached = await cache.match(cacheKey); if (cached) return cached; url.pathname = apiPath; const response = await snapshotResponse(url, env.PRIVATE, ctx); if (response.ok) ctx.waitUntil(cache.put(cacheKey, response.clone())); return response; } if (request.method === "POST" && url.pathname === "/ingest") return json({ error: "ingestion moved to HeavyPad" }, 410); if (request.method === "PUT" && url.pathname.startsWith("/archive/")) return putArchive(request, url.pathname.slice("/archive/".length), env); if (request.method !== "GET") return json({ error: "method not allowed" }, 405); switch (url.pathname) { case "/api/partitions": return partitions(url, env); case "/": return new Response(page, { headers: { "content-type": "text/html; charset=utf-8", "cache-control": "public, max-age=60", }, }); case "/ui.js": return new Response(ui, { headers: { "content-type": "text/javascript; charset=utf-8", "cache-control": url.searchParams.get("v") === UI_VERSION ? "public, max-age=31536000, immutable" : "no-cache", }, }); case "/api": return json({ name: "strata", what: "per-segment, per-collection event counts of the stream.waow.tech archive", endpoints: [ "/api/progress", "/api/heat?step=N&top=N&group=namespaces|collections", "/api/heat?step=N&ns=<authority>", "/api/namespaces", "/api/collections?ns=<authority>", "/api/partitions?ns=<authority>", ], source: "https://tangled.org/zat.dev/stream/tree/main/strata", }); default: return json({ error: "not found" }, 404); } },} satisfies ExportedHandler<Env>;