diff --git a/src/app/api/cron/snapshot-hl-cohort/route.ts b/src/app/api/cron/snapshot-hl-cohort/route.ts new file mode 100644 index 00000000..5812df41 --- /dev/null +++ b/src/app/api/cron/snapshot-hl-cohort/route.ts @@ -0,0 +1,111 @@ +import { timingSafeEqual } from "node:crypto"; +import { NextResponse, type NextRequest } from "next/server"; +import { + fetchHlCohortFresh, + fetchHlHip3CohortFresh, +} from "@/lib/benches/hyperliquid/builder-stats"; +import { + cohortSnapshotConfigured, + writeCohortSnapshot, +} from "@/lib/cohort-snapshot"; + +export const runtime = "nodejs"; +export const dynamic = "force-dynamic"; + +/** + * Vercel cron: refresh the /hyperliquid hub's two cohort snapshots in + * Upstash. Single endpoint that updates both keys (hl-frontends, hl-hip3) + * so a single cron entry covers the whole hub. + * + * Runs every minute. Bypasses the snapshot-first readers (which would + * loop back to their own blob); calls the *Fresh helpers directly so the + * write reflects live Prom state. Token-gated by CRON_SECRET. When the + * Upstash creds are unset the route 200s with `{configured: false}` so + * an unprovisioned environment doesn't break the cron schedule. + * + * Per-cohort errors are isolated: a Prom miss on HIP-3 doesn't block a + * fresh frontends write, and vice versa. The response body lists the + * outcome of each key so the Vercel cron log shows partial recoveries. + */ + +function isAuthorized(req: NextRequest): boolean { + const secret = (process.env.CRON_SECRET ?? "").trim(); + const header = (req.headers.get("authorization") ?? "").trim(); + if (!secret) { + return process.env.NODE_ENV !== "production"; + } + const expected = Buffer.from(`Bearer ${secret}`); + const provided = Buffer.from(header); + if (provided.length !== expected.length) return false; + return timingSafeEqual(provided, expected); +} + +type KeyOutcome = + | { key: string; ok: true; asOf: number; rowCount: number } + | { key: string; ok: false; error: string }; + +async function refresh( + key: string, + fetcher: () => Promise, +): Promise { + let result: T | null; + try { + result = await fetcher(); + } catch (err) { + return { + key, + ok: false, + error: `fetch: ${err instanceof Error ? err.message : String(err)}`, + }; + } + if (!result) { + return { + key, + ok: false, + error: "fetch returned null (prom unreachable or empty)", + }; + } + try { + await writeCohortSnapshot(key, result); + } catch (err) { + return { + key, + ok: false, + error: `write: ${err instanceof Error ? err.message : String(err)}`, + }; + } + return { key, ok: true, asOf: result.asOf, rowCount: result.rows.length }; +} + +export async function GET(req: NextRequest) { + if (!isAuthorized(req)) { + return NextResponse.json({ error: "unauthorized" }, { status: 401 }); + } + if (!cohortSnapshotConfigured()) { + return NextResponse.json( + { + ok: true, + configured: false, + message: + "cohort snapshot store not configured (KV_REST_API_URL / UPSTASH_REDIS_REST_URL absent)", + }, + { status: 200 }, + ); + } + + const startedAt = Date.now(); + const [frontends, hip3] = await Promise.all([ + refresh("hl-frontends", fetchHlCohortFresh), + refresh("hl-hip3", fetchHlHip3CohortFresh), + ]); + + const okCount = (frontends.ok ? 1 : 0) + (hip3.ok ? 1 : 0); + return NextResponse.json( + { + ok: okCount > 0, + results: [frontends, hip3], + durationMs: Date.now() - startedAt, + }, + { status: okCount === 0 ? 502 : 200 }, + ); +} diff --git a/src/app/api/cron/snapshot-perp-cohort/route.ts b/src/app/api/cron/snapshot-perp-cohort/route.ts new file mode 100644 index 00000000..290ace1f --- /dev/null +++ b/src/app/api/cron/snapshot-perp-cohort/route.ts @@ -0,0 +1,111 @@ +import { timingSafeEqual } from "node:crypto"; +import { NextResponse, type NextRequest } from "next/server"; +import { fetchPerpCohortFresh } from "@/lib/perp-stats"; +import { + cohortSnapshotConfigured, + writeCohortSnapshot, +} from "@/lib/cohort-snapshot"; + +export const runtime = "nodejs"; +// No ISR. The cron's whole job is to refresh the cohort blob: a cached +// 200 from a previous run would silently skip the Prom call. +export const dynamic = "force-dynamic"; + +/** + * Vercel cron: refresh the /perps hub cohort snapshot in Upstash. + * + * Runs every minute (vercel.json crons block). Fetches the cohort straight + * from Prom (bypassing the snapshot-first reader in fetchPerpCohort so the + * cron never loops on its own blob) and SETs the result under + * ocb:cohort:perp-cohort:v1 with a 24 h safety-net TTL. + * + * Token-gated by CRON_SECRET (Bearer header). When Upstash creds are + * missing the route still 200s with `{configured: false}` so the cron + * schedule keeps working before the integration is provisioned. + */ + +function isAuthorized(req: NextRequest): boolean { + // Trim both sides. The vercel UI / `vercel env add` paste flow has + // historically appended a trailing newline that produced a constant + // 401 with no visible reason. + const secret = (process.env.CRON_SECRET ?? "").trim(); + const header = (req.headers.get("authorization") ?? "").trim(); + if (!secret) { + // Fail closed in prod, permissive in dev to keep local manual hits + // working without exporting a fake secret. + return process.env.NODE_ENV !== "production"; + } + const expected = Buffer.from(`Bearer ${secret}`); + const provided = Buffer.from(header); + if (provided.length !== expected.length) return false; + return timingSafeEqual(provided, expected); +} + +export async function GET(req: NextRequest) { + if (!isAuthorized(req)) { + return NextResponse.json({ error: "unauthorized" }, { status: 401 }); + } + if (!cohortSnapshotConfigured()) { + return NextResponse.json( + { + ok: true, + configured: false, + message: + "cohort snapshot store not configured (KV_REST_API_URL / UPSTASH_REDIS_REST_URL absent)", + }, + { status: 200 }, + ); + } + + const startedAt = Date.now(); + let result: Awaited>; + try { + result = await fetchPerpCohortFresh(); + } catch (err) { + return NextResponse.json( + { + ok: false, + stage: "fetch", + error: err instanceof Error ? err.message : String(err), + }, + { status: 502 }, + ); + } + + if (!result) { + // Prom unreachable or completely empty. Do NOT write null over the + // existing blob: the reader's stale-tolerance window would surface + // the null, but we'd rather the reader fall through to its own live + // path and keep the previous (still-valid-for-now) snapshot until + // either the cron or a request restores live data. + return NextResponse.json( + { + ok: false, + stage: "fetch", + error: "fetchPerpCohortFresh returned null (prom unreachable or empty)", + }, + { status: 502 }, + ); + } + + try { + await writeCohortSnapshot("perp-cohort", result); + } catch (err) { + return NextResponse.json( + { + ok: false, + stage: "write", + error: err instanceof Error ? err.message : String(err), + }, + { status: 502 }, + ); + } + + return NextResponse.json({ + ok: true, + asOf: result.asOf, + venueCount: result.venues.length, + trackedVenues: result.totals.trackedVenues, + durationMs: Date.now() - startedAt, + }); +} diff --git a/src/lib/cohort-snapshot.ts b/src/lib/cohort-snapshot.ts new file mode 100644 index 00000000..1eace41f --- /dev/null +++ b/src/lib/cohort-snapshot.ts @@ -0,0 +1,147 @@ +/** + * Cohort-level snapshot layer for the hub pages (/perps, /hyperliquid). + * + * Mirrors the per-bench `src/lib/materialize/store.ts` pattern, simplified + * for cohort aggregates: one blob per key, no pointer indirection, single + * envelope `{ data, asOf }` with a TTL safety net. + * + * Why this exists: hub pages fetch live from Prom on every ISR cycle. A + * single Prom hiccup (Railway redeploy, scrape miss, harness restart) + * stores a null cohort in the Vercel ISR cache and the user sees "..." + * everywhere until the next refresh. With this layer a Vercel cron writes + * a fresh JSON every 60s, the fetcher reads it first, and only falls + * through to Prom when the snapshot is missing or stale. + * + * Storage: Upstash REST (KV_REST_API_URL + KV_REST_API_TOKEN, falling back + * to UPSTASH_REDIS_REST_URL + UPSTASH_REDIS_REST_TOKEN for parity with the + * materialize layer). When neither pair is configured, every function is a + * silent no-op so the live Prom path keeps working in local dev. + */ + +const URL_ENV = ["KV_REST_API_URL", "UPSTASH_REDIS_REST_URL"] as const; +const TOKEN_ENV = ["KV_REST_API_TOKEN", "UPSTASH_REDIS_REST_TOKEN"] as const; + +const KEY_PREFIX = "ocb:cohort:"; +const KEY_SUFFIX = ":v1"; +/** Safety-net TTL on the blob itself. Cron writes every minute so the + * blob normally rolls over well within this window; the TTL only kicks + * in if the cron is dead. 24h matches the materialize layer's intent + * ("data is always there for a day even if the writer dies"). */ +const BLOB_TTL_SEC = 24 * 60 * 60; +/** Default staleness ceiling for the reader. 10 minutes covers ~10 cron + * misses; anything older and the reader prefers a fresh Prom hit. */ +const DEFAULT_MAX_AGE_MS = 10 * 60 * 1000; + +type Envelope = { data: T; asOf: number }; + +function creds(): { url: string; token: string } | null { + const url = URL_ENV.map((k) => process.env[k]?.trim()).find(Boolean); + const token = TOKEN_ENV.map((k) => process.env[k]?.trim()).find(Boolean); + return url && token ? { url: url.replace(/\/+$/, ""), token } : null; +} + +function isConfigured(): boolean { + return creds() !== null; +} + +function blobKey(key: string): string { + return `${KEY_PREFIX}${key}${KEY_SUFFIX}`; +} + +async function redisCommand( + cmd: (string | number)[], + timeoutMs = 4_000, +): Promise { + const c = creds(); + if (!c) throw new Error("cohort-snapshot: no upstash creds"); + const res = await fetch(c.url, { + method: "POST", + headers: { + Authorization: `Bearer ${c.token}`, + "Content-Type": "application/json", + }, + body: JSON.stringify(cmd), + signal: AbortSignal.timeout(timeoutMs), + cache: "no-store", + }); + if (!res.ok) { + const detail = await res.text().catch(() => ""); + throw new Error( + `cohort-snapshot: http ${res.status}: ${detail.slice(0, 200)}`, + ); + } + const body = (await res.json()) as { result?: unknown; error?: string }; + if (body.error) throw new Error(`cohort-snapshot: ${body.error}`); + return body.result; +} + +/** + * Write a fresh cohort snapshot. Called from the Vercel cron after a + * successful live Prom fetch (or from the reader's own fall-through path + * after a brownout-recovery fetch). Silently no-ops when creds are + * absent. Failures throw so the cron logs them. + */ +export async function writeCohortSnapshot( + key: string, + data: T, +): Promise { + if (!isConfigured()) return; + const envelope: Envelope = { data, asOf: Date.now() }; + const json = JSON.stringify(envelope); + await redisCommand( + ["SET", blobKey(key), json, "EX", BLOB_TTL_SEC], + 8_000, + ); +} + +/** + * Read the most recent cohort snapshot. Returns null when: + * - creds are not configured + * - the blob key is missing + * - the payload fails JSON parse / envelope shape check + * - the envelope's asOf is older than maxAgeMs + * - the network call fails or times out + * + * Caller is expected to fall through to the live Prom path on null. + */ +export async function readCohortSnapshot( + key: string, + maxAgeMs: number = DEFAULT_MAX_AGE_MS, +): Promise<{ data: T; asOfMs: number; ageMs: number } | null> { + if (!isConfigured()) return null; + try { + const raw = await redisCommand(["GET", blobKey(key)], 3_000); + if (typeof raw !== "string" || !raw) return null; + let parsed: unknown; + try { + parsed = JSON.parse(raw); + } catch { + return null; + } + if ( + !parsed || + typeof parsed !== "object" || + !("data" in parsed) || + !("asOf" in parsed) + ) { + return null; + } + const envelope = parsed as Envelope; + const asOfMs = Number(envelope.asOf); + if (!Number.isFinite(asOfMs) || asOfMs <= 0) return null; + const ageMs = Date.now() - asOfMs; + if (ageMs > maxAgeMs) return null; + return { data: envelope.data, asOfMs, ageMs }; + } catch (err) { + console.warn( + `cohort-snapshot read failed for ${key}: ${ + err instanceof Error ? err.message : String(err) + }`, + ); + return null; + } +} + +export function cohortSnapshotConfigured(): boolean { + return isConfigured(); +} diff --git a/src/lib/hl-builder-stats.ts b/src/lib/hl-builder-stats.ts index b861e1da..4f1ef6c3 100644 --- a/src/lib/hl-builder-stats.ts +++ b/src/lib/hl-builder-stats.ts @@ -9,6 +9,10 @@ import { unstable_cache } from "next/cache"; import { Prometheus } from "@/lib/prometheus"; import { getSpecs } from "@/lib/spec"; +import { + readCohortSnapshot, + writeCohortSnapshot, +} from "@/lib/cohort-snapshot"; export type CoinShare = { coin: string; share: number }; export type PercentileBucket = { @@ -241,6 +245,13 @@ export type HlHip3Summary = { asOf: number; }; +/** Upstash keys for the hub's two cohort blobs, written by + * /api/cron/snapshot-hl-cohort. Bump the suffix here if either summary + * shape changes so a stale-shape blob can never deserialize into a + * misaligned payload. The cohort-snapshot module appends its own `:v1`. */ +const HL_FRONTENDS_KEY = "hl-frontends"; +const HL_HIP3_KEY = "hl-hip3"; + /** * Fetch a leaderboard-ready slice of every tracked HL builder, in 4 * vector queries instead of 4 × 104 scalar fan-out. Used by the @@ -253,8 +264,12 @@ export type HlHip3Summary = { * so the two agree. We re-compute it here so the leaderboard total * stays internally consistent if some builders drop out of the * filtered set. + * + * Exported so the cron route can call the uncached Prom path directly + * before parking the result in Upstash; the public reader (fetchHlCohort) + * goes through the snapshot layer + unstable_cache below. */ -export async function fetchHlCohort(): Promise { +export async function fetchHlCohortFresh(): Promise { const url = promUrl(); if (!url) return null; let prom: Prometheus; @@ -349,8 +364,13 @@ export async function fetchHlCohort(): Promise { * * Dex names come from the hyperliquid-hip3-deployers spec's providers * list; an unknown dex slug surfaces under its raw namespace. + * + * Exported so the cron route can call the uncached Prom path directly + * before parking the result in Upstash; the public reader + * (fetchHlHip3Cohort) goes through the snapshot layer + unstable_cache + * below. */ -export async function fetchHlHip3Cohort(): Promise { +export async function fetchHlHip3CohortFresh(): Promise { const url = promUrl(); if (!url) return null; let prom: Prometheus; @@ -467,6 +487,72 @@ export async function fetchHlHip3Cohort(): Promise { }; } +/** + * Snapshot-first reader for the frontends cohort. Same protocol as the + * /perps hub: Upstash blob → live Prom + writeback → null. The 60 s + * unstable_cache wrapper around this collapses concurrent requests; the + * cron writer keeps the Upstash blob fresh so the typical render never + * touches Prom. + */ +async function fetchHlCohortRaw(): Promise { + const snapshot = await readCohortSnapshot(HL_FRONTENDS_KEY); + if (snapshot) return snapshot.data; + const fresh = await fetchHlCohortFresh(); + if (fresh) { + try { + await writeCohortSnapshot(HL_FRONTENDS_KEY, fresh); + } catch (err) { + console.warn( + `hl-frontends writeback failed: ${ + err instanceof Error ? err.message : String(err) + }`, + ); + } + } + return fresh; +} + +const fetchHlCohortCached = unstable_cache( + fetchHlCohortRaw, + ["hl-frontends-cohort-v1"], + { revalidate: 60, tags: ["hl-cohort"] }, +); + +export async function fetchHlCohort(): Promise { + return fetchHlCohortCached(); +} + +/** Snapshot-first reader for the HIP-3 cohort. Same shape as + * fetchHlCohort, separate Upstash key so the two leaderboards refresh + * independently and a brownout on one doesn't poison the other. */ +async function fetchHlHip3CohortRaw(): Promise { + const snapshot = await readCohortSnapshot(HL_HIP3_KEY); + if (snapshot) return snapshot.data; + const fresh = await fetchHlHip3CohortFresh(); + if (fresh) { + try { + await writeCohortSnapshot(HL_HIP3_KEY, fresh); + } catch (err) { + console.warn( + `hl-hip3 writeback failed: ${ + err instanceof Error ? err.message : String(err) + }`, + ); + } + } + return fresh; +} + +const fetchHlHip3CohortCached = unstable_cache( + fetchHlHip3CohortRaw, + ["hl-hip3-cohort-v1"], + { revalidate: 60, tags: ["hl-cohort"] }, +); + +export async function fetchHlHip3Cohort(): Promise { + return fetchHlHip3CohortCached(); +} + /** * Tiny vector helper. Returns `[{ labels, value }]` from an instant * vector query, or empty on error/empty result. Kept local to this diff --git a/src/lib/perp-stats.ts b/src/lib/perp-stats.ts index 4cacafb1..5b88d7fe 100644 --- a/src/lib/perp-stats.ts +++ b/src/lib/perp-stats.ts @@ -17,6 +17,10 @@ import { unstable_cache } from "next/cache"; import { Prometheus } from "@/lib/prometheus"; +import { + readCohortSnapshot, + writeCohortSnapshot, +} from "@/lib/cohort-snapshot"; export type PerpVenueType = "onchain"; @@ -105,14 +109,24 @@ function promUrl(): string | null { return process.env.PROMETHEUS_URL?.trim() || null; } +/** Upstash key under which the cron writer parks the live cohort snapshot. + * Bumped any time the PerpCohortSummary shape changes so a stale-shape + * blob from a previous deploy can never deserialize into a misaligned + * payload. The cohort-snapshot module appends its own `:v1` suffix. */ +const PERP_COHORT_KEY = "perp-cohort"; + /** * Fetch the cohort in one Promise.all fan out. Returns null when Prom * is unreachable so the page can render a configuration banner instead * of a blank leaderboard. A reachable Prom with no series yet (harness * not yet deployed) returns a fully populated shape with every numeric * field nulled; the leaderboard shows dashes and the page still ships. + * + * Exported so the cron route can call the uncached Prom path directly + * before parking the result in Upstash; the public reader (fetchPerpCohort) + * goes through the snapshot layer + unstable_cache below. */ -export async function fetchPerpCohort(): Promise { +export async function fetchPerpCohortFresh(): Promise { const url = promUrl(); if (!url) return null; let prom: Prometheus; @@ -280,6 +294,53 @@ export async function fetchPerpCohort(): Promise { }; } +/** + * Snapshot-first cohort reader. Order of preference: + * 1. Upstash blob written by the /api/cron/snapshot-perp-cohort writer. + * A live blob (asOf within the 10 min window) is the fast path; this + * is the common case once the cron is running. + * 2. Live Prom fetch via fetchPerpCohortFresh(). Used when the blob is + * missing (first deploy, blob expired, creds unconfigured) or stale + * (writer dead for > 10 min). On a successful live fetch we also + * write the result back so the next reader on this lambda hits the + * snapshot path instead of paying another Prom round-trip. + * 3. null. Returned only when both paths fail; the page already handles + * this with a "data temporarily unavailable" banner. + * + * The whole thing is wrapped in unstable_cache below so concurrent + * requests in the same lambda collapse onto one Upstash GET / one Prom + * fan-out. + */ +async function fetchPerpCohortRaw(): Promise { + const snapshot = await readCohortSnapshot( + PERP_COHORT_KEY, + ); + if (snapshot) return snapshot.data; + const fresh = await fetchPerpCohortFresh(); + if (fresh) { + try { + await writeCohortSnapshot(PERP_COHORT_KEY, fresh); + } catch (err) { + console.warn( + `perp-cohort writeback failed: ${ + err instanceof Error ? err.message : String(err) + }`, + ); + } + } + return fresh; +} + +const fetchPerpCohortCached = unstable_cache( + fetchPerpCohortRaw, + ["perp-cohort-v1"], + { revalidate: 60, tags: ["perp-cohort"] }, +); + +export async function fetchPerpCohort(): Promise { + return fetchPerpCohortCached(); +} + /** * Per-asset cross-venue funding matrix for the "By asset" tab on /perps. * Pivots the harness gauge `perp_venue_funding_24h_bps{venue, asset}` diff --git a/vercel.json b/vercel.json index 7b2ecc38..15552d6c 100644 --- a/vercel.json +++ b/vercel.json @@ -14,6 +14,14 @@ { "path": "/api/cron/indexnow", "schedule": "30 4 * * *" + }, + { + "path": "/api/cron/snapshot-perp-cohort", + "schedule": "* * * * *" + }, + { + "path": "/api/cron/snapshot-hl-cohort", + "schedule": "* * * * *" } ] }