feat(earnings): Phase 3 public read API — per-agent + platform + leaderboard (#978)

app/api/agents/[address]/earnings/route.ts

@@ -0,0 +1,135 @@
+import { NextRequest, NextResponse } from "next/server";
+import { getCloudflareContext } from "@opennextjs/cloudflare";
+import { withEdgeCache, buildEdgeCacheKey } from "@/lib/edge-cache";
+import {
+  classifyAddress,
+  lookupProfileByStxAddress,
+  lookupProfileByBtcAddress,
+  lookupProfileByAgentId,
+} from "@/lib/cache/agent-profile";
+import { getAgentRollup, getAgentLineItems } from "@/lib/earnings/reads";
+
+const DEFAULT_LIMIT = 25;
+const MAX_LIMIT = 100;
+const CACHE_TTL_SECONDS = 300; // 5 min — indexer cadence is 30 min, so plenty fresh.
+
+function selfDoc() {
+  return NextResponse.json(
+    {
+      endpoint: "/api/agents/{address}/earnings",
+      method: "GET",
+      description:
+        "Verified on-chain earnings for one agent: a 7d/30d/lifetime USD rollup plus recent line items. " +
+        "Earnings are indexed from confirmed inbound sBTC/STX/aeUSDC transfers, classified by counterparty, " +
+        "and priced in USD — self-dealing (self-funded / ring / alt-address) is excluded.",
+      pathParameters: {
+        address: "Agent STX address (SP…/SM…), BTC address, or numeric agent id.",
+      },
+      queryParameters: {
+        limit: `Line items per page (1–${MAX_LIMIT}, default ${DEFAULT_LIMIT}).`,
+        offset: "Line item offset (default 0).",
+      },
+      responseFormat: {
+        address: "string (as supplied)",
+        stxAddress: "string (canonical)",
+        rollup: {
+          earnings_7d_usd: "number",
+          earnings_30d_usd: "number",
+          earnings_lifetime_usd: "number",
+          unique_payers_30d: "number",
+          top_source_class_30d: "string | null",
+        },
+        lineItems:
+          "Array<{ txId, eventIndex, blockTime, sender, asset, amountRaw, amountUsd, sourceClass, sourceSubclass, explorerUrl }>",
+        pagination: { limit: "number", offset: "number", hasMore: "boolean" },
+      },
+      relatedEndpoints: { platform: "/api/stats/earnings" },
+    },
+    { headers: { "Cache-Control": "public, max-age=3600, s-maxage=86400" } }
+  );
+}
+
+export async function GET(
+  request: NextRequest,
+  { params }: { params: Promise<{ address: string }> }
+) {
+  const { address } = await params;
+  const url = new URL(request.url);
+  if (url.searchParams.get("docs") === "1") return selfDoc();
+
+  const limit = Math.min(
+    MAX_LIMIT,
+    Math.max(1, Number(url.searchParams.get("limit")) || DEFAULT_LIMIT)
+  );
+  const offset = Math.max(0, Number(url.searchParams.get("offset")) || 0);
+
+  const cacheKey = buildEdgeCacheKey(
+    "/api/agents",
+    address,
+    `/earnings?limit=${limit}&offset=${offset}`
+  );
+
+  // All work runs inside the loader so edge-cache hits skip the DB + context
+  // resolve entirely (mirrors app/api/agents/[address]/route.ts).
+  return withEdgeCache(cacheKey, CACHE_TTL_SECONDS, async () => {
+    const { env } = await getCloudflareContext();
+    const db = env.DB as D1Database | undefined;
+    if (!db) {
+      return NextResponse.json(
+        { error: "Database unavailable." },
+        { status: 503, headers: { "Cache-Control": "no-store" } }
+      );
+    }
+
+    // Resolve any address form → canonical agent stx_address.
+    const branch = classifyAddress(address);
+    let row = null;
+    if (branch === "stx") row = await lookupProfileByStxAddress(db, address);
+    else if (branch === "btc") row = await lookupProfileByBtcAddress(db, address);
+    else if (branch === "numeric") {
+      const id = parseInt(address, 10);
+      if (Number.isFinite(id)) row = await lookupProfileByAgentId(db, id);
+    }
+
+    if (!row) {
+      return NextResponse.json(
+        {
+          error:
+            "Agent not found. Provide a registered STX address (SP…/SM…), BTC address, or numeric agent id.",
+        },
+        { status: 404, headers: { "Cache-Control": "no-store" } }
+      );
+    }
+
+    const stxAddress = row.stx_address;
+    const now = Date.now();
+
+    const [rollup, items] = await Promise.all([
+      getAgentRollup(db, stxAddress, now),
+      getAgentLineItems(db, stxAddress, limit + 1, offset),
+    ]);
+
+    const hasMore = items.length > limit;
+    const lineItems = (hasMore ? items.slice(0, limit) : items).map((i) => ({
+      txId: i.tx_id,
+      eventIndex: i.event_index,
+      blockTime: i.block_time,
+      sender: i.sender_stx,
+      asset: i.asset,
+      amountRaw: i.amount_raw,
+      amountUsd: i.amount_usd,
+      sourceClass: i.source_class,
+      sourceSubclass: i.source_subclass,
+      explorerUrl: `https://explorer.hiro.so/txid/${i.tx_id}?chain=mainnet`,
+    }));
+
+    return NextResponse.json(
+      { address, stxAddress, rollup, lineItems, pagination: { limit, offset, hasMore } },
+      {
+        headers: {
+          "Cache-Control": `public, max-age=${CACHE_TTL_SECONDS}, s-maxage=${CACHE_TTL_SECONDS}`,
+        },
+      }
+    );
+  });
+}

app/api/stats/earnings/route.ts

@@ -0,0 +1,96 @@
+import { NextRequest, NextResponse } from "next/server";
+import { getCloudflareContext } from "@opennextjs/cloudflare";
+import { withEdgeCache, buildEdgeCacheKey } from "@/lib/edge-cache";
+import {
+  getPlatformEarnings,
+  getEarningsLeaderboard,
+  type EarningsWindow,
+} from "@/lib/earnings/reads";
+
+// Fixed top-N ranking, like the trading leaderboard SSR. Crucially the cache key
+// is keyed on WINDOW only (not limit/offset), so the GROUP BY scan runs at most
+// 3×/hour/colo regardless of traffic — no pagination crawl can multiply D1
+// rows-read into a cost spike. Clients slice the top-N for display.
+const LEADERBOARD_SIZE = 100;
+const CACHE_TTL_SECONDS = 3600; // 1h — platform aggregate + ranking change slowly.
+const WINDOWS: ReadonlySet<EarningsWindow> = new Set(["7d", "30d", "lifetime"]);
+
+function parseWindow(raw: string | null): EarningsWindow {
+  return raw && WINDOWS.has(raw as EarningsWindow) ? (raw as EarningsWindow) : "30d";
+}
+
+function selfDoc() {
+  return NextResponse.json(
+    {
+      endpoint: "/api/stats/earnings",
+      method: "GET",
+      description:
+        "Platform-wide verified earnings: total USD earned by all agents over 7d/30d/lifetime, " +
+        "a 30d breakdown by source class, and the top agents ranked by earnings in the chosen window.",
+      queryParameters: {
+        window:
+          "Leaderboard ranking window: 7d | 30d | lifetime (default 30d). Platform totals always include all three.",
+      },
+      responseFormat: {
+        platform: {
+          total_7d_usd: "number",
+          total_30d_usd: "number",
+          total_lifetime_usd: "number",
+          by_source_class_30d: "Array<{ source_class, total_usd }>",
+        },
+        leaderboard: `Top ${LEADERBOARD_SIZE} Array<{ rank, stxAddress, btcAddress, displayName, bnsName, earningsUsd, uniquePayers, latestAt }>`,
+        window: "string",
+      },
+      relatedEndpoints: { perAgent: "/api/agents/{address}/earnings" },
+    },
+    { headers: { "Cache-Control": "public, max-age=3600, s-maxage=86400" } }
+  );
+}
+
+export async function GET(request: NextRequest) {
+  const url = new URL(request.url);
+  if (url.searchParams.get("docs") === "1") return selfDoc();
+
+  const window = parseWindow(url.searchParams.get("window"));
+
+  // Cache key = window only → at most 3 distinct keys, so the leaderboard scan
+  // can never be multiplied by query-param cardinality.
+  const cacheKey = buildEdgeCacheKey("/api/stats", "earnings", `?window=${window}`);
+
+  return withEdgeCache(cacheKey, CACHE_TTL_SECONDS, async () => {
+    const { env } = await getCloudflareContext();
+    const db = env.DB as D1Database | undefined;
+    if (!db) {
+      return NextResponse.json(
+        { error: "Database unavailable." },
+        { status: 503, headers: { "Cache-Control": "no-store" } }
+      );
+    }
+
+    const now = Date.now();
+    const [platform, rows] = await Promise.all([
+      getPlatformEarnings(db, now),
+      getEarningsLeaderboard(db, window, LEADERBOARD_SIZE, 0, now),
+    ]);
+
+    const leaderboard = rows.map((r, i) => ({
+      rank: i + 1,
+      stxAddress: r.stx_address,
+      btcAddress: r.btc_address,
+      displayName: r.display_name,
+      bnsName: r.bns_name,
+      earningsUsd: r.earnings_usd,
+      uniquePayers: r.unique_payers,
+      latestAt: r.latest_at,
+    }));
+
+    return NextResponse.json(
+      { platform, leaderboard, window },
+      {
+        headers: {
+          "Cache-Control": `public, max-age=${CACHE_TTL_SECONDS}, s-maxage=${CACHE_TTL_SECONDS}`,
+        },
+      }
+    );
+  });
+}

docs/earnings-ledger-architecture.md

@@ -372,11 +372,13 @@ would multiply Hiro calls by the cadence with no freshness benefit for a 30-day
   `agent_peer` earnings, wired into the indexer's `resolveRow`. Known limit: a ring
   whose two legs land in the *same* sweep tick isn't caught (the reverse leg isn't
   persisted yet); cross-tick rings are.
