From a493ae33c84ce5e33915f7848d1d0e725c82beec Mon Sep 17 00:00:00 2001 From: JSONbored <49853598+JSONbored@users.noreply.github.com> Date: Fri, 10 Jul 2026 21:32:58 -0700 Subject: [PATCH] feat(data-api): migrate account_position_daily to Postgres MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Gap-closure from the #4832 completeness sweep: account_position_daily was D1-only because its Postgres table didn't exist yet. Its source data (the `neurons` snapshot) is already synced into Postgres via handleNeuronsSync (#4771), so this rolls it into the SAME write transaction rather than standing up new infrastructure — no new external trigger needed. - Add the table + indexes to deploy/postgres/schema.sql (mirrors D1 migrations/0038_account_position_daily.sql). - Extend handleNeuronsSync to upsert account_position_daily rows (hotkey IS NOT NULL, same filter as the D1 rollup) in the same sql.begin() transaction as neurons/neuron_daily. - Add the read route (GET /api/v1/accounts/:ss58/subnets/:netuid/history) and wire tryPostgresTier into handleAccountPositionHistory. Refs #4832 --- deploy/postgres/schema.sql | 28 ++++++++ tests/data-api.test.mjs | 71 ++++++++++++++++++- tests/request-handlers-entities.test.mjs | 45 ++++++++++++ workers/data-api.mjs | 90 ++++++++++++++++++++++++ workers/request-handlers/entities.mjs | 9 +-- 5 files changed, 238 insertions(+), 5 deletions(-) diff --git a/deploy/postgres/schema.sql b/deploy/postgres/schema.sql index f9de26ffd..fa88ddf0a 100644 --- a/deploy/postgres/schema.sql +++ b/deploy/postgres/schema.sql @@ -179,6 +179,34 @@ CREATE INDEX IF NOT EXISTS idx_nd_netuid_date ON neuron_daily (netuid, snapshot_ CREATE INDEX IF NOT EXISTS idx_nd_uid_date ON neuron_daily (netuid, uid, snapshot_date); CREATE INDEX IF NOT EXISTS idx_nd_hotkey_date ON neuron_daily (hotkey, snapshot_date); +-- Per-account daily position HISTORY (#4832 gap-closure; mirrors D1 +-- migrations/0038_account_position_daily.sql). Rolled from the SAME `neurons` +-- snapshot as neuron_daily, in the SAME handleNeuronsSync write (#4771) -- +-- account = hotkey ss58, matching loadAccountPortfolio's "WHERE hotkey = ?" +-- framing (src/account-portfolio.mjs). +CREATE TABLE IF NOT EXISTS account_position_daily ( + account TEXT NOT NULL, + netuid INTEGER NOT NULL, + snapshot_date DATE NOT NULL, + uid INTEGER, + coldkey TEXT, + active BOOLEAN, + validator_permit BOOLEAN, + rank NUMERIC, + trust NUMERIC, + incentive NUMERIC, + dividends NUMERIC, + stake_tao NUMERIC, + emission_tao NUMERIC, + captured_at BIGINT NOT NULL, + updated_at BIGINT NOT NULL, + PRIMARY KEY (account, netuid, snapshot_date) +); +CREATE INDEX IF NOT EXISTS idx_account_position_daily_netuid_date + ON account_position_daily (netuid, snapshot_date); +CREATE INDEX IF NOT EXISTS idx_account_position_daily_date + ON account_position_daily (snapshot_date); + -- Account daily rollup (#2079 / audit: removes the temp-sort on default account history). CREATE TABLE IF NOT EXISTS account_events_daily ( hotkey TEXT NOT NULL, diff --git a/tests/data-api.test.mjs b/tests/data-api.test.mjs index 22a7110bf..42fb35f28 100644 --- a/tests/data-api.test.mjs +++ b/tests/data-api.test.mjs @@ -1660,6 +1660,62 @@ test("GET /api/v1/subnets/movers reports subnet_count:0 on a cold store (no boun expect(body.subnet_count).toBe(0); }); +// #4832 gap-closure: GET /api/v1/accounts/:ss58/subnets/:netuid/history, the +// read path for the account_position_daily rollup added to handleNeuronsSync. + +test("GET /api/v1/accounts/:ss58/subnets/:netuid/history shapes one wallet's per-subnet position trend", async () => { + mockRows.current = [ + { + snapshot_date: "2026-07-01", + captured_at: "1780000000000", + uid: 3, + coldkey: "5Cold", + active: true, + validator_permit: true, + rank: "0.5", + trust: "0.9", + incentive: "0.6", + dividends: "0.4", + stake_tao: "10", + emission_tao: "1", + }, + ]; + const res = await req(`/api/v1/accounts/${SS58}/subnets/7/history`); + expect(res.status).toBe(200); + const body = await res.json(); + expect(body.ss58).toBe(SS58); + expect(body.netuid).toBe(7); + expect(body.points[0].snapshot_date).toBe("2026-07-01"); + expect(queryText()).toContain("FROM account_position_daily"); + expect(queryText()).toContain("WHERE account ="); +}); + +test("GET /api/v1/accounts/:ss58/subnets/:netuid/history?window=all skips the snapshot_date cutoff", async () => { + mockRows.current = [ + { + snapshot_date: "2026-01-01", + captured_at: "1780000000000", + uid: 3, + coldkey: "5Cold", + active: true, + validator_permit: true, + rank: "0.5", + trust: "0.9", + incentive: "0.6", + dividends: "0.4", + stake_tao: "10", + emission_tao: "1", + }, + ]; + const res = await req( + `/api/v1/accounts/${SS58}/subnets/7/history?window=all`, + ); + expect(res.status).toBe(200); + const body = await res.json(); + expect(body.window).toBe("all"); + expect(queryText()).not.toContain("snapshot_date >="); +}); + // #4771: POST /api/v1/internal/neurons-sync -- the one write route in this // otherwise-read-only Worker (see workers/data-api.mjs's handleNeuronsSync). function neuronSyncRow(overrides = {}) { @@ -1841,7 +1897,7 @@ test("neurons-sync rejects an empty array (400)", async () => { expect(res.status).toBe(400); }); -test("neurons-sync upserts neurons + neuron_daily and reports written counts", async () => { +test("neurons-sync upserts neurons + neuron_daily + account_position_daily and reports written counts", async () => { const res = await postNeurons( [neuronSyncRow(), neuronSyncRow({ uid: 4, netuid: 9 })], { secret: NEURONS_SYNC_SECRET }, @@ -1852,13 +1908,26 @@ test("neurons-sync upserts neurons + neuron_daily and reports written counts", a ok: true, neurons_written: 2, neuron_daily_written: 2, + account_position_daily_written: 2, netuids_covered: 2, }); expect(queryText()).toMatch(/INSERT INTO neurons\b/); expect(queryText()).toMatch(/INSERT INTO neuron_daily/); + expect(queryText()).toMatch(/INSERT INTO account_position_daily/); expect(queryText()).toMatch(/DELETE FROM neurons/); }); +test("neurons-sync skips account_position_daily for a row with a null hotkey", async () => { + const { hotkey: _hotkey, ...rest } = neuronSyncRow(); + const res = await postNeurons([{ ...rest, hotkey: null }], { + secret: NEURONS_SYNC_SECRET, + }); + expect(res.status).toBe(200); + const body = await res.json(); + expect(body.account_position_daily_written).toBe(0); + expect(queryText()).not.toMatch(/INSERT INTO account_position_daily/); +}); + test("neurons-sync computes one max captured_at per netuid across its many UID rows (the realistic multi-UID-per-subnet case)", async () => { // Real payloads have ~256 UID rows per netuid sharing one captured_at; a // later row for a netuid already seen must not incorrectly lower or diff --git a/tests/request-handlers-entities.test.mjs b/tests/request-handlers-entities.test.mjs index afacd2d45..156bea96e 100644 --- a/tests/request-handlers-entities.test.mjs +++ b/tests/request-handlers-entities.test.mjs @@ -66,6 +66,7 @@ import { handleAccountStakeFlow, handleAccountStakeMoves, handleAccountSubnets, + handleAccountPositionHistory, handleSubnetEventSummary, handleSubnetEvents, handleAccountBalance, @@ -8494,6 +8495,50 @@ describe("D1 -> Postgres serving-cutover flag (#4656 followup)", () => { assert.equal(body.data.marker, undefined); assert.ok(captures.sql.length > 0); }); + + // #4832 gap-closure: handleAccountPositionHistory (account_position_daily, + // rolled from the same neurons snapshot as neuron_daily). + + test("handleAccountPositionHistory: flag=postgres uses Postgres data, D1 never queried", async () => { + const { env, captures } = dbWith({}); + env.METAGRAPH_NEURONS_SOURCE = "postgres"; + env.DATA_API = { + fetch: async () => + Response.json({ schema_version: 1, marker: "pg", points: [] }), + }; + const body = await json( + await handleAccountPositionHistory( + req(`/api/v1/accounts/${SS58}/subnets/${NETUID}/history`), + env, + SS58, + NETUID, + url(`/api/v1/accounts/${SS58}/subnets/${NETUID}/history`), + ), + ); + assert.equal(body.data.marker, "pg"); + assert.deepEqual(captures.sql, []); + }); + + test("handleAccountPositionHistory: flag=postgres falls back to D1 on failure", async () => { + const { env, captures } = dbWith({}); + env.METAGRAPH_NEURONS_SOURCE = "postgres"; + env.DATA_API = { + fetch: async () => { + throw new Error("boom"); + }, + }; + const body = await json( + await handleAccountPositionHistory( + req(`/api/v1/accounts/${SS58}/subnets/${NETUID}/history`), + env, + SS58, + NETUID, + url(`/api/v1/accounts/${SS58}/subnets/${NETUID}/history`), + ), + ); + assert.equal(body.data.marker, undefined); + assert.ok(captures.sql.length > 0); + }); }); // ---- Cross-handler contract smoke tests ------------------------------------- diff --git a/workers/data-api.mjs b/workers/data-api.mjs index 948dca3bc..e6bfc785a 100644 --- a/workers/data-api.mjs +++ b/workers/data-api.mjs @@ -288,6 +288,7 @@ import { GLOBAL_VALIDATOR_LIMIT_MAX, NEURON_INSERT_COLUMNS, } from "../src/metagraph-neurons.mjs"; +import { buildAccountPositionHistory } from "../src/account-position-history.mjs"; const MAX_LIMIT = 200; const DEFAULT_LIMIT = 50; @@ -554,6 +555,58 @@ async function handleNeuronsSync(request, env) { WHERE neuron_daily.captured_at <= EXCLUDED.captured_at`; } + // Per-account daily position rollup (#4832 gap-closure, mirrors D1's + // rollupAccountPositionDaily / src/account-position-history.mjs): the SAME + // snapshot as neuron_daily above, re-keyed by (account, netuid, + // snapshot_date) with account = hotkey. `hotkey IS NOT NULL` mirrors the D1 + // rollup's own filter -- account is NOT NULL and part of the primary key, + // but a neuron row's hotkey can itself be null. + const positionRows = dailyRows + .filter((row) => row.hotkey != null) + .map((row) => ({ + account: row.hotkey, + netuid: row.netuid, + snapshot_date: row.snapshot_date, + uid: row.uid, + coldkey: row.coldkey, + active: row.active, + validator_permit: row.validator_permit, + rank: row.rank, + trust: row.trust, + incentive: row.incentive, + dividends: row.dividends, + stake_tao: row.stake_tao, + emission_tao: row.emission_tao, + captured_at: row.captured_at, + updated_at: row.updated_at, + })); + for ( + let i = 0; + i < positionRows.length; + i += NEURONS_SYNC_ROWS_PER_STATEMENT + ) { + const chunk = positionRows.slice( + i, + i + NEURONS_SYNC_ROWS_PER_STATEMENT, + ); + await sql` + INSERT INTO account_position_daily ${sql(chunk, "account", "netuid", "snapshot_date", "uid", "coldkey", "active", "validator_permit", "rank", "trust", "incentive", "dividends", "stake_tao", "emission_tao", "captured_at", "updated_at")} + ON CONFLICT (account, netuid, snapshot_date) DO UPDATE SET + uid = EXCLUDED.uid, + coldkey = EXCLUDED.coldkey, + active = EXCLUDED.active, + validator_permit = EXCLUDED.validator_permit, + rank = EXCLUDED.rank, + trust = EXCLUDED.trust, + incentive = EXCLUDED.incentive, + dividends = EXCLUDED.dividends, + stake_tao = EXCLUDED.stake_tao, + emission_tao = EXCLUDED.emission_tao, + captured_at = EXCLUDED.captured_at, + updated_at = EXCLUDED.updated_at + WHERE account_position_daily.captured_at <= EXCLUDED.captured_at`; + } + // Prune UIDs that no longer appear in the snapshot for a netuid this // batch actually covers (deregistered/replaced UIDs) -- scoped to ONLY // the netuids present in this payload, so a partial-coverage batch can @@ -595,6 +648,7 @@ async function handleNeuronsSync(request, env) { ok: true, neurons_written: rows.length, neuron_daily_written: dailyRows.length, + account_position_daily_written: positionRows.length, netuids_covered: netuids.length, deregistered_pruned: pruned.length, }); @@ -2646,6 +2700,42 @@ export default { ); } + // GET /api/v1/accounts/:ss58/subnets/:netuid/history?window= (#4832 + // gap-closure): one wallet's position on one subnet over time, + // mirroring src/account-position-history.mjs's buildAccountPositionHistory. + const positionHistoryMatch = url.pathname.match( + /^\/api\/v1\/accounts\/([^/]+)\/subnets\/(\d+)\/history$/, + ); + if (positionHistoryMatch) { + const ss58 = decodeURIComponent(positionHistoryMatch[1]); + const netuid = Number(positionHistoryMatch[2]); + const cutoff = windowCutoffDate( + url, + HISTORY_WINDOWS, + DEFAULT_HISTORY_WINDOW, + ); + const rows = cutoff + ? await sql` + SELECT snapshot_date::text AS snapshot_date, captured_at, uid, coldkey, active, validator_permit, rank, trust, incentive, dividends, stake_tao, emission_tao + FROM account_position_daily + WHERE account = ${ss58} AND netuid = ${netuid} AND snapshot_date >= ${cutoff} + ORDER BY snapshot_date DESC LIMIT ${MAX_HISTORY_POINTS}` + : await sql` + SELECT snapshot_date::text AS snapshot_date, captured_at, uid, coldkey, active, validator_permit, rank, trust, incentive, dividends, stake_tao, emission_tao + FROM account_position_daily + WHERE account = ${ss58} AND netuid = ${netuid} + ORDER BY snapshot_date DESC LIMIT ${MAX_HISTORY_POINTS}`; + return json( + buildAccountPositionHistory(rows, ss58, netuid, { + window: windowLabelFor( + url, + HISTORY_WINDOWS, + DEFAULT_HISTORY_WINDOW, + ), + }), + ); + } + return json({ error: "not found" }, 404); }); } catch (err) { diff --git a/workers/request-handlers/entities.mjs b/workers/request-handlers/entities.mjs index 8f6b9afed..4d86908ae 100644 --- a/workers/request-handlers/entities.mjs +++ b/workers/request-handlers/entities.mjs @@ -3218,10 +3218,11 @@ export async function handleAccountPositionHistory( } sql += " ORDER BY snapshot_date DESC LIMIT ?"; params.push(MAX_HISTORY_POINTS); - const rows = await d1All(env, sql, params); - const data = buildAccountPositionHistory(rows, ss58, netuid, { - window: label, - }); + const data = + (await tryPostgresTier(env, request, "METAGRAPH_NEURONS_SOURCE")) ?? + buildAccountPositionHistory(await d1All(env, sql, params), ss58, netuid, { + window: label, + }); return envelopeResponse( request, {