Self-hosted telemetry on Cloudflare D1 + password-gated admin dashboard (CG-7) (#1497)
* feat(telemetry): D1 schema + migrations for raw events and daily rollups First step of replacing PostHog with self-hosted telemetry on Cloudflare D1. Creates the codegraph-telemetry database binding and the initial migration; no worker code paths change yet (the ingest write path and the nightly rollup cron land next). Schema is raw events plus daily rollups: `events` holds one row per sanitized event with the envelope broken out into columns and event-specific props as JSON; `daily_machines`, `daily_event_counts` and `daily_dim_counts` are the nightly rollups the dashboard reads; `machine_first_seen` and `machine_days` carry the retention cohorts and are never purged. One generic dimension table covers every bar and pie, so a new breakdown is a cron change rather than a migration. The migration is commented as an audit surface, like the rest of this worker — every column, and which dashboard chart each rollup table serves. Three judgment calls worth flagging, all documented in the file: - `events` gets `(day, event)` instead of the separate `(day)` and `(event, day)` indexes. D1 bills a row write per index touched, so a third index on the hot table costs ~97k writes/day, and `(day, event)` is a covering index for plain day-range scans anyway (verified with EXPLAIN QUERY PLAN). - `daily_event_counts` and `daily_dim_counts` carry a `machines` column, and `machine_days` a `prod` flag. The "users by ..." panels and the production-user count are distinct-machine numbers, not event counts, and they are unrecoverable once raw events are purged. - No CHECK constraint on `event`: the worker's allowlist is the source of truth and the write path is fail-silent, so a rejected INSERT would lose data quietly instead of erroring loudly. Volume note in the migration footer: ~30M row writes/month against the 50M included on Workers Paid. Storage is the tighter constraint — raw events grow ~74 MB/day, so retention should start at 90 days (~6.7 GB) rather than 180, which would exceed D1's 10 GB per-database cap. * feat(telemetry): admin dashboard worker — scaffold + shared-password auth New Cloudflare Worker at telemetry-dashboard/, sibling of telemetry-worker/ and bound read-only to the same D1 database. Serves a static frontend plus a JSON API behind a shared password, on stats.getcodegraph.com. Auth is the simplest thing that is actually safe for exactly two users: one password in a secret, compared in constant time over SHA-256 digests, and an HMAC-signed cookie (HttpOnly; Secure; SameSite=Lax; Path=/) with a one-year expiry so you sign in once per browser. The cookie is a signed assertion, not a lookup key — no session store. Its payload carries a fingerprint of the password it was minted against, so rotating ADMIN_PASSWORD signs everyone out. Login attempts are capped at 5/min per IP via a ratelimit binding. Everything is deny-by-default: assets.run_worker_first routes every request through the worker before the static-asset server sees it, so the dashboard HTML, its JS, its CSS and the chart library are all behind the session check. The login page is rendered inline by the worker rather than served from public/, which leaves no "is this file public?" judgement calls in the asset directory. Unauthenticated pages 302 to /login, unauthenticated /api/* gets 401. A missing secret fails closed rather than opening the dashboard. scripts/smoke-auth.sh is the regression net — 54 assertions against a throwaway `wrangler dev` covering the gate, cookie flags and persistence, forged/flipped/ truncated cookies, open-redirect refusal, brute-force capping, and password rotation invalidating live sessions. Refs CG-11. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * chore(telemetry-dashboard): simplify the chart-library probe in the shell Refs CG-11. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * feat(telemetry): nightly rollup cron + raw-event retention purge (CG-10) Adds a scheduled() handler to the ingest worker that recomputes daily_event_counts / daily_dim_counts / daily_machines for the just-completed UTC day plus a 2-day overlap (late-arriving offline buffers), then purges raw events past the retention window. Rollup writes are idempotent upserts, so a re-run never double-counts. Also adds an ADMIN_TOKEN-guarded POST /admin/rollup?day=YYYY-MM-DD for backfill/repair, and drops the PostHog forwarding path. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * feat(telemetry): dashboard charts — SQL API over D1 + the Chart.js views (CG-12, CG-13) Replaces the scaffold page with the dashboard proper: 19 panels covering every view of the PostHog dashboard this retires, driven by one filter row. src/api.ts is the read API CG-12 specified: /api/{meta,summary,timeseries, breakdown,activation,retention}, all range-scoped, all parameterized against a closed set of dims and metrics, all shaped labels[] + datasets[] so the frontend does no arithmetic. Rollups answer everything except the activation funnel, which needs raw events and says where they start. The frontend splits into a DOM-free panel registry (public/panels.js) and the page that mounts it (public/app.js), so the render check can drive the same registry the browser rendered from. Panels fail alone, refetch dims rather than flashing, and every chart carries a table twin. Two numbers are labelled rather than rounded off: range-wide "users" per dimension is machine-days (the rollups cannot give distinct machines, and per-day counts are taken as the largest single-event count so one machine's install + index + usage is not counted three times), and recent activation and retention cohorts are marked as still-converting instead of drawn as a cliff. Both colour scales were run through the data-viz validator against the panel surface, not picked by eye; the results are recorded in public/theme.js. Verification, all against the committed fixture (12 machines over 10 days, every expected number worked out by hand from the events, not recorded from a run): scripts/smoke-api.sh 98 assertions scripts/render-check.mjs 79 assertions — real Chromium over CDP, no new deps scripts/smoke-auth.sh 54 assertions (unchanged, still green) Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * feat(telemetry): cutover runbook + the end-to-end gate that de-risks it (CG-14) The account-level steps of the PostHog cutover are the maintainer's to run, so this lands the runbook they follow and the check that has to pass first. The runbook (telemetry-worker/README.md) walks the six steps in the order that keeps them reversible: Workers Paid → migrate → deploy → watch 24h → verify the first rollup and the dashboard → only then delete POSTHOG_KEY and cancel the subscription. Step 3 records the outgoing version id because `wrangler rollback` is the escape hatch for the whole verification window, and that window is precisely why the PostHog key is deleted last rather than first. The new gate (scripts/smoke-cutover.sh, `npm run smoke:cutover`) covers the one seam nothing else did. Both workers declare the same D1 database_id, so pointing them at a single --persist-to directory runs the real chain: a client batch → the ingest worker → D1 → the nightly rollup → the dashboard API reading the numbers back. Every other suite stops at one link — smoke-ingest at the events table, smoke-rollup at hand-checked SQL, smoke-api at a hand-written fixture that the cron never touched. That left the dimension names the rollup WRITES versus the ones the dashboard READS agreeing by convention across two branches, where a mismatch is silent: no error, no failed request, just a panel reading zero forever. 61 assertions, all 13 dimensions, and three deliberate traps — a ci machine that is active but not a production user, usage_rollup counts that must be summed rather than tallied, and an uninstall's `targets` that must not leak into the install-scoped breakdown. Writing it caught that the activation funnel's denominator is first-seen machines, not install events (deliberate — a reinstall must not re-enter the funnel), so the suite now pins that distinction rather than assuming it. Also rewords the last PostHog reference in dashboard code: a comment justifying the 14-day retention curve by pointing at a dashboard step 6 deletes. The reasoning now stands on its own. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * docs(telemetry): tell the truth about where events are stored (CG-15) The telemetry docs are a privacy contract, and they still described a managed analytics store that no longer receives anything. Replace that with what actually happens now — events land in our own D1 database on Cloudflare, the endpoint makes no outbound requests, raw events are purged after 90 days and only anonymous daily rollups outlive them. This strengthens the guarantee rather than restating it: there is no second party to share with. - TELEMETRY.md: new "Where it is stored" section; the never-collected IP bullet no longer leans on a vendor-side setting to hold. - docs/design/telemetry.md: ingest section rewritten around D1 + the nightly rollup/retention cron; volume math redone on Workers Paid and the D1 quota (storage, not writes, is what sets the 90-day window); new section documenting the dashboard worker and cross-linking it. - Fixed three drifts from the worker allowlist the sweep surfaced: schema_version was still 1, client_name/client_version was still marked "plumbing to add" though session.ts passes it today, and the legacy sqlite_backend field the worker still accepts was undocumented. - telemetry-worker/README.md: step 6 claimed a repo-wide grep came back clean, which this runbook itself falsifies. Added step 7 — deleting the runbook is what makes that grep true, and is the completion check. - smoke-cutover.sh: the vendor guarantee is now asserted by class (no analytics-ingest endpoint referenced) rather than by one vendor's name, so it keeps working once the name is gone. Verified it still catches a planted forwarding URL. 61/61 pass. Retention is documented as 90 days, not the 180 in the task notes: 180 days of raw events exceeds D1's 10 GB per-database cap, and the code purges at 90. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * chore: untrack local Kommandr issue DB and ignore its sqlite artifacts Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Colby Mchenry
co-authored by
Claude Opus 5
Claude Opus 5
parent
f6ac7b36e6
commit
49c11fc2e0
@@ -0,0 +1,397 @@
|
||||
/**
|
||||
* codegraph telemetry — nightly rollup + raw-event retention purge.
|
||||
*
|
||||
* Public for the same reason the ingest path is: this is every read and every write
|
||||
* we make over the stored events, including the one that deletes them.
|
||||
*
|
||||
* Two jobs, both driven by the cron trigger in wrangler.jsonc (00:30 UTC daily):
|
||||
*
|
||||
* 1. ROLL UP the just-completed UTC day into `daily_machines`, `daily_event_counts`
|
||||
* and `daily_dim_counts` — plus the two days before it, because clients buffer
|
||||
* offline and ship completed-day rollups late, so a day keeps growing after it
|
||||
* ends. Every write is an upsert that OVERWRITES the recomputed value rather than
|
||||
* adding to it, so re-running a day is a no-op and never double-counts.
|
||||
*
|
||||
* 2. PURGE raw `events` past the retention window, in bounded batches. Rollups are
|
||||
* kept forever, so only ad-hoc drill-down has a horizon; `machine_days` and
|
||||
* `machine_first_seen` are never purged, because retention cohorts need the full
|
||||
* history and they are two orders of magnitude smaller than the raw rows.
|
||||
*
|
||||
* `POST /admin/rollup` re-runs a day (or a short range) on demand for backfill and
|
||||
* repair, guarded by the ADMIN_TOKEN secret. Like everything else here it makes no
|
||||
* outbound requests — the only thing this worker talks to is its own D1 database.
|
||||
*/
|
||||
|
||||
/** Raw-event retention when RETENTION_DAYS is unset or nonsense. Storage-bound — see README. */
|
||||
export const DEFAULT_RETENTION_DAYS = 90;
|
||||
/** The just-completed day, plus the two before it (late offline buffers). */
|
||||
export const ROLLUP_LOOKBACK_DAYS = 3;
|
||||
/** Widest range one manual /admin/rollup call will attempt. */
|
||||
export const MAX_MANUAL_DAYS = 31;
|
||||
|
||||
/** Rows per purge DELETE — bounded so one statement stays well inside D1's limits. */
|
||||
const PURGE_BATCH_ROWS = 5_000;
|
||||
/** Ceiling on one night's deletions (≈1.5 days of ingest at current volume). */
|
||||
const PURGE_MAX_BATCHES = 60;
|
||||
|
||||
const DAY_MS = 86_400_000;
|
||||
|
||||
/** UTC YYYY-MM-DD — the key every event, rollup and chart is bucketed on. */
|
||||
export function utcDay(atMs: number): string {
|
||||
return new Date(atMs).toISOString().slice(0, 10);
|
||||
}
|
||||
|
||||
/** Rejects both the wrong shape and impossible dates (`2026-02-31` round-trips as `2026-03-03`). */
|
||||
export function isValidDay(day: string): boolean {
|
||||
if (!/^\d{4}-\d{2}-\d{2}$/.test(day)) return false;
|
||||
const t = Date.parse(`${day}T00:00:00Z`);
|
||||
return Number.isFinite(t) && utcDay(t) === day;
|
||||
}
|
||||
|
||||
/** Configured retention, clamped to something sane; falls back to the default. */
|
||||
export function retentionDays(env: Env): number {
|
||||
const raw = Number(env.RETENTION_DAYS);
|
||||
return Number.isInteger(raw) && raw >= 1 && raw <= 3650 ? raw : DEFAULT_RETENTION_DAYS;
|
||||
}
|
||||
|
||||
/** Oldest day kept: everything strictly before this is purged. */
|
||||
export function retentionCutoff(atMs: number, keepDays: number): string {
|
||||
return utcDay(atMs - keepDays * DAY_MS);
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// The rollup statements
|
||||
// ---------------------------------------------------------------------------
|
||||
// One `INSERT … SELECT … ON CONFLICT DO UPDATE` per table or dimension: the whole
|
||||
// aggregation happens inside D1, so a day rolls up in one round trip and no event
|
||||
// row ever crosses the wire. Each takes exactly one bound parameter — the day.
|
||||
//
|
||||
// Adding a breakdown is a line in ROLLUP_STATEMENTS, never a migration — that is
|
||||
// what the generic (dim, value) shape of daily_dim_counts buys.
|
||||
|
||||
/**
|
||||
* A group's event volume. For install/index/uninstall one row is one event, but a
|
||||
* usage_rollup row is a counter the client pre-aggregated (one per machine × day ×
|
||||
* tool), so its `count` prop is what has to be summed — counting rows there would
|
||||
* silently report "machines that used the tool" and undercount by an order of magnitude.
|
||||
*/
|
||||
const COUNT = `CASE WHEN e.event = 'usage_rollup'
|
||||
THEN sum(coalesce(json_extract(e.props, '$.count'), 0))
|
||||
ELSE count(*) END`;
|
||||
|
||||
const DIM_CONFLICT = `ON CONFLICT (day, event, dim, value) DO UPDATE
|
||||
SET count = excluded.count, machines = excluded.machines`;
|
||||
|
||||
const prop = (name: string): string => `json_extract(e.props, '$.${name}')`;
|
||||
const quoted = (values: readonly string[]): string => values.map((v) => `'${v}'`).join(', ');
|
||||
const onlyEvents = (...events: readonly string[]): string => ` AND e.event IN (${quoted(events)})`;
|
||||
|
||||
/** One dimension whose value is a scalar column or a scalar prop. */
|
||||
function dimStatement(dim: string, value: string, where = ''): string {
|
||||
return `INSERT INTO daily_dim_counts (day, event, dim, value, count, machines)
|
||||
SELECT e.day, e.event, '${dim}', CAST(${value} AS TEXT), ${COUNT}, count(DISTINCT e.machine_id)
|
||||
FROM events e
|
||||
WHERE e.day = ? AND ${value} IS NOT NULL AND ${value} <> ''${where}
|
||||
GROUP BY e.day, e.event, ${value}
|
||||
${DIM_CONFLICT}`;
|
||||
}
|
||||
|
||||
/**
|
||||
* One dimension unnested from a JSON array prop — one row per element, so an index
|
||||
* of a TypeScript+Go repo counts once under each language. `json_each` over a path
|
||||
* the props do not have yields no rows, which is exactly the wanted behaviour for
|
||||
* events that omit the array.
|
||||
*/
|
||||
function arrayDimStatement(dim: string, path: string, events: readonly string[]): string {
|
||||
return `INSERT INTO daily_dim_counts (day, event, dim, value, count, machines)
|
||||
SELECT e.day, e.event, '${dim}', CAST(j.value AS TEXT), count(*), count(DISTINCT e.machine_id)
|
||||
FROM events e, json_each(e.props, '${path}') j
|
||||
WHERE e.day = ? AND e.event IN (${quoted(events)}) AND j.value <> ''
|
||||
GROUP BY e.day, e.event, j.value
|
||||
${DIM_CONFLICT}`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Rebuilt from `machine_days`, not from `events`: that table is never purged, so this
|
||||
* number stays right for days whose raw rows are long gone. `prod` is already the
|
||||
* per-machine-day maximum the ingest path maintains (0 only if every event that
|
||||
* machine sent that day carried ci = 1).
|
||||
*
|
||||
* So this is also the one rollup that can still be rebuilt for a day whose raw events
|
||||
* are long gone.
|
||||
*/
|
||||
const DAILY_MACHINES = `INSERT INTO daily_machines (day, machines, prod_machines)
|
||||
SELECT day, count(*), coalesce(sum(prod), 0) FROM machine_days WHERE day = ? GROUP BY day
|
||||
ON CONFLICT (day) DO UPDATE
|
||||
SET machines = excluded.machines, prod_machines = excluded.prod_machines`;
|
||||
|
||||
const ROLLUP_STATEMENTS: readonly string[] = [
|
||||
DAILY_MACHINES,
|
||||
|
||||
`INSERT INTO daily_event_counts (day, event, count, machines)
|
||||
SELECT e.day, e.event, ${COUNT}, count(DISTINCT e.machine_id)
|
||||
FROM events e
|
||||
WHERE e.day = ?
|
||||
GROUP BY e.day, e.event
|
||||
ON CONFLICT (day, event) DO UPDATE
|
||||
SET count = excluded.count, machines = excluded.machines`,
|
||||
|
||||
// Envelope dimensions — every event type carries them.
|
||||
dimStatement('os', 'e.os'),
|
||||
dimStatement('arch', 'e.arch'),
|
||||
dimStatement('codegraph_version', 'e.codegraph_version'),
|
||||
dimStatement('node_major', 'e.node_major'),
|
||||
|
||||
// Event-specific scalar props.
|
||||
dimStatement('file_count_bucket', prop('file_count_bucket'), onlyEvents('index')),
|
||||
dimStatement('duration_bucket', prop('duration_bucket'), onlyEvents('index')),
|
||||
dimStatement('scope', prop('scope'), onlyEvents('install')),
|
||||
// `kind` is fresh/upgrade/reinstall on install and mcp_tool/cli_command on
|
||||
// usage_rollup; `event` is part of the primary key, so both live here without colliding.
|
||||
dimStatement('kind', prop('kind'), onlyEvents('install', 'usage_rollup')),
|
||||
dimStatement('name', prop('name'), onlyEvents('usage_rollup')),
|
||||
dimStatement('client_name', prop('client_name'), onlyEvents('usage_rollup')),
|
||||
|
||||
// Array props.
|
||||
arrayDimStatement('language', '$.languages', ['index']),
|
||||
arrayDimStatement('target', '$.targets', ['install', 'uninstall']),
|
||||
|
||||
// Errors per tool/command. Not in the migration's documented dim list because dims
|
||||
// are a cron concern rather than a schema one, but rolled up because it is the one
|
||||
// usage number that is gone for good after the purge. Only groups with at least one
|
||||
// error are stored, so `count` is errors and `machines` is the machines that saw one
|
||||
// — NOT the machines that ran the tool (that is the `name` dim).
|
||||
`INSERT INTO daily_dim_counts (day, event, dim, value, count, machines)
|
||||
SELECT e.day, e.event, 'name_error', CAST(${prop('name')} AS TEXT),
|
||||
sum(${prop('error_count')}), count(DISTINCT e.machine_id)
|
||||
FROM events e
|
||||
WHERE e.day = ? AND e.event = 'usage_rollup'
|
||||
AND ${prop('name')} IS NOT NULL AND coalesce(${prop('error_count')}, 0) > 0
|
||||
GROUP BY e.day, e.event, ${prop('name')}
|
||||
${DIM_CONFLICT}`,
|
||||
];
|
||||
|
||||
/** Rollup tables derived from raw `events` — the ones `reset` wipes before recomputing. */
|
||||
const EVENT_DERIVED_TABLES = ['daily_event_counts', 'daily_dim_counts'] as const;
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Running it
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
export interface DayResult {
|
||||
day: string;
|
||||
/** Rollup rows written for the day. */
|
||||
rows: number;
|
||||
/** Day is past the retention window — a `reset` on it is ignored (see below). */
|
||||
pastRetention: boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
* Recompute every rollup for one UTC day. One D1 `batch()` = one implicit
|
||||
* transaction, so a day is either fully recomputed or not touched at all.
|
||||
*
|
||||
* Plain (upsert-only) runs are safe on any day: a day whose raw events are already
|
||||
* purged selects nothing, so nothing is written and the rollups it earned while the
|
||||
* events were still around survive untouched. That is what keeps rollups permanent.
|
||||
*
|
||||
* `reset` drops the day's event-derived rollup rows first instead of upserting over
|
||||
* them — repair for when the dimension list itself changes and a value that no longer
|
||||
* exists would otherwise linger. It is IGNORED past the retention window, where it
|
||||
* would delete rows and then find no events to rebuild them from: silently blanking a
|
||||
* real day is the one irreversible thing this file could do.
|
||||
*/
|
||||
export async function rollupDay(
|
||||
env: Env,
|
||||
day: string,
|
||||
opts: { cutoff: string; reset?: boolean },
|
||||
): Promise<DayResult> {
|
||||
const pastRetention = day < opts.cutoff;
|
||||
const statements: D1PreparedStatement[] = [];
|
||||
|
||||
if (opts.reset && !pastRetention) {
|
||||
for (const table of EVENT_DERIVED_TABLES) {
|
||||
statements.push(env.DB.prepare(`DELETE FROM ${table} WHERE day = ?`).bind(day));
|
||||
}
|
||||
}
|
||||
for (const sql of ROLLUP_STATEMENTS) {
|
||||
statements.push(env.DB.prepare(sql).bind(day));
|
||||
}
|
||||
|
||||
const results = await env.DB.batch(statements);
|
||||
const rows = results.reduce((total, r) => total + (r.meta?.changes ?? 0), 0);
|
||||
return { day, rows, pastRetention };
|
||||
}
|
||||
|
||||
export interface PurgeResult {
|
||||
/** Everything strictly before this day was deleted. */
|
||||
cutoff: string;
|
||||
deleted: number;
|
||||
batches: number;
|
||||
/** Hit the per-run batch ceiling — more rows are still due, next run takes them. */
|
||||
capped: boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
* Delete raw events older than the window, oldest first, in bounded batches.
|
||||
* `id` is a rowid alias and the purge only ever removes the oldest rows, so the
|
||||
* keyset subquery stays a cheap index range scan on (day, event).
|
||||
*/
|
||||
export async function purgeOldEvents(env: Env, cutoff: string): Promise<PurgeResult> {
|
||||
const del = env.DB.prepare(
|
||||
`DELETE FROM events WHERE id IN (SELECT id FROM events WHERE day < ? LIMIT ${PURGE_BATCH_ROWS})`,
|
||||
);
|
||||
let deleted = 0;
|
||||
for (let batch = 1; batch <= PURGE_MAX_BATCHES; batch++) {
|
||||
const { meta } = await del.bind(cutoff).run();
|
||||
const removed = meta?.changes ?? 0;
|
||||
deleted += removed;
|
||||
if (removed < PURGE_BATCH_ROWS) return { cutoff, deleted, batches: batch, capped: false };
|
||||
}
|
||||
return { cutoff, deleted, batches: PURGE_MAX_BATCHES, capped: true };
|
||||
}
|
||||
|
||||
/**
|
||||
* The cron body: roll up the completed day and the two before it, then purge.
|
||||
*
|
||||
* Logs one line of counts — never a day's contents, never a machine id. Throws if
|
||||
* anything failed so the invocation is marked failed (and retried) rather than
|
||||
* quietly skipping a day; every write here is idempotent, so a retry is safe.
|
||||
*/
|
||||
export async function runNightly(env: Env, atMs: number): Promise<void> {
|
||||
const started = Date.now();
|
||||
const keepDays = retentionDays(env);
|
||||
const cutoff = retentionCutoff(atMs, keepDays);
|
||||
|
||||
const rolled: string[] = [];
|
||||
const failed: string[] = [];
|
||||
let rows = 0;
|
||||
for (let back = 1; back <= ROLLUP_LOOKBACK_DAYS; back++) {
|
||||
const day = utcDay(atMs - back * DAY_MS);
|
||||
try {
|
||||
rows += (await rollupDay(env, day, { cutoff })).rows;
|
||||
rolled.push(day);
|
||||
} catch (err) {
|
||||
failed.push(day);
|
||||
console.error(JSON.stringify({ msg: 'rollup day failed', day, err: String(err) }));
|
||||
}
|
||||
}
|
||||
|
||||
let purge: PurgeResult | null = null;
|
||||
try {
|
||||
purge = await purgeOldEvents(env, cutoff);
|
||||
} catch (err) {
|
||||
console.error(JSON.stringify({ msg: 'purge failed', cutoff, err: String(err) }));
|
||||
}
|
||||
|
||||
console.log(
|
||||
JSON.stringify({
|
||||
msg: 'nightly rollup',
|
||||
days: rolled,
|
||||
rows,
|
||||
failed: failed.length,
|
||||
retention_days: keepDays,
|
||||
purged_before: cutoff,
|
||||
purged: purge?.deleted ?? null,
|
||||
purge_batches: purge?.batches ?? null,
|
||||
purge_capped: purge?.capped ?? null,
|
||||
ms: Date.now() - started,
|
||||
}),
|
||||
);
|
||||
|
||||
if (failed.length > 0 || purge === null) {
|
||||
throw new Error(`nightly rollup incomplete: ${failed.length} day(s) failed, purge ${purge ? 'ok' : 'failed'}`);
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// POST /admin/rollup — manual backfill / repair
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
const json = (body: unknown, status = 200): Response =>
|
||||
new Response(JSON.stringify(body), {
|
||||
status,
|
||||
headers: { 'content-type': 'application/json; charset=utf-8' },
|
||||
});
|
||||
|
||||
/** Constant-time over digests, so neither the length nor a prefix of the token leaks. */
|
||||
async function tokenMatches(provided: string, expected: string): Promise<boolean> {
|
||||
const encoder = new TextEncoder();
|
||||
const [a, b] = await Promise.all([
|
||||
crypto.subtle.digest('SHA-256', encoder.encode(provided)),
|
||||
crypto.subtle.digest('SHA-256', encoder.encode(expected)),
|
||||
]);
|
||||
return crypto.subtle.timingSafeEqual(a, b);
|
||||
}
|
||||
|
||||
/**
|
||||
* `POST /admin/rollup?day=YYYY-MM-DD[&days=N][&reset=1]`, header `x-admin-token`.
|
||||
*
|
||||
* Re-runs the rollup for `day` (default: yesterday), or for the `N` days ending on it.
|
||||
* Exists so a backfill or a repair never needs a redeploy. It only ever recomputes
|
||||
* aggregates from stored rows — there is no path here that deletes raw events; the
|
||||
* purge runs on the cron and nowhere else.
|
||||
*/
|
||||
export async function handleAdminRollup(request: Request, env: Env, url: URL): Promise<Response> {
|
||||
// No secret configured ⇒ no admin surface at all, and nothing that hints there is one.
|
||||
const expected = env.ADMIN_TOKEN;
|
||||
if (typeof expected !== 'string' || expected.length === 0) {
|
||||
return new Response('not found\n', { status: 404 });
|
||||
}
|
||||
if (request.method !== 'POST') {
|
||||
return new Response('method not allowed\n', { status: 405, headers: { allow: 'POST' } });
|
||||
}
|
||||
|
||||
if (!(await tokenMatches(request.headers.get('x-admin-token') ?? '', expected))) {
|
||||
// Cap how fast the token can be guessed at. Only failures spend the budget, so a
|
||||
// chunked backfill loop is never throttled. Best-effort and fails open like the
|
||||
// ingest limiter — the token itself is the guard, this only slows a guesser down.
|
||||
try {
|
||||
const { success } = await env.ADMIN_RATE_LIMITER.limit({ key: 'admin' });
|
||||
if (!success) return new Response('rate limited\n', { status: 429 });
|
||||
} catch (err) {
|
||||
console.error(JSON.stringify({ msg: 'rate limiter unavailable', err: String(err) }));
|
||||
}
|
||||
return new Response('unauthorized\n', { status: 401 });
|
||||
}
|
||||
|
||||
const now = Date.now();
|
||||
const day = url.searchParams.get('day') ?? utcDay(now - DAY_MS);
|
||||
if (!isValidDay(day)) return json({ error: 'day must be YYYY-MM-DD' }, 400);
|
||||
|
||||
const requested = url.searchParams.get('days');
|
||||
const span = requested === null ? 1 : Number(requested);
|
||||
if (!Number.isInteger(span) || span < 1 || span > MAX_MANUAL_DAYS) {
|
||||
return json({ error: `days must be an integer between 1 and ${MAX_MANUAL_DAYS}` }, 400);
|
||||
}
|
||||
|
||||
const reset = url.searchParams.get('reset') === '1';
|
||||
const cutoff = retentionCutoff(now, retentionDays(env));
|
||||
const endMs = Date.parse(`${day}T00:00:00Z`);
|
||||
|
||||
const days: DayResult[] = [];
|
||||
try {
|
||||
for (let back = span - 1; back >= 0; back--) {
|
||||
days.push(await rollupDay(env, utcDay(endMs - back * DAY_MS), { cutoff, reset }));
|
||||
}
|
||||
} catch (err) {
|
||||
console.error(JSON.stringify({ msg: 'manual rollup failed', through: day, err: String(err) }));
|
||||
return json({ error: 'rollup failed', through: day, completed: days }, 500);
|
||||
}
|
||||
|
||||
const rows = days.reduce((total, d) => total + d.rows, 0);
|
||||
// A day past the window kept its rollups but ignored the reset — say so rather than
|
||||
// reporting a repair that did not happen.
|
||||
const resetIgnored = reset ? days.filter((d) => d.pastRetention).map((d) => d.day) : [];
|
||||
console.log(
|
||||
JSON.stringify({
|
||||
msg: 'manual rollup',
|
||||
through: day,
|
||||
days: span,
|
||||
reset,
|
||||
reset_ignored: resetIgnored.length,
|
||||
rows,
|
||||
ms: Date.now() - now,
|
||||
}),
|
||||
);
|
||||
return json({ ok: true, through: day, retention_cutoff: cutoff, rows, reset_ignored: resetIgnored, days });
|
||||
}
|
||||
Reference in New Issue
Block a user