-- **Reprice pass (Phase 3 scope).** Phase 1 stores `amount_usd = NULL`,
-  `price_source = 'none'` for transfers indexed during a Tenero gap. There is **no
-  reprice task yet** — add one in Phase 3 (a bounded sweep over `price_source = 'none'`
-  rows that re-reads the Tenero cache), so the gap doesn't get lost between phases.
-- **Phase 3 — Public API.** `/api/agents/{addr}/earnings`, `/api/stats/earnings`,
+- **Reprice pass (small follow-up, post-Phase 3).** Phase 1 stores `amount_usd = NULL`,
+  `price_source = 'none'` for transfers indexed during a Tenero gap. A bounded reprice
+  sweep (over a `price_source = 'none'` partial index) re-reads the Tenero cache and
+  fills them — deferred to keep the Phase 3 API PR focused; tracked here so it isn't
+  lost.
+- **Phase 3 — Public API. DONE.** `/api/agents/{addr}/earnings` (per-agent rollup +
+  line items), `/api/stats/earnings` (platform totals + ranked earnings leaderboard),
   trading-board earnings ranking (all read-time + edge-cached).
 - **Phase 4 — UI.** leaderboard chip + new default, profile Earnings section, homepage
   hero stat, `$10–$100k` Club badges.

lib/earnings/__tests__/reads.test.ts

