feat(data-api): add the account_events_daily rollup write path to Postgres

.github/workflows/rollup-account-events-daily.yml

@@ -0,0 +1,43 @@
+name: Rollup account_events_daily (Postgres)
+
+# Triggers the Postgres-side account_events_daily rollup (#4832 gap-closure):
+# indexer-rs writes account_events continuously and directly into Postgres
+# (not through any Worker route), so unlike refresh-metagraph.yml's
+# neurons-sync step there is no existing write request to piggyback this on
+# -- a bare hourly trigger is the whole job. Mirrors D1's own
+# rollupAccountEventsDaily cadence (src/account-events.mjs, called from the
+# HEALTH_PRUNE_CRON hourly tick) so the two tiers stay on the same schedule.
+# See workers/data-api.mjs's handleRollupAccountEventsDaily for the actual
+# rollup SQL.
+on:
+  schedule:
+    - cron: "17 * * * *" # hourly, offset from the top of the hour
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+concurrency:
+  group: rollup-account-events-daily
+  cancel-in-progress: false
+
+jobs:
+  rollup:
+    runs-on: ubuntu-latest
+    timeout-minutes: 5
+    steps:
+      - name: Checkout
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        with:
+          persist-credentials: false
+
+      - name: Trigger the Postgres rollup
+        run: |
+          if [ -z "$ROLLUP_SYNC_SECRET" ]; then
+            echo "ROLLUP_SYNC_SECRET not set; skipping"
+            exit 0
+          fi
+          curl -fsS -X POST "https://api.metagraph.sh/api/v1/internal/rollup-account-events-daily" \
+            -H "x-rollup-sync-token: ${ROLLUP_SYNC_SECRET}"
+        env:
+          ROLLUP_SYNC_SECRET: ${{ secrets.ROLLUP_SYNC_SECRET }}

tests/data-api.test.mjs

@@ -3,6 +3,7 @@
 // Hyperdrive→Railway path is validated separately.
 import { beforeEach, test, expect, vi } from "vitest";
 import { BLOCK_PAGINATION, MAX_OFFSET } from "../workers/request-params.mjs";
