chore(prod): cherry-pick cohort materialize for /perps + /hyperliquid

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<T extends { asOf: number; rows: unknown[] }>(
+ key: string,
+ fetcher: () => Promise<T | null>,
+): Promise<KeyOutcome> {
+ 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 },
+ );
+}

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<ReturnType<typeof fetchPerpCohortFresh>>;
+ 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,
+ });
+}

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<T> = { 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<unknown> {
+ 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<T>(
+ key: string,
+ data: T,
+): Promise<void> {
+ if (!isConfigured()) return;
+ const envelope: Envelope<T> = { 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<T>(
+ 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<T>;
+ 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();
+}