@@ -0,0 +1,113 @@
+import { describe, it, expect, vi } from "vitest";
+import {
+  windowStart,
+  getAgentRollup,
+  getPlatformEarnings,
+  getEarningsLeaderboard,
+} from "../reads";
+
+const NOW = 1_000_000_000_000; // fixed unix ms
+const NOW_SEC = Math.floor(NOW / 1000);
+const DAY = 86_400;
+
+describe("windowStart", () => {
+  it("computes 7d / 30d / lifetime bounds in unix seconds", () => {
+    expect(windowStart("7d", NOW)).toBe(NOW_SEC - 7 * DAY);
+    expect(windowStart("30d", NOW)).toBe(NOW_SEC - 30 * DAY);
+    expect(windowStart("lifetime", NOW)).toBe(0);
+  });
+});
+
+/** D1 mock returning preset rows per query, capturing bind args. */
+function makeDb(handlers: { match: (sql: string) => boolean; first?: unknown; all?: unknown[] }[]) {
+  const calls: { sql: string; args: unknown[] }[] = [];
+  return {
+    calls,
+    db: {
+      prepare: (sql: string) => ({
+        bind: (...args: unknown[]) => ({
+          first: async () => {
+            calls.push({ sql, args });
+            const h = handlers.find((x) => x.match(sql));
+            return h?.first ?? null;
+          },
+          all: async () => {
+            calls.push({ sql, args });
+            const h = handlers.find((x) => x.match(sql));
+            return { results: h?.all ?? [] };
+          },
+        }),
+      }),
+    } as unknown as D1Database,
+  };
+}
+
+describe("getAgentRollup", () => {
+  it("maps the totals + top source class query results", async () => {
+    const { db } = makeDb([
+      { match: (s) => s.includes("COUNT(DISTINCT"), first: { e7: 10, e30: 42.5, elife: 100, payers30: 3 } },
+      { match: (s) => s.includes("GROUP BY source_class"), first: { source_class: "inbox_message" } },
+    ]);
+    const r = await getAgentRollup(db, "SP_AGENT", NOW);
+    expect(r).toEqual({
+      earnings_7d_usd: 10,
+      earnings_30d_usd: 42.5,
+      earnings_lifetime_usd: 100,
+      unique_payers_30d: 3,
+      top_source_class_30d: "inbox_message",
+    });
+  });
+
+  it("defaults to zeros / null when the agent has no earnings", async () => {
+    const { db } = makeDb([]);
+    const r = await getAgentRollup(db, "SP_NEW", NOW);
+    expect(r).toEqual({
+      earnings_7d_usd: 0,
+      earnings_30d_usd: 0,
+      earnings_lifetime_usd: 0,
+      unique_payers_30d: 0,
+      top_source_class_30d: null,
+    });
+  });
+});
+
+describe("getPlatformEarnings", () => {
+  it("returns totals + the 30d source breakdown", async () => {
+    const { db } = makeDb([
+      { match: (s) => s.includes("FROM agent_earnings WHERE is_earning = 1"), first: { e7: 5, e30: 20, elife: 50 } },
+      {
+        match: (s) => s.includes("GROUP BY source_class"),
+        all: [
+          { source_class: "inbox_message", total_usd: 12 },
+          { source_class: "bounty", total_usd: 8 },
+        ],
+      },
+    ]);
+    const r = await getPlatformEarnings(db, NOW);
+    expect(r.total_7d_usd).toBe(5);
+    expect(r.total_30d_usd).toBe(20);
+    expect(r.total_lifetime_usd).toBe(50);
+    expect(r.by_source_class_30d).toHaveLength(2);
+    expect(r.by_source_class_30d[0]).toEqual({ source_class: "inbox_message", total_usd: 12 });
+  });
+});
+
+describe("getEarningsLeaderboard", () => {
+  it("binds the window start and returns ranked rows", async () => {
+    const { db, calls } = makeDb([
+      {
+        match: (s) => s.includes("LEFT JOIN agents"),
+        all: [
+          { stx_address: "SP_A", btc_address: "bc1a", display_name: "A", bns_name: null, earnings_usd: 99, unique_payers: 4, latest_at: 123 },
+        ],
+      },
+    ]);
+    const rows = await getEarningsLeaderboard(db, "30d", 20, 0, NOW);
+    expect(rows).toHaveLength(1);
+    expect(rows[0].stx_address).toBe("SP_A");
+    // first bind arg is the window start (30d ago, unix seconds)
+    expect(calls[0].args[0]).toBe(NOW_SEC - 30 * DAY);
+    expect(calls[0].args[1]).toBe(20); // limit
+    expect(calls[0].args[2]).toBe(0); // offset
+  });
+});