+import { encodeCursor } from "../src/cursor.mjs";
 
 const sqlCalls = vi.hoisted(() => []);
 const mockRows = vi.hoisted(() => ({
@@ -30,6 +31,8 @@ const mockQueue = vi.hoisted(() => ({ current: [] }));
 // every GET-route test above.
 const neuronsSyncFailure = vi.hoisted(() => ({ error: null }));
 const neuronsSyncPruneRows = vi.hoisted(() => ({ current: [] }));
+// State for the account-events-daily rollup write route's tests only (#4832).
+const rollupFailure = vi.hoisted(() => ({ error: null }));
 
 vi.mock("postgres", () => ({
   default: () => {
@@ -74,6 +77,12 @@ vi.mock("postgres", () => ({
       if (neuronsSyncFailure.error && /INSERT INTO neurons\b/.test(text)) {
         return Promise.reject(neuronsSyncFailure.error);
       }
+      if (
+        rollupFailure.error &&
+        /INSERT INTO account_events_daily/.test(text)
+      ) {
+        return Promise.reject(rollupFailure.error);
+      }
       if (/DELETE FROM neurons/.test(text)) {
         return Promise.resolve(neuronsSyncPruneRows.current);
       }
@@ -110,9 +119,11 @@ vi.mock("postgres", () => ({
 
 const { default: worker } = await import("../workers/data-api.mjs");
 const NEURONS_SYNC_SECRET = "test-neurons-sync-secret";
+const ROLLUP_SYNC_SECRET = "test-rollup-sync-secret";
 const env = {
   HYPERDRIVE: { connectionString: "postgres://mock" },
   NEURONS_SYNC_SECRET,
+  ROLLUP_SYNC_SECRET,
 };
 const ctx = { waitUntil() {} };
 const req = (path, init) =>
@@ -124,6 +135,7 @@ beforeEach(() => {
   mockQueue.current = [];
   neuronsSyncFailure.error = null;
   neuronsSyncPruneRows.current = [];
+  rollupFailure.error = null;
   mockRows.current = [
     {
       block_number: "123",
@@ -2606,3 +2618,138 @@ test("GET /api/v1/accounts/:ss58/counterparties?counterparty= returns the relati
   expect(body.counterparty).toBe("5Cold");
   expect(queryText()).toContain("(hotkey = ");
 });
+
+// #4832 gap-closure: POST /api/v1/internal/rollup-account-events-daily -- the
+// account_events_daily write path account_events itself lacked (indexer-rs
+// writes account_events continuously, but nothing rolled it into the daily
+// summary table in Postgres), plus its read path,
+// GET /api/v1/accounts/:ss58/history.
+function postRollup({ secret } = {}) {
+  const headers = {};
+  if (secret !== undefined) headers["x-rollup-sync-token"] = secret;
+  return req("/api/v1/internal/rollup-account-events-daily", {
+    method: "POST",
+    headers,
+  });
+}
+
+test("rollup-account-events-daily rejects a missing or wrong token (401)", async () => {
+  const wrong = await postRollup({ secret: "wrong" });
+  expect(wrong.status).toBe(401);
+  const missing = await postRollup();
+  expect(missing.status).toBe(401);
+});
+
+test("rollup-account-events-daily is disabled (503) when ROLLUP_SYNC_SECRET is not configured", async () => {
+  const res = await worker.fetch(
+    new Request("https://d/api/v1/internal/rollup-account-events-daily", {
+      method: "POST",
+      headers: { "x-rollup-sync-token": ROLLUP_SYNC_SECRET },
+    }),
+    { HYPERDRIVE: { connectionString: "postgres://mock" } },
+    ctx,
+  );
+  expect(res.status).toBe(503);
+});
+
+test("rollup-account-events-daily returns 503 when the HYPERDRIVE binding is unavailable", async () => {
+  const res = await worker.fetch(
+    new Request("https://d/api/v1/internal/rollup-account-events-daily", {
+      method: "POST",
+      headers: { "x-rollup-sync-token": ROLLUP_SYNC_SECRET },
+    }),
+    { ROLLUP_SYNC_SECRET },
+    ctx,
+  );
+  expect(res.status).toBe(503);
+});
+
+test("rollup-account-events-daily rolls up the two active UTC days and reports them", async () => {
+  const res = await postRollup({ secret: ROLLUP_SYNC_SECRET });
+  expect(res.status).toBe(200);
+  const body = await res.json();
+  expect(body.ok).toBe(true);
+  expect(body.rolled).toHaveLength(2);
+  expect(queryText()).toMatch(/INSERT INTO account_events_daily/);
+  expect(queryText()).toMatch(/FROM account_events/);
+  expect(queryText()).toMatch(/string_agg\(DISTINCT event_kind/);
+});
+
+test("rollup-account-events-daily maps a DB failure to a clean 502 instead of throwing", async () => {
+  rollupFailure.error = new Error("connection reset");
+  const res = await postRollup({ secret: ROLLUP_SYNC_SECRET });
+  expect(res.status).toBe(502);
+  expect((await res.json()).error).toBe("rollup failed");
+});
+
+test("GET /api/v1/accounts/:ss58/history shapes the durable per-day activity series", async () => {
+  mockRows.current = [
+    {
+      day: "2026-07-01",
+      netuid: 7,
+      event_count: "3",
+      event_kinds: "StakeAdded,WeightsSet",
+      first_block: "100",
+      last_block: "200",
+    },
+  ];
+  const res = await req(`/api/v1/accounts/${SS58}/history`);
+  expect(res.status).toBe(200);
+  const body = await res.json();
+  expect(body.ss58).toBe(SS58);
+  expect(body.days[0].day).toBe("2026-07-01");
+  expect(queryText()).toContain("FROM account_events_daily");
+  expect(queryText()).toContain("WHERE hotkey =");
+});
+
+test("GET /api/v1/accounts/:ss58/history?netuid= filters to one subnet", async () => {
+  mockRows.current = [
+    {
+      day: "2026-07-01",
+      netuid: 7,
+      event_count: "1",
+      event_kinds: "StakeAdded",
+      first_block: "100",
+      last_block: "100",
+    },
+  ];
+  const res = await req(`/api/v1/accounts/${SS58}/history?netuid=7`);
+  expect(res.status).toBe(200);
+  expect(queryText()).toContain("AND netuid =");
+});
+
+test("GET /api/v1/accounts/:ss58/history?from=&to= filters the day range", async () => {
+  mockRows.current = [];
+  const res = await req(
+    `/api/v1/accounts/${SS58}/history?from=2026-06-01&to=2026-06-30`,
+  );
+  expect(res.status).toBe(200);
+  expect(queryText()).toContain("AND day >=");
+  expect(queryText()).toContain("AND day <=");
+});
+
+test("GET /api/v1/accounts/:ss58/history?cursor= seeks past the encoded (day, netuid) pair", async () => {
+  mockRows.current = [];
+  const cursor = encodeCursor([20260701, 7]);
+  const res = await req(`/api/v1/accounts/${SS58}/history?cursor=${cursor}`);
+  expect(res.status).toBe(200);
+  expect(queryText()).toContain("AND (day, netuid) <");
+  expect(queryText()).not.toContain("OFFSET");
+});
+
+test("GET /api/v1/accounts/:ss58/history?limit=1 emits a next_cursor when the page is full", async () => {
+  mockRows.current = [
+    {
+      day: "2026-07-01",
+      netuid: 7,
+      event_count: "1",
+      event_kinds: "StakeAdded",
+      first_block: "100",
+      last_block: "100",
+    },
+  ];
+  const res = await req(`/api/v1/accounts/${SS58}/history?limit=1`);
+  expect(res.status).toBe(200);
+  const body = await res.json();
+  expect(body.next_cursor).not.toBeNull();
+});

tests/request-handlers-entities.test.mjs

@@ -8539,6 +8539,48 @@ describe("D1 -> Postgres serving-cutover flag (#4656 followup)", () => {
     assert.equal(body.data.marker, undefined);
     assert.ok(captures.sql.length > 0);
   });
+
+  // #4832 gap-closure: handleAccountHistory (account_events_daily, now
+  // populated by a dedicated hourly Postgres-side rollup route).
+
+  test("handleAccountHistory: flag=postgres uses Postgres data, D1 never queried", async () => {
+    const { env, captures } = dbWith({});
+    env.METAGRAPH_ACCOUNT_EVENTS_SOURCE = "postgres";
+    env.DATA_API = {
+      fetch: async () =>
+        Response.json({ schema_version: 1, marker: "pg", days: [] }),
+    };
+    const body = await json(
+      await handleAccountHistory(
+        req(`/api/v1/accounts/${SS58}/history`),
+        env,
+        SS58,
+        url(`/api/v1/accounts/${SS58}/history`),
+      ),
+    );
+    assert.equal(body.data.marker, "pg");
+    assert.deepEqual(captures.sql, []);
+  });
+
+  test("handleAccountHistory: flag=postgres falls back to D1 on failure", async () => {
+    const { env, captures } = dbWith({});
+    env.METAGRAPH_ACCOUNT_EVENTS_SOURCE = "postgres";
+    env.DATA_API = {
+      fetch: async () => {
+        throw new Error("boom");
+      },
+    };
+    const body = await json(
+      await handleAccountHistory(
+        req(`/api/v1/accounts/${SS58}/history`),
+        env,
+        SS58,
+        url(`/api/v1/accounts/${SS58}/history`),
+      ),
+    );
+    assert.equal(body.data.marker, undefined);
+    assert.ok(captures.sql.length > 0);
+  });
 });
 
 // ---- Cross-handler contract smoke tests -------------------------------------

tests/rollup-account-events-daily-proxy.test.mjs

@@ -0,0 +1,108 @@
+// Unit tests for the /api/v1/internal/rollup-account-events-daily proxy
+// (workers/api.mjs's handleRollupAccountEventsDailyProxy, #4832), which
+// forwards to workers/data-api.mjs's handleRollupAccountEventsDaily via the
+// EXISTING DATA_API service binding (shares proxyToDataApi with
+// handleNeuronsSyncProxy -- see tests/neurons-sync-proxy.test.mjs for that
+// sibling route's equivalent coverage). The downstream rollup logic itself
+// is covered by tests/data-api.test.mjs.
+import assert from "node:assert/strict";
+import { test } from "vitest";
+import { handleRequest } from "../workers/api.mjs";
+
+function post({ method = "POST" } = {}) {
+  return new Request(
+    "https://api.metagraph.sh/api/v1/internal/rollup-account-events-daily",
+    { method },
+  );
+}
+
+test("rejects non-POST before reaching the binding (405)", async () => {
+  let calls = 0;
+  const res = await handleRequest(
+    post({ method: "GET" }),
+    {
+      DATA_API: {
+        fetch() {
+          calls += 1;
+          return new Response("{}", { status: 200 });
+        },
+      },
+    },
+    {},
+  );
+  assert.equal(res.status, 405);
+  assert.equal(calls, 0);
+});
+
+test("returns 503 when DATA_API is not bound", async () => {
+  const res = await handleRequest(post(), {}, {});
+  assert.equal(res.status, 503);
+});
+
+test("forwards the request to DATA_API and relays its response body + status", async () => {
+  let receivedToken;
+  let receivedPath;
+  const res = await handleRequest(
+    new Request(
+      "https://api.metagraph.sh/api/v1/internal/rollup-account-events-daily",
+      { method: "POST", headers: { "x-rollup-sync-token": "shared-secret" } },
+    ),
+    {
+      DATA_API: {
+        fetch(req) {
+          receivedToken = req.headers.get("x-rollup-sync-token");
+          receivedPath = new URL(req.url).pathname;
+          return new Response(
+            JSON.stringify({ ok: true, rolled: ["2026-07-01", "2026-06-30"] }),
+            { status: 200 },
+          );
+        },
+      },
+    },
+    {},
+  );
+  assert.equal(receivedToken, "shared-secret");
+  assert.equal(receivedPath, "/api/v1/internal/rollup-account-events-daily");
+  assert.equal(res.status, 200);
+  assert.deepEqual(await res.json(), {
+    ok: true,
+    rolled: ["2026-07-01", "2026-06-30"],
+  });
+});
+
+test("relays a non-2xx upstream status (e.g. 401) unchanged", async () => {
+  const res = await handleRequest(
+    post(),
+    {
+      DATA_API: {
+        fetch() {
+          return new Response(JSON.stringify({ error: "unauthorized" }), {
+            status: 401,
+          });
+        },
+      },
+    },
+    {},
+  );
+  assert.equal(res.status, 401);
+  assert.deepEqual(await res.json(), { error: "unauthorized" });
+});
+
+test("returns 502 when the upstream response body is unreadable", async () => {
+  const res = await handleRequest(
+    post(),
+    {
+      DATA_API: {
+        fetch() {
+          return new Response("not json", { status: 200 });
+        },
+      },
+    },
+    {},
+  );
+  assert.equal(res.status, 502);
+  assert.equal(
+    (await res.json()).error.code,
+    "rollup_account_events_daily_unavailable",
+  );
+});