feat: snapshot + carry-forward for /perps and /hyperliquid cohorts

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/benches/hyperliquid/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<HlCohortSummary | null> {
+export async function fetchHlCohortFresh(): Promise<HlCohortSummary | null> {
  const url = promUrl();
  if (!url) return null;
  let prom: Prometheus;
@@ -349,8 +364,13 @@ export async function fetchHlCohort(): Promise<HlCohortSummary | null> {
  *
  * 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<HlHip3Summary | null> {
+export async function fetchHlHip3CohortFresh(): Promise<HlHip3Summary | null> {
  const url = promUrl();
  if (!url) return null;
  let prom: Prometheus;
@@ -467,6 +487,72 @@ export async function fetchHlHip3Cohort(): Promise<HlHip3Summary | null> {
  };
 }
 
+/**
+ * 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<HlCohortSummary | null> {
+ const snapshot = await readCohortSnapshot<HlCohortSummary>(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<HlCohortSummary | null> {
+ 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<HlHip3Summary | null> {
+ const snapshot = await readCohortSnapshot<HlHip3Summary>(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<HlHip3Summary | null> {
+ return fetchHlHip3CohortCached();
+}
+
 /**
  * Tiny vector helper. Returns `[{ labels, value }]` from an instant
  * vector query, or empty on error/empty result. Kept local to this