diff --git a/strata/README.md b/strata/README.md index caa0f98..a67ba67 100644 --- a/strata/README.md +++ b/strata/README.md @@ -8,7 +8,7 @@ archive, summary, or profile API requests. ## service -The Cloudflare API, D1 schema, R2 archive writer, and drain scripts live in +The Cloudflare API, R2 snapshot reader, archive writer, and drain scripts live in [`service/`](service/). The service was imported with its Git history from the former local `zat.dev/strata` repository. This directory owns both the map and its backing service; they deploy independently. @@ -20,10 +20,11 @@ npm run check npm run deploy ``` -The Worker remains `strata` at `strata.zat.dev`, with the existing D1 database, -R2 buckets, and secrets. Do not run schema migrations for a source move. +The Worker remains `strata` at `strata.zat.dev`, with its existing R2 buckets and +secrets. It no longer binds D1. HeavyPad reconciles footer counts in SQLite and +publishes only changed serving objects; see [snapshot architecture](service/docs/snapshots.md). The hourly ingestion flow stays in `my-prefect-server/flows/strata.py` and -continues to use the same API. The deployed drain stays at `/opt/strata` on +publishes snapshots instead of writing D1. The deployed drain stays at `/opt/strata` on the Stream host: its binaries, progress file, and work directory are runtime state, not repository files. Deploy script updates there without replacing that state. See [service architecture](service/docs/how-it-works.md). diff --git a/strata/service/README.md b/strata/service/README.md index af85d44..4b2eeae 100644 --- a/strata/service/README.md +++ b/strata/service/README.md @@ -1,10 +1,12 @@ # strata the shape of the [stream.waow.tech](https://stream.waow.tech) archive, read -from outside: per-segment, per-collection event counts, stored in D1 and -served by a Worker. the archive process is not involved. +from outside: per-segment, per-collection event counts reconciled in SQLite on +HeavyPad and published as R2 snapshots, served by a Worker. Public reads do not +depend on HeavyPad availability. The archive process is not involved. -see [docs/how-it-works.md](docs/how-it-works.md). +See [the current architecture and recovery guide](docs/snapshots.md). +[The earlier architecture](docs/how-it-works.md) describes the retired D1 ingestion path. This is Stream's Strata service. Run commands from this directory: `npm ci`, `npm run check`, and `npm run deploy`. See [the parent README](../README.md) diff --git a/strata/service/docs/how-it-works.md b/strata/service/docs/how-it-works.md index e4da5be..3c99bfb 100644 --- a/strata/service/docs/how-it-works.md +++ b/strata/service/docs/how-it-works.md @@ -1,5 +1,9 @@ # how it works +Historical design notes. Summary storage moved from D1 to HeavyPad SQLite and +R2 serving snapshots on 2026-10-11; see [the current architecture](snapshots.md). +The footer format and witnessed-time caveat below still apply. + ## the read every sealed `.jss` segment carries a footer whose layout is fixed by diff --git a/strata/service/docs/snapshots.md b/strata/service/docs/snapshots.md new file mode 100644 index 0000000..0af8580 --- /dev/null +++ b/strata/service/docs/snapshots.md @@ -0,0 +1,79 @@ +# HeavyPad snapshots + +Strata's public map is a deploy-time snapshot. The original interactive heatmap, +local explorer and compatibility API still support live queries of the most recently +published summaries. Neither experience needs a database write per archive collection. + +## Consumers checked for the October 2026 migration + +A source search across the local Tangled repositories and the available personal +GitHub checkout found these consumers. This is a source inventory, not a claim that +public API access logs establish every external consumer. + +| Consumer | Required data | +| --- | --- | +| `scripts/deploy-strata-preview` | progress, collection totals, segment buckets (`heat?step=100&top=1`) | +| `strata/refresh-identities.py` | collection names and namespaces | +| `strata/server.py` | overview plus per-namespace heatmaps at step 100 | +| `strata/service/src/ui.js` | progress, namespace rankings, arbitrary supported step/top heatmaps, namespace expansion, public Parquet partition listing | +| Stream Caddy `/strata/api/*` compatibility proxy | existing API contracts | +| Evergreen service inventory | original landing page | +| Prefect reconciliation | local segment checksums and footer counts | +| Archive drain | authenticated R2 uploads; retained independently of summary ingestion | + +All existing read endpoints remain supported, including arbitrary valid heatmap +bucket sizes and the top-60 ranking limit. Public/private Parquet policy is unchanged. +The old `/ingest` write endpoint is retired after cutover. No event bodies, profiles, +block data or additional archive attributes are collected by the summary job. + +## Data path + +The existing hourly Prefect flow runs on HeavyPad. Its SQLite checkpoint is +`/home/stoat/prefect-analytics/strata/checkpoint.sqlite`. A file lock prevents overlapping +writers. The job lists sealed segments, compares checksums locally, and reads only +new or changed collection footers. Collection rows are updated only when their counts +change; removed collections are deleted in the same transaction as segment metadata. + +The builder produces progress, checksums, collection and namespace totals, segment +metadata, global top-60 heatmap series with an `other` remainder, and per-namespace +series. Object names are SHA-256 content hashes. Unchanged objects are reused. +The initial D1 export produced 2,511 objects totaling 13.5 MB, including 2,504 namespace +objects; these are derived serving data, not additional archive gathering. + +The authenticated publisher uploads objects to the existing private R2 bucket under +`snapshots/objects/`, then replaces `snapshots/manifest`. A failed object upload leaves +the previous manifest serving. Each request reads one manifest generation. The Worker +caches immutable objects and API responses, and aggregates only the requested view. +HeavyPad is never in the public request path. API response caching remains five minutes. +The public static map's deployment cadence and assets remain unchanged. + +The builder records uploaded hashes locally and removes objects not referenced for +seven days. Identical generations skip publication and checkpoint upload entirely. +It never deletes an object referenced by the current manifest. This bounds +historical storage while retaining recent rollback data. A compressed SQLite checkpoint +is backed up to private R2 before publication. Existing D1 data is retained for rollback; +the final Worker has no D1 binding and cannot write to it. + +## Recovery and rollout + +Restore explicitly by running `strata-hourly` with `restore_checkpoint=true`. This only +installs a checkpoint when the local file is absent; it verifies SQLite integrity and +atomically renames the restored file. It clears upload bookkeeping so a recovered job +cannot assume expired R2 objects still exist. A missing checkpoint without the explicit +restore flag fails instead of launching an accidental full archive crawl. + +For bootstrap, use `max_segments=0` with restoration to validate recovery/publication +without reading archive footers. Then run normal reconciliation before enabling the +hourly schedule. Runtime secrets remain existing Prefect Secret block references. + +The migration compares seeded snapshot responses against saved live D1 responses before +cutover. `scripts/verify-snapshot.mjs SNAPSHOT_DIRECTORY BASELINE_DIRECTORY`, run with +`node --import tsx`, verifies data equivalence. Worker tests compare heatmap aggregation +against real SQLite, including sparse indices and zero counts. Python tests cover +transaction rollback, unchanged count writes, removals and failed publication. + +Rollback can restore `snapshots/previous-manifest` while its objects remain retained +or redeploy the previous D1 Worker version. Keep the D1-writing hourly deployment +disabled; restoring that old ingestion +path would restore its spend. D1 and the new checkpoint diverge after cutover, so D1 is +only a dated fallback, never the current source of truth. diff --git a/strata/service/package.json b/strata/service/package.json index 4d037c9..2d73e37 100644 --- a/strata/service/package.json +++ b/strata/service/package.json @@ -6,9 +6,9 @@ "typecheck": "tsc --noEmit", "test": "node --test --import tsx tests/*.test.ts", "deploy": "wrangler deploy", - "migrate": "wrangler d1 execute strata --remote --file schema.sql", "lint": "oxlint src tests", - "check": "tsc --noEmit && oxlint src tests && node --test --import tsx tests/*.test.ts" + "check": "tsc --noEmit && oxlint src tests && node --test --import tsx tests/*.test.ts", + "types": "wrangler types src/env.d.ts --include-runtime false --env-interface Bindings" }, "devDependencies": { "@cloudflare/workers-types": "5.20260829.1", diff --git a/strata/service/scripts/verify-snapshot.mjs b/strata/service/scripts/verify-snapshot.mjs new file mode 100644 index 0000000..2577964 --- /dev/null +++ b/strata/service/scripts/verify-snapshot.mjs @@ -0,0 +1,56 @@ +import { renderHeat, heatSchema, segmentsSchema } from "../src/snapshot.ts"; +import { safeParse } from "@atcute/lexicons/validations"; +import { readFile, readdir } from "node:fs/promises"; +import assert from "node:assert/strict"; + +const directory = process.argv[2]; +const baseline = process.argv[3]; +if (!directory || !baseline) + throw Error( + "usage: node scripts/verify-snapshot.mjs SNAPSHOT_DIRECTORY BASELINE_DIRECTORY", + ); +const manifest = JSON.parse( + await readFile(`${directory}/manifest.json`, "utf8"), +); +async function load(name) { + const ref = manifest.objects.find((r) => r.name === name); + return JSON.parse(await readFile(`${directory}/${ref.digest}`, "utf8")); +} +const paths = JSON.parse(await readFile(`${baseline}/paths.json`, "utf8")); +function canonical(value) { + if (Array.isArray(value)) + return value + .map(canonical) + .sort((a, b) => JSON.stringify(a).localeCompare(JSON.stringify(b))); + if (value && typeof value === "object") + return Object.fromEntries( + Object.entries(value) + .sort(([a], [b]) => a.localeCompare(b)) + .map(([k, v]) => [k, canonical(v)]), + ); + return value; +} +for (const [i, path] of paths.entries()) { + if (path.startsWith("/api/partitions")) continue; + const start = performance.now(); + const url = new URL("https://strata.test" + path); + let actual; + if (url.pathname === "/api/heat") { + const segments = safeParse(segmentsSchema, await load("segments")); + const heat = safeParse( + heatSchema, + await load( + url.searchParams.has("ns") + ? "ns/" + url.searchParams.get("ns") + : "heat/" + (url.searchParams.get("group") ?? "namespaces"), + ), + ); + assert.ok(segments.ok); + assert.ok(heat.ok); + actual = renderHeat(segments.value, heat.value, url); + } else actual = await load(url.pathname.slice("/api/".length)); + const expected = JSON.parse(await readFile(`${baseline}/${i}.json`, "utf8")); + assert.deepEqual(canonical(actual), canonical(expected), path); + console.log("parity", path, Math.round(performance.now() - start), "ms"); +} +console.log("snapshot parity passed"); diff --git a/strata/service/src/env.d.ts b/strata/service/src/env.d.ts new file mode 100644 index 0000000..5727abe --- /dev/null +++ b/strata/service/src/env.d.ts @@ -0,0 +1,21 @@ +/* eslint-disable */ +// Generated by Wrangler by running `wrangler types --include-runtime=false --env-interface=Bindings src/env.d.ts` (hash: 952e2b96636b49c7d5934e24e0b88a0b) +interface __BaseEnv_Bindings { + PUBLIC: R2Bucket; + PRIVATE: R2Bucket; +} +declare namespace Cloudflare { + interface GlobalProps { + mainModule: typeof import("./index"); + } + interface Env extends __BaseEnv_Bindings {} +} +interface Bindings extends __BaseEnv_Bindings {} +declare module "*.html" { + const value: string; + export default value; +} +declare module "*.js" { + const value: string; + export default value; +} \ No newline at end of file diff --git a/strata/service/src/index.ts b/strata/service/src/index.ts index d693bd0..d882179 100644 --- a/strata/service/src/index.ts +++ b/strata/service/src/index.ts @@ -1,33 +1,27 @@ -/** strata — the shape of the stream.waow.tech archive, read from outside. - * - * a prefect flow walks the archive's sealed segments, decodes each footer's - * collection index, and POSTs the result here. this worker only stores and - * aggregates; it holds no schedule and never talks to the archive itself. */ +import { safeParse, array, object, string } from "@atcute/lexicons/validations"; -import { safeParse } from "@atcute/lexicons/validations"; +import { manifestSchema, snapshotResponse } from "./snapshot"; -import { IngestBatchSchema, checkIngestBatch, namespaceOf, type SegmentRecord } from "./ingest"; 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); + 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 { - readonly DB: D1Database; - readonly PUBLIC: R2Bucket; - readonly PRIVATE: R2Bucket; +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)$/; +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). */ @@ -35,34 +29,74 @@ 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 { +async function putArchive( + request: Request, + key: string, + env: Env, +): Promise { if (!authorized(request, env)) return json({ error: "unauthorized" }, 401); - if (!ARCHIVE_KEY.test(key)) return json({ error: "key must be events/ns=/seg=.parquet, agg/seg=.parquet, or manifest/seg=.json" }, 400); + if (!ARCHIVE_KEY.test(key)) + return json( + { + error: + "key must be events/ns=/seg=.parquet, agg/seg=.parquet, or manifest/seg=.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 ( + !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" }, + 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", }); - 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); + return json( + { error: err instanceof Error ? err.message : "put failed" }, + 503, + ); } } -const MAX_BODY_BYTES = 8 << 20; -const MAX_HEAT_BUCKETS = 2000; - /** 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> = {}): Response { +type JsonValue = + | string + | number + | boolean + | null + | undefined + | readonly JsonValue[] + | { readonly [key: string]: JsonValue }; + +function json( + body: JsonValue, + status = 200, + headers: Readonly> = {}, +): Response { return new Response(JSON.stringify(body), { status, - headers: { "content-type": "application/json", "access-control-allow-origin": "*", ...headers }, + headers: { + "content-type": "application/json", + "access-control-allow-origin": "*", + ...headers, + }, }); } @@ -76,318 +110,230 @@ function timingSafeEqual(a: string, b: string): boolean { function authorized(request: Request, env: Env): boolean { const header = request.headers.get("authorization") ?? ""; - const token = header.startsWith("Bearer ") ? header.slice("Bearer ".length) : ""; + const token = header.startsWith("Bearer ") + ? header.slice("Bearer ".length) + : ""; return token.length > 0 && timingSafeEqual(token, env.STRATA_INGEST_TOKEN); } -function segmentStatements(db: D1Database, segment: SegmentRecord, ingestedAt: number): D1PreparedStatement[] { - const upsertSegment = db - .prepare( - `INSERT INTO segments (idx, name, size_bytes, checksum, event_count, block_count, min_seq, max_seq, - min_witnessed_at, max_witnessed_at, ingested_at) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11) - ON CONFLICT(idx) DO UPDATE SET name = excluded.name, size_bytes = excluded.size_bytes, - checksum = excluded.checksum, event_count = excluded.event_count, block_count = excluded.block_count, - min_seq = excluded.min_seq, max_seq = excluded.max_seq, min_witnessed_at = excluded.min_witnessed_at, - max_witnessed_at = excluded.max_witnessed_at, ingested_at = excluded.ingested_at`, - ) - .bind( - segment.idx, - segment.name, - segment.sizeBytes, - segment.checksum, - segment.eventCount, - segment.blockCount, - segment.minSeq, - segment.maxSeq, - segment.minWitnessedAt, - segment.maxWitnessedAt, - ingestedAt, - ); - const clear = db.prepare("DELETE FROM segment_collections WHERE idx = ?1").bind(segment.idx); - const upsertCollection = db.prepare("INSERT INTO segment_collections (idx, nsid, ns, count) VALUES (?1, ?2, ?3, ?4)"); - return [ - upsertSegment, - clear, - ...segment.collections.map((c) => upsertCollection.bind(segment.idx, c.nsid, namespaceOf(c.nsid), c.count)), - ]; -} - -async function ingest(request: Request, env: Env): Promise { - if (!authorized(request, env)) return json({ error: "unauthorized" }, 401); - const length = Number(request.headers.get("content-length") ?? "0"); - if (!Number.isFinite(length) || length > MAX_BODY_BYTES) return json({ error: "body too large" }, 413); - let body; - try { - body = await request.json(); - } catch { - return json({ error: "invalid json" }, 400); - } - const parsed = safeParse(IngestBatchSchema, body); - if (!parsed.ok) return json({ error: parsed.message }, 400); - const problem = checkIngestBatch(parsed.value); - if (problem !== null) return json({ error: problem }, 400); - const ingestedAt = Date.now(); - for (const segment of parsed.value.segments) { - await env.DB.batch(segmentStatements(env.DB, segment, ingestedAt)); - } - if (parsed.value.archiveSegments !== undefined) { - await env.DB.prepare( - "INSERT INTO meta (key, value, updated_at) VALUES ('archive_segments', ?1, ?2) ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at", - ) - .bind(parsed.value.archiveSegments, ingestedAt) - .run(); - } - return json({ ingested: parsed.value.segments.length }); -} - -type ProgressRow = { - readonly segments: number; - readonly max_idx: number | null; - readonly max_seq: number | null; - readonly events: number | null; - readonly updated_at: number | null; -}; - -async function progress(env: Env): Promise { - const [row, kinds, meta] = await env.DB.batch([ - env.DB.prepare( - "SELECT COUNT(*) AS segments, MAX(idx) AS max_idx, MAX(max_seq) AS max_seq, SUM(event_count) AS events, MAX(ingested_at) AS updated_at FROM segments", - ), - env.DB.prepare("SELECT COUNT(DISTINCT nsid) AS collections, COUNT(DISTINCT ns) AS namespaces FROM segment_collections"), - env.DB.prepare("SELECT value FROM meta WHERE key = 'archive_segments'"), - ]); - // SAFETY: the third statement selects exactly this one column. - const archive = meta?.results[0] as { value: number } | undefined; - // SAFETY: the first statement selects exactly ProgressRow's columns. - const p = row?.results[0] as ProgressRow | undefined; - // SAFETY: the second statement selects exactly these two counts. - const k = kinds?.results[0] as { collections: number; namespaces: number } | undefined; - return json({ - segments: p?.segments ?? 0, - maxIdx: p?.max_idx ?? null, - maxSeq: p?.max_seq ?? null, - events: p?.events ?? 0, - updatedAt: p?.updated_at ?? null, - collections: k?.collections ?? 0, - namespaces: k?.namespaces ?? 0, - archiveSegments: archive?.value ?? null, - }); -} - -type HeatRow = { - readonly bucket: number; - readonly key: string; - readonly count: number; -}; - -type BucketRow = { - readonly bucket: number; - readonly first_idx: number; - readonly last_idx: number; - readonly min_seq: number; - readonly max_seq: number; - readonly min_witnessed_at: number; - readonly max_witnessed_at: number; - readonly event_count: number; -}; - -const MAX_TOP = 60; - -interface HeatQuery { - readonly step: number; - readonly top: number; - readonly scope: { readonly kind: "namespaces" } | { readonly kind: "namespace"; readonly ns: string } | { readonly kind: "collections" }; -} - -type HeatQueryResult = { readonly ok: true; readonly query: HeatQuery } | { readonly ok: false; readonly error: string }; - -function parseHeatQuery(url: URL): HeatQueryResult { - const step = Number(url.searchParams.get("step") ?? "10"); - const top = Number(url.searchParams.get("top") ?? "12"); - if (!Number.isSafeInteger(step) || step < 1) return { ok: false, error: "step must be a positive integer" }; - if (!Number.isSafeInteger(top) || top < 1 || top > MAX_TOP) return { ok: false, error: `top must be 1..${MAX_TOP}` }; - const ns = url.searchParams.get("ns"); - const group = url.searchParams.get("group") ?? "namespaces"; - if (ns !== null) { - if (!/^[a-z0-9-]+\.[a-z0-9-]+$/i.test(ns)) return { ok: false, error: "ns must be an authority like fm.teal" }; - return { ok: true, query: { step, top, scope: { kind: "namespace", ns } } }; - } - if (group === "collections") return { ok: true, query: { step, top, scope: { kind: "collections" } } }; - if (group === "namespaces") return { ok: true, query: { step, top, scope: { kind: "namespaces" } } }; - return { ok: false, error: "group must be namespaces or collections" }; -} - -/** rows of the heat map: (bucket, key, count) where key is a namespace, or an nsid inside one namespace, or a - * top-N nsid; everything outside the top N folds into "other" so the response stays small at any archive size. */ -function heatRowsStatement(db: D1Database, q: HeatQuery): D1PreparedStatement { - switch (q.scope.kind) { - case "namespaces": - return db - .prepare( - `WITH ranked AS (SELECT ns FROM segment_collections GROUP BY ns ORDER BY SUM(count) DESC LIMIT CAST(?2 AS INTEGER)) - SELECT idx / CAST(?1 AS INTEGER) AS bucket, - CASE WHEN ns IN (SELECT ns FROM ranked) THEN ns ELSE 'other' END AS key, SUM(count) AS count - FROM segment_collections GROUP BY bucket, 2 ORDER BY bucket, 2`, - ) - .bind(q.step, q.top); - case "namespace": - return db - .prepare( - `SELECT idx / CAST(?1 AS INTEGER) AS bucket, nsid AS key, SUM(count) AS count - FROM segment_collections WHERE ns = ?2 GROUP BY bucket, nsid ORDER BY bucket, nsid`, - ) - .bind(q.step, q.scope.ns); - case "collections": - return db - .prepare( - `WITH ranked AS (SELECT nsid FROM segment_collections GROUP BY nsid ORDER BY SUM(count) DESC LIMIT CAST(?2 AS INTEGER)) - SELECT idx / CAST(?1 AS INTEGER) AS bucket, - CASE WHEN nsid IN (SELECT nsid FROM ranked) THEN nsid ELSE 'other' END AS key, SUM(count) AS count - FROM segment_collections GROUP BY bucket, 2 ORDER BY bucket, 2`, - ) - .bind(q.step, q.top); - } -} - -function heatKeysStatement(db: D1Database, q: HeatQuery): D1PreparedStatement { - switch (q.scope.kind) { - case "namespaces": - return db - .prepare( - `SELECT ns AS key, SUM(count) AS count, COUNT(DISTINCT nsid) AS members - FROM segment_collections GROUP BY ns ORDER BY count DESC LIMIT CAST(?1 AS INTEGER)`, - ) - .bind(q.top); - case "namespace": - return db - .prepare( - `SELECT nsid AS key, SUM(count) AS count, 1 AS members FROM segment_collections WHERE ns = ?1 - GROUP BY nsid ORDER BY count DESC`, - ) - .bind(q.scope.ns); - case "collections": - return db - .prepare( - `SELECT nsid AS key, SUM(count) AS count, 1 AS members FROM segment_collections - GROUP BY nsid ORDER BY count DESC LIMIT CAST(?1 AS INTEGER)`, - ) - .bind(q.top); - } -} - -type KeyRow = { - readonly key: string; - readonly count: number; - readonly members: number; -}; - -async function heat(url: URL, env: Env): Promise { - const parsed = parseHeatQuery(url); - if (!parsed.ok) return json({ error: parsed.error }, 400); - const q = parsed.query; - const progressRow = await env.DB.prepare("SELECT MAX(idx) AS max_idx FROM segments").first<{ max_idx: number | null }>(); - const maxIdx = progressRow?.max_idx ?? -1; - if (Math.floor(maxIdx / q.step) + 1 > MAX_HEAT_BUCKETS) { - return json({ error: `step ${q.step} yields more than ${MAX_HEAT_BUCKETS} buckets` }, 400); - } - const results = await env.DB.batch([ - env.DB.prepare( - `SELECT idx / CAST(?1 AS INTEGER) AS bucket, MIN(idx) AS first_idx, MAX(idx) AS last_idx, MIN(min_seq) AS min_seq, - MAX(max_seq) AS max_seq, MIN(min_witnessed_at) AS min_witnessed_at, MAX(max_witnessed_at) AS max_witnessed_at, - SUM(event_count) AS event_count - FROM segments GROUP BY bucket ORDER BY bucket`, - ).bind(q.step), - heatRowsStatement(env.DB, q), - heatKeysStatement(env.DB, q), - ]); - const [buckets, cells, keys] = results; - if (buckets === undefined || cells === undefined || keys === undefined) { - return json({ error: "batch returned fewer results than statements" }, 500); - } - // SAFETY: the first statement selects exactly BucketRow's columns. - const bucketRows = buckets.results as BucketRow[]; - // SAFETY: heatRowsStatement selects exactly HeatRow's columns in every scope. - const heatRows = cells.results as HeatRow[]; - // SAFETY: heatKeysStatement selects exactly KeyRow's columns in every scope. - const keyRows = keys.results as KeyRow[]; - return json( - { step: q.step, top: q.top, scope: q.scope, buckets: bucketRows, cells: heatRows, keys: keyRows }, - 200, - { "cache-control": "public, max-age=300" }, - ); -} - -async function collections(url: URL, env: Env): Promise { - 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); - const stmt = ns === null - ? env.DB.prepare("SELECT nsid, ns, SUM(count) AS count, COUNT(*) AS segments FROM segment_collections GROUP BY nsid ORDER BY count DESC") - : env.DB.prepare("SELECT nsid, ns, SUM(count) AS count, COUNT(*) AS segments FROM segment_collections WHERE ns = ?1 GROUP BY nsid ORDER BY count DESC").bind(ns); - const { results } = await stmt.all<{ nsid: string; ns: string; count: number; segments: number }>(); - return json({ collections: results }, 200, { "cache-control": "public, max-age=300" }); -} - 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 { 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 }); + 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" }); + return json({ ns, files: urls, bytes, truncated: listed.truncated }, 200, { + "cache-control": "public, max-age=120", + }); } -/** every summarised segment's checksum, so the flow can re-read segments compaction has rewritten. */ -async function checksums(env: Env): Promise { - const { results } = await env.DB.prepare("SELECT idx, checksum FROM segments ORDER BY idx").all<{ idx: number; checksum: string }>(); - return json({ segments: results }); -} +async function publishSnapshot( + request: Request, + env: Env, + key: string, +): Promise { + 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 }); + } -async function namespaces(env: Env): Promise { - const { results } = await env.DB.prepare( - "SELECT ns, SUM(count) AS count, COUNT(DISTINCT nsid) AS members FROM segment_collections GROUP BY ns ORDER BY count DESC", - ).all<{ ns: string; count: number; members: number }>(); - return json({ namespaces: results }, 200, { "cache-control": "public, max-age=300" }); + 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): Promise { + async fetch( + request: Request, + env: Env, + ctx: ExecutionContext, + ): Promise { const url = new URL(request.url); - if (request.method === "POST" && url.pathname === "/ingest") return ingest(request, env); - 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); + 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/progress": - return progress(env); - case "/api/heat": - return heat(url, env); - case "/api/collections": - return collections(url, env); - case "/api/namespaces": - return namespaces(env); - case "/api/checksums": - return checksums(env); 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" } }); + 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", + "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=", "/api/namespaces", "/api/collections?ns=", "/api/partitions?ns="], + endpoints: [ + "/api/progress", + "/api/heat?step=N&top=N&group=namespaces|collections", + "/api/heat?step=N&ns=", + "/api/namespaces", + "/api/collections?ns=", + "/api/partitions?ns=", + ], source: "https://tangled.org/zat.dev/stream/tree/main/strata", }); default: diff --git a/strata/service/src/snapshot.ts b/strata/service/src/snapshot.ts new file mode 100644 index 0000000..4e57897 --- /dev/null +++ b/strata/service/src/snapshot.ts @@ -0,0 +1,221 @@ +import { + array, + integer, + object, + string, + literal, + nullable, + safeParse, + constrain, + arrayLength, + type InferOutput, +} from "@atcute/lexicons/validations"; + +const keySchema = object({ + key: string(), + count: integer(), + members: integer(), +}); +export const heatSchema = object({ + keys: array(keySchema), + cells: array(constrain(array(integer()), [arrayLength(3, 3)])), +}); +export const segmentsSchema = array( + constrain(array(integer()), [arrayLength(6, 6)]), +); +export const manifestSchema = object({ + version: literal(1), + updatedAt: nullable(integer()), + objects: array(object({ name: string(), digest: string() })), +}); + +export function heatQuery(url: URL) { + const step = Number(url.searchParams.get("step") ?? "10"); + const top = Number(url.searchParams.get("top") ?? "12"); + const ns = url.searchParams.get("ns"); + const group = url.searchParams.get("group") ?? "namespaces"; + if (!Number.isSafeInteger(step) || step < 1) + throw new RangeError("step must be a positive integer"); + if (!Number.isSafeInteger(top) || top < 1 || top > 60) + throw new RangeError("top must be 1..60"); + if (ns !== null && !/^[a-z0-9-]+\.[a-z0-9-]+$/i.test(ns)) + throw new RangeError("ns must be an authority like fm.teal"); + if (ns === null && group !== "collections" && group !== "namespaces") + throw new RangeError("group must be namespaces or collections"); + return { step, top, ns, group }; +} + +export function renderHeat( + segments: InferOutput, + heat: InferOutput, + url: URL, +) { + const q = heatQuery(url); + const maxIdx = segments.at(-1)?.[0] ?? -1; + if (Math.floor(maxIdx / q.step) + 1 > 2000) + throw new RangeError(`step ${q.step} yields more than 2000 buckets`); + type Bucket = { + bucket: number; + first_idx: number; + last_idx: number; + min_seq: number; + max_seq: number; + min_witnessed_at: number; + max_witnessed_at: number; + event_count: number; + }; + const buckets = new Map(); + for (const [idx, minSeq, maxSeq, minAt, maxAt, events] of segments) { + if (idx === undefined) throw new Error("invalid segment index"); + const bucket = Math.floor(idx / q.step); + if ( + minSeq === undefined || + maxSeq === undefined || + minAt === undefined || + maxAt === undefined || + events === undefined + ) + throw new Error("invalid segment"); + const b = buckets.get(bucket); + if (b) { + b.last_idx = idx; + b.min_seq = Math.min(b.min_seq, minSeq); + b.max_seq = Math.max(b.max_seq, maxSeq); + b.min_witnessed_at = Math.min(b.min_witnessed_at, minAt); + b.max_witnessed_at = Math.max(b.max_witnessed_at, maxAt); + b.event_count += events; + } else + buckets.set(bucket, { + bucket, + first_idx: idx, + last_idx: idx, + min_seq: minSeq, + max_seq: maxSeq, + min_witnessed_at: minAt, + max_witnessed_at: maxAt, + event_count: events, + }); + } + const cells = new Map>(); + for (const [idx, rank, count] of heat.cells) { + if (idx === undefined) throw new Error("invalid segment index"); + const bucket = Math.floor(idx / q.step); + if (rank === undefined || count === undefined) + throw new Error("invalid cell"); + const key = + q.ns !== null || rank < q.top + ? (heat.keys[rank]?.key ?? "other") + : "other"; + let row = cells.get(bucket); + if (!row) { + row = new Map(); + cells.set(bucket, row); + } + row.set(key, (row.get(key) ?? 0) + count); + } + return { + step: q.step, + top: q.top, + scope: q.ns !== null ? { kind: "namespace", ns: q.ns } : { kind: q.group }, + buckets: [...buckets.values()], + cells: [...cells] + .sort(([a], [b]) => a - b) + .flatMap(([bucket, row]) => + [...row] + .sort(([a], [b]) => (a < b ? -1 : a > b ? 1 : 0)) + .map(([key, count]) => ({ bucket, key, count })), + ), + keys: q.ns !== null ? heat.keys : heat.keys.slice(0, q.top), + }; +} + +export async function snapshotResponse( + url: URL, + bucket: R2Bucket, + ctx: ExecutionContext, +): Promise { + const cache = caches.default; + async function read(key: string, ttl: number) { + const cacheKey = new Request( + `https://strata.zat.dev/_snapshot-cache/${key}`, + ); + const cached = await cache.match(cacheKey); + if (cached) return cached; + const object = await bucket.get(key); + if (!object) return null; + const response = new Response(object.body, { + headers: { etag: object.etag, "cache-control": `public, max-age=${ttl}` }, + }); + ctx.waitUntil(cache.put(cacheKey, response.clone())); + return response; + } + const pointer = await read("snapshots/manifest", 30); + if (!pointer) + return Response.json({ error: "snapshot unavailable" }, { status: 503 }); + const parsed = safeParse(manifestSchema, await pointer.json()); + if (!parsed.ok) throw new Error("invalid snapshot manifest"); + const manifest = parsed.value; + async function get(name: string) { + const digest = manifest.objects.find((o) => o.name === name)?.digest; + if (!digest || !/^[a-f0-9]{64}$/.test(digest)) return null; + return read(`snapshots/objects/${digest}`, 31536000); + } + const headers = { + "content-type": "application/json", + "access-control-allow-origin": "*", + "cache-control": "public, max-age=300", + "x-strata-generation": pointer.headers.get("etag") ?? "", + }; + try { + if (url.pathname === "/api/heat") { + const q = heatQuery(url); + const [segments, heat] = await Promise.all([ + get("segments"), + get(q.ns !== null ? `ns/${q.ns}` : `heat/${q.group}`), + ]); + if (!segments) throw new Error("snapshot segments missing"); + if (!heat && q.ns === null) throw new Error("snapshot heat missing"); + const segmentData = safeParse(segmentsSchema, await segments.json()); + const heatData = safeParse( + heatSchema, + heat ? await heat.json() : { keys: [], cells: [] }, + ); + if (!segmentData.ok || !heatData.ok) throw new Error("invalid snapshot"); + return Response.json(renderHeat(segmentData.value, heatData.value, url), { + headers, + }); + } + const name = url.pathname.slice("/api/".length); + const payload = await get(name); + if (!payload) return Response.json({ error: "not found" }, { status: 404 }); + if (name === "collections" && url.searchParams.has("ns")) { + const ns = url.searchParams.get("ns"); + if (ns === null || !/^[a-z0-9-]+\.[a-z0-9-]+$/i.test(ns)) + throw new RangeError("ns must be an authority like fm.teal"); + const schema = object({ + collections: array( + object({ + nsid: string(), + ns: string(), + count: integer(), + segments: integer(), + }), + ), + }); + const data = safeParse(schema, await payload.json()); + if (!data.ok) throw new Error("invalid collections snapshot"); + return Response.json( + { collections: data.value.collections.filter((c) => c.ns === ns) }, + { headers }, + ); + } + return new Response(payload.body, { headers }); + } catch (error) { + if (error instanceof RangeError) + return Response.json( + { error: error.message }, + { status: 400, headers: { "access-control-allow-origin": "*" } }, + ); + throw error; + } +} diff --git a/strata/service/tests/snapshot.test.ts b/strata/service/tests/snapshot.test.ts new file mode 100644 index 0000000..100bdf8 --- /dev/null +++ b/strata/service/tests/snapshot.test.ts @@ -0,0 +1,81 @@ +import assert from "node:assert/strict"; +import { test } from "node:test"; +import { DatabaseSync } from "node:sqlite"; +import { renderHeat, heatQuery } from "../src/snapshot"; + +test("snapshot aggregation matches SQLite for sparse segments, zero counts and arbitrary bucket sizes", () => { + const db = new DatabaseSync(":memory:"); + db.exec("CREATE TABLE cells(idx INTEGER,key TEXT,count INTEGER)"); + const cells = [ + [0, 0, 10], + [0, 1, 0], + [7, 0, 20], + [7, 1, 4], + [25, 1, 6], + [25, 2, 8], + ]; + const keys = [ + { key: "app.a", count: 30, members: 1 }, + { key: "app.b", count: 10, members: 1 }, + ]; + const names = ["app.a", "app.b", "app.c"]; + for (const [idx, rank, count] of cells) { + assert.notEqual(rank, undefined); + if (rank === undefined) throw Error("missing rank"); + db.prepare("INSERT INTO cells VALUES(?,?,?)").run( + idx ?? 0, + names[rank] ?? "", + count ?? 0, + ); + } + for (const step of [1, 3, 10, 100]) + for (const top of [1, 2, 60]) { + const result = renderHeat( + [ + [0, 1, 9, 100, 200, 10], + [7, 10, 33, 200, 300, 24], + [25, 34, 47, 400, 500, 14], + ], + { keys, cells }, + new URL(`https://test/api/heat?step=${step}&top=${top}`), + ); + const expected = db + .prepare( + "SELECT idx / CAST(? AS INTEGER) AS bucket,CASE WHEN key IN (SELECT key FROM cells GROUP BY key ORDER BY SUM(count) DESC LIMIT ?) THEN key ELSE 'other' END AS key,SUM(count) AS count FROM cells GROUP BY bucket,2 ORDER BY bucket,2", + ) + .all(step, Math.min(top, 2)); + assert.deepEqual( + result.cells, + expected.map((row) => ({ ...row })), + ); + assert.equal( + result.buckets.reduce((sum, b) => sum + b.event_count, 0), + 48, + ); + } + db.close(); +}); + +test("query validation preserves bounds", () => { + for (const query of [ + "step=0", + "step=1.5", + "top=61", + "group=nope", + "ns=../x", + ]) { + assert.throws( + () => heatQuery(new URL("https://test/api/heat?" + query)), + RangeError, + ); + } + assert.throws( + () => + renderHeat( + [[7000, 1, 2, 1, 2, 1]], + { keys: [], cells: [] }, + new URL("https://test/api/heat?step=1"), + ), + RangeError, + ); +}); diff --git a/strata/service/wrangler.jsonc b/strata/service/wrangler.jsonc index e22c599..e5538b7 100644 --- a/strata/service/wrangler.jsonc +++ b/strata/service/wrangler.jsonc @@ -2,20 +2,35 @@ "$schema": "node_modules/wrangler/config-schema.json", "name": "strata", "main": "src/index.ts", - "rules": [{ "type": "Text", "globs": ["**/*.html", "**/ui.js"], "fallthrough": true }], + "rules": [ + { + "type": "Text", + "globs": [ + "**/*.html", + "**/ui.js" + ], + "fallthrough": true + } + ], "compatibility_date": "2026-08-01", - "observability": { "enabled": true }, + "observability": { + "enabled": true + }, "workers_dev": false, - "routes": [{ "pattern": "strata.zat.dev", "custom_domain": true }], - "r2_buckets": [ - { "binding": "PUBLIC", "bucket_name": "strata-archive" }, - { "binding": "PRIVATE", "bucket_name": "strata-archive-private" } + "routes": [ + { + "pattern": "strata.zat.dev", + "custom_domain": true + } ], - "d1_databases": [ + "r2_buckets": [ + { + "binding": "PUBLIC", + "bucket_name": "strata-archive" + }, { - "binding": "DB", - "database_name": "strata", - "database_id": "3585ad5d-a112-4781-9e0c-aeea8ab4cb3c" + "binding": "PRIVATE", + "bucket_name": "strata-archive-private" } ] }