From 0243d8d1762d8761fc075d0ef1c5f6fe191b9592 Mon Sep 17 00:00:00 2001 From: shinorh247 Date: Tue, 27 Oct 2026 07:29:16 +0000 Subject: [PATCH] feat(keys): index SnapshotTaken events and serve GET /keys/:key_id/snapshots (#1112) --- prisma/schema/key-snapshot.prisma | 40 +++ .../migration.sql | 28 ++ src/constants/redis.constants.ts | 26 ++ .../indexer/indexer-pipeline.service.ts | 13 + .../key-snapshot-indexer.service.test.ts | 239 +++++++++++++ .../indexer/key-snapshot-indexer.service.ts | 327 ++++++++++++++++++ .../keys/__tests__/key-snapshots.test.ts | 233 +++++++++++++ src/modules/keys/key-snapshots.service.ts | 89 +++++ src/modules/keys/keys.routes.ts | 55 ++- 9 files changed, 1049 insertions(+), 1 deletion(-) create mode 100644 prisma/schema/key-snapshot.prisma create mode 100644 prisma/schema/migrations/20261027090000_add_key_snapshots/migration.sql create mode 100644 src/modules/indexer/key-snapshot-indexer.service.test.ts create mode 100644 src/modules/indexer/key-snapshot-indexer.service.ts create mode 100644 src/modules/keys/__tests__/key-snapshots.test.ts create mode 100644 src/modules/keys/key-snapshots.service.ts diff --git a/prisma/schema/key-snapshot.prisma b/prisma/schema/key-snapshot.prisma new file mode 100644 index 00000000..33e8bf04 --- /dev/null +++ b/prisma/schema/key-snapshot.prisma @@ -0,0 +1,40 @@ +// prisma/schema/key-snapshot.prisma +// +// Daily key analytics snapshots indexed from SnapshotTaken contract events +// (#1112). Powers GET /keys/:key_id/snapshots and the historical performance +// charts on the frontend. +// +// - One row per (creatorId, snapshotDate). The contract emits one snapshot a +// day per key; if it ever emits more than one for the same UTC day, the +// event with the highest (ledger, eventIndex) wins, so replays and +// out-of-order delivery always converge on the same row. +// - `creatorId` is intentionally NOT a foreign key, as for twap_snapshots: a +// snapshot can arrive before CREATOR_REGISTERED has been indexed. +// - Monetary values (price, volume, marketCap) are stroops in BIGINT. + +model KeyAnalyticsSnapshot { + id String @id @default(cuid()) + /** Key ID (CreatorProfile.id) the snapshot belongs to. */ + creatorId String + /** UTC calendar day the snapshot represents. */ + snapshotDate DateTime @db.Date + /** Key price in stroops at snapshot time. */ + price BigInt + /** Trading volume for the day in stroops. */ + volume BigInt + /** Number of distinct holders at snapshot time. */ + holderCount Int + /** Market cap in stroops at snapshot time. */ + marketCap BigInt + /** Exact time the contract took the snapshot. */ + takenAt DateTime + ledger Int + txHash String + eventIndex Int + indexedAt DateTime @default(now()) + updatedAt DateTime @updatedAt + + @@unique([creatorId, snapshotDate]) + @@index([ledger]) + @@map("key_snapshots") +} diff --git a/prisma/schema/migrations/20261027090000_add_key_snapshots/migration.sql b/prisma/schema/migrations/20261027090000_add_key_snapshots/migration.sql new file mode 100644 index 00000000..60d73f17 --- /dev/null +++ b/prisma/schema/migrations/20261027090000_add_key_snapshots/migration.sql @@ -0,0 +1,28 @@ +-- CreateTable +CREATE TABLE "key_snapshots" ( + "id" TEXT NOT NULL, + "creatorId" TEXT NOT NULL, + "snapshotDate" DATE NOT NULL, + "price" BIGINT NOT NULL, + "volume" BIGINT NOT NULL, + "holderCount" INTEGER NOT NULL, + "marketCap" BIGINT NOT NULL, + "takenAt" TIMESTAMP(3) NOT NULL, + "ledger" INTEGER NOT NULL, + "txHash" TEXT NOT NULL, + "eventIndex" INTEGER NOT NULL, + "indexedAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updatedAt" TIMESTAMP(3) NOT NULL, + + CONSTRAINT "key_snapshots_pkey" PRIMARY KEY ("id"), + CONSTRAINT "key_snapshots_price_check" CHECK ("price" >= 0), + CONSTRAINT "key_snapshots_volume_check" CHECK ("volume" >= 0), + CONSTRAINT "key_snapshots_holderCount_check" CHECK ("holderCount" >= 0), + CONSTRAINT "key_snapshots_marketCap_check" CHECK ("marketCap" >= 0) +); + +-- CreateIndex +CREATE UNIQUE INDEX "key_snapshots_creatorId_snapshotDate_key" ON "key_snapshots"("creatorId", "snapshotDate"); + +-- CreateIndex +CREATE INDEX "key_snapshots_ledger_idx" ON "key_snapshots"("ledger"); diff --git a/src/constants/redis.constants.ts b/src/constants/redis.constants.ts index 563d8849..cf658a78 100644 --- a/src/constants/redis.constants.ts +++ b/src/constants/redis.constants.ts @@ -125,3 +125,29 @@ export const keyTipContentCacheKey = (messageHash: string): string => export const keyTipsCachePattern = (keyId: string): string => `key:tips:${keyId}:*`; + +// ── Key analytics snapshots endpoint (#1112) ───────────────────── +// GET /keys/:key_id/snapshots?range=7d|30d|90d|all is served from the indexed +// key_snapshots table and cached for 5 minutes. The SnapshotTaken indexer +// drops every range for the key when a new snapshot lands (literal DELs). +export const KEY_SNAPSHOT_RANGES = ['7d', '30d', '90d', 'all'] as const; +export type KeySnapshotRange = (typeof KEY_SNAPSHOT_RANGES)[number]; + +/** Days covered by each bounded range, counting today. `all` is unbounded. */ +export const KEY_SNAPSHOT_RANGE_DAYS: Record< + Exclude, + number +> = { + '7d': 7, + '30d': 30, + '90d': 90, +}; + +export const KEY_SNAPSHOT_DEFAULT_RANGE: KeySnapshotRange = '30d'; + +export const KEY_SNAPSHOT_CACHE_TTL_SECONDS = 5 * 60; + +export const keySnapshotCacheKey = ( + keyId: string, + range: KeySnapshotRange +): string => `key-snapshots:${keyId}:${range}`; diff --git a/src/modules/indexer/indexer-pipeline.service.ts b/src/modules/indexer/indexer-pipeline.service.ts index cd8944fa..08ba546d 100644 --- a/src/modules/indexer/indexer-pipeline.service.ts +++ b/src/modules/indexer/indexer-pipeline.service.ts @@ -421,6 +421,19 @@ export async function processIndexerEvents( await processTwapUpdatedEvents(twapEvents); } + // SnapshotTaken (#1112). Awaited before the cursor moves so a failure + // redelivers the batch and key_snapshots never misses a day. + const snapshotEvents = events.filter( + event => + event.eventType === 'SNAPSHOT_TAKEN' || + event.eventType === 'SnapshotTaken' + ); + if (snapshotEvents.length > 0) { + const { processSnapshotTakenEvents } = + await import('./key-snapshot-indexer.service'); + await processSnapshotTakenEvents(snapshotEvents); + } + // DepthUpdated (#1152). Awaited before the cursor moves so a failure // redelivers the batch and key_depth never misses a level change. const depthEvents = events.filter( diff --git a/src/modules/indexer/key-snapshot-indexer.service.test.ts b/src/modules/indexer/key-snapshot-indexer.service.test.ts new file mode 100644 index 00000000..75177d87 --- /dev/null +++ b/src/modules/indexer/key-snapshot-indexer.service.test.ts @@ -0,0 +1,239 @@ +jest.mock('../../utils/prisma.utils', () => ({ + prisma: { + $executeRaw: jest.fn(), + }, +})); + +jest.mock('../../utils/redis.utils', () => ({ + cacheInvalidate: jest.fn().mockResolvedValue(undefined), +})); + +jest.mock('../../utils/indexer-dlq.utils', () => ({ + moveToDLQ: jest.fn().mockResolvedValue({}), +})); + +jest.mock('../../utils/logger.utils', () => ({ + logger: { + warn: jest.fn(), + info: jest.fn(), + debug: jest.fn(), + error: jest.fn(), + isLevelEnabled: jest.fn().mockReturnValue(false), + }, +})); + +import { + parseSnapshotTakenEvent, + processSnapshotTakenEvent, + processSnapshotTakenEvents, + toUtcDay, + KEY_SNAPSHOT_INDEXER_DLQ_JOB_TYPE, +} from './key-snapshot-indexer.service'; +import { prisma } from '../../utils/prisma.utils'; +import { cacheInvalidate } from '../../utils/redis.utils'; +import { moveToDLQ } from '../../utils/indexer-dlq.utils'; +import { IndexerChainEvent } from '../../utils/indexer-event-processor.utils'; + +const mockExec = (prisma as any).$executeRaw as jest.Mock; +const mockInvalidate = cacheInvalidate as jest.Mock; +const mockDlq = moveToDLQ as jest.Mock; + +const TAKEN_AT = '2026-10-26T23:59:00.000Z'; + +function snapshotEvent( + overrides: Record = {} +): IndexerChainEvent { + return { + txHash: 'tx-1', + eventIndex: 0, + eventType: 'SnapshotTaken', + ledger: 5000, + creatorId: 'key-1', + price: '12500000', + volume: '980000000', + holder_count: 42, + market_cap: '525000000', + timestamp: TAKEN_AT, + ...overrides, + }; +} + +beforeEach(() => { + jest.clearAllMocks(); + mockExec.mockResolvedValue(1); +}); + +describe('parseSnapshotTakenEvent', () => { + it('parses all four metrics from a snake_case event', () => { + const result = parseSnapshotTakenEvent(snapshotEvent()); + expect(result).toEqual({ + ok: true, + value: { + creatorId: 'key-1', + snapshotDate: new Date('2026-10-26T00:00:00.000Z'), + price: 12500000n, + volume: 980000000n, + holderCount: 42, + marketCap: 525000000n, + takenAt: new Date(TAKEN_AT), + ledger: 5000, + txHash: 'tx-1', + eventIndex: 0, + }, + }); + }); + + it('accepts camelCase fields and keyId', () => { + const result = parseSnapshotTakenEvent( + snapshotEvent({ + creatorId: undefined, + keyId: 'key-2', + holder_count: undefined, + holderCount: '7', + market_cap: undefined, + marketCap: 100n, + timestamp: undefined, + takenAt: TAKEN_AT, + }) + ); + expect(result.ok).toBe(true); + if (result.ok) { + expect(result.value.creatorId).toBe('key-2'); + expect(result.value.holderCount).toBe(7); + expect(result.value.marketCap).toBe(100n); + } + }); + + it('keeps values above Number.MAX_SAFE_INTEGER exact', () => { + const big = '9007199254740993'; + const result = parseSnapshotTakenEvent( + snapshotEvent({ price: big, market_cap: BigInt(big) }) + ); + expect(result.ok && result.value.price).toBe(BigInt(big)); + expect(result.ok && result.value.marketCap).toBe(BigInt(big)); + }); + + it('reads epoch seconds and epoch milliseconds timestamps', () => { + const secs = Math.floor(Date.parse(TAKEN_AT) / 1000); + const fromSecs = parseSnapshotTakenEvent( + snapshotEvent({ timestamp: secs }) + ); + const fromMs = parseSnapshotTakenEvent( + snapshotEvent({ timestamp: String(secs * 1000) }) + ); + expect(fromSecs.ok && fromSecs.value.takenAt.toISOString()).toBe( + TAKEN_AT + ); + expect(fromMs.ok && fromMs.value.takenAt.toISOString()).toBe(TAKEN_AT); + }); + + it.each([ + ['creatorId', { creatorId: '' }], + ['price', { price: undefined }], + ['price', { price: '-1' }], + ['volume', { volume: 1.5 }], + ['holder_count', { holder_count: -3 }], + ['holder_count', { holder_count: undefined }], + ['market_cap', { market_cap: 'abc' }], + ['timestamp', { timestamp: 'not-a-date' }], + ['ledger', { ledger: undefined }], + ['txHash/eventIndex', { txHash: '' }], + ])('rejects an invalid %s', (field, overrides) => { + const result = parseSnapshotTakenEvent(snapshotEvent(overrides)); + expect(result.ok).toBe(false); + if (!result.ok) expect(result.reason).toContain(field); + }); +}); + +describe('toUtcDay', () => { + it('truncates to midnight UTC', () => { + expect(toUtcDay(new Date('2026-10-27T23:30:00+01:00'))).toEqual( + new Date('2026-10-27T00:00:00.000Z') + ); + }); +}); + +describe('processSnapshotTakenEvent', () => { + it('upserts the snapshot and invalidates every range for the key', async () => { + await expect(processSnapshotTakenEvent(snapshotEvent())).resolves.toBe( + true + ); + expect(mockExec).toHaveBeenCalledTimes(1); + const [strings, ...values] = mockExec.mock.calls[0]; + const sql = (strings as string[]).join('?'); + expect(sql).toContain('ON CONFLICT ("creatorId", "snapshotDate")'); + expect(sql).toContain('< (EXCLUDED."ledger", EXCLUDED."eventIndex")'); + expect(values).toEqual( + expect.arrayContaining([ + 'key-1', + 12500000n, + 980000000n, + 42, + 525000000n, + 5000, + 'tx-1', + 0, + ]) + ); + expect(mockInvalidate).toHaveBeenCalledWith( + 'key-snapshots:key-1:7d', + 'key-snapshots:key-1:30d', + 'key-snapshots:key-1:90d', + 'key-snapshots:key-1:all' + ); + }); + + it('invalidates after the write, never before', async () => { + const order: string[] = []; + mockExec.mockImplementation(async () => { + order.push('write'); + return 1; + }); + mockInvalidate.mockImplementation(async () => { + order.push('invalidate'); + }); + await processSnapshotTakenEvent(snapshotEvent()); + expect(order).toEqual(['write', 'invalidate']); + }); + + it('returns false for a stale or duplicate event but still invalidates', async () => { + mockExec.mockResolvedValue(0); + await expect(processSnapshotTakenEvent(snapshotEvent())).resolves.toBe( + false + ); + expect(mockInvalidate).toHaveBeenCalledTimes(1); + }); + + it('sends malformed events to the DLQ without writing', async () => { + await expect( + processSnapshotTakenEvent(snapshotEvent({ price: undefined })) + ).resolves.toBe(false); + expect(mockExec).not.toHaveBeenCalled(); + expect(mockInvalidate).not.toHaveBeenCalled(); + expect(mockDlq).toHaveBeenCalledWith( + expect.objectContaining({ + jobType: KEY_SNAPSHOT_INDEXER_DLQ_JOB_TYPE, + failureReason: 'missing or invalid price', + }) + ); + }); + + it('rethrows database errors so the cursor does not advance', async () => { + mockExec.mockRejectedValue(new Error('db down')); + await expect(processSnapshotTakenEvent(snapshotEvent())).rejects.toThrow( + 'db down' + ); + expect(mockInvalidate).not.toHaveBeenCalled(); + }); +}); + +describe('processSnapshotTakenEvents', () => { + it('processes only SnapshotTaken events in a mixed batch', async () => { + await processSnapshotTakenEvents([ + snapshotEvent(), + snapshotEvent({ eventType: 'KEY_BOUGHT', txHash: 'tx-2' }), + snapshotEvent({ eventType: 'SNAPSHOT_TAKEN', txHash: 'tx-3' }), + ]); + expect(mockExec).toHaveBeenCalledTimes(2); + }); +}); diff --git a/src/modules/indexer/key-snapshot-indexer.service.ts b/src/modules/indexer/key-snapshot-indexer.service.ts new file mode 100644 index 00000000..7e2a88db --- /dev/null +++ b/src/modules/indexer/key-snapshot-indexer.service.ts @@ -0,0 +1,327 @@ +// src/modules/indexer/key-snapshot-indexer.service.ts +// +// Indexes SnapshotTaken contract events into the `key_snapshots` table (#1112). +// +// Each SnapshotTaken event carries a key's daily metrics: price, volume, +// holder_count and market_cap. They are stored as one row per +// (creatorId, UTC day) and served by GET /keys/:key_id/snapshots. +// +// Guarantees: +// - Idempotent and order-independent. The write is an upsert on +// (creatorId, snapshotDate) that only overwrites an existing row when the +// incoming event is newer by (ledger, eventIndex). Replays, overlapping +// ingestion windows and out-of-order delivery all converge on the same row. +// - Nothing is silently dropped. A well-formed event is always stored, even +// when its key is not registered yet (no FK on creatorId). A malformed event +// goes to the indexer DLQ so it is visible and replayable. +// - Failures propagate. Any database error rethrows so the pipeline +// dispatcher does NOT advance the ledger cursor past the batch. +// +// After every event, the key's cached snapshot responses (all ranges) are +// invalidated, including when the upsert was a no-op, to cover a crash that +// happened between the write and the invalidation. + +import { randomUUID } from 'crypto'; +import { prisma } from '../../utils/prisma.utils'; +import { logger } from '../../utils/logger.utils'; +import { + processIndexerChainEvents, + IndexerChainEvent, +} from '../../utils/indexer-event-processor.utils'; +import { moveToDLQ } from '../../utils/indexer-dlq.utils'; +import { cacheInvalidate } from '../../utils/redis.utils'; +import { + KEY_SNAPSHOT_RANGES, + keySnapshotCacheKey, +} from '../../constants/redis.constants'; + +/** Event discriminators. Both spellings are accepted, as for TWAPUpdated. */ +export const SNAPSHOT_TAKEN_EVENT_TYPES: ReadonlySet = new Set([ + 'SNAPSHOT_TAKEN', + 'SnapshotTaken', +]); + +export const KEY_SNAPSHOT_INDEXER_DLQ_JOB_TYPE = 'SNAPSHOT_TAKEN'; + +/** + * Decoded SnapshotTaken event as delivered by the indexing worker. Metric + * fields are accepted in camelCase or snake_case (the contract emits + * snake_case). Monetary values are stroops as string/bigint/number. + * The timestamp is an ISO string, epoch milliseconds, or epoch seconds. + */ +export interface SnapshotTakenEvent extends IndexerChainEvent { + creatorId?: string; + keyId?: string; + key_id?: string; + price?: string | number | bigint; + volume?: string | number | bigint; + holderCount?: number | string; + holder_count?: number | string; + marketCap?: string | number | bigint; + market_cap?: string | number | bigint; + timestamp?: string | number; + takenAt?: string | number; +} + +export function isSnapshotTakenEvent( + event: IndexerChainEvent +): event is SnapshotTakenEvent { + return SNAPSHOT_TAKEN_EVENT_TYPES.has(event.eventType); +} + +export interface ParsedKeySnapshot { + creatorId: string; + /** Midnight UTC of the day the snapshot belongs to. */ + snapshotDate: Date; + price: bigint; + volume: bigint; + holderCount: number; + marketCap: bigint; + takenAt: Date; + ledger: number; + txHash: string; + eventIndex: number; +} + +export type KeySnapshotParseResult = + | { ok: true; value: ParsedKeySnapshot } + | { ok: false; reason: string }; + +function parseTimestamp(value: unknown): Date | null { + if (value === undefined || value === null || value === '') return null; + let date: Date; + if (typeof value === 'number' || typeof value === 'bigint') { + const n = Number(value); + // Values below 1e12 are epoch seconds (Soroban timestamps); above, ms. + date = new Date(n < 1e12 ? n * 1000 : n); + } else if (typeof value === 'string' && /^\d+$/.test(value.trim())) { + const n = Number(value.trim()); + date = new Date(n < 1e12 ? n * 1000 : n); + } else { + date = new Date(String(value)); + } + return Number.isNaN(date.getTime()) ? null : date; +} + +function parseNonNegativeBigInt(value: unknown): bigint | null { + try { + if (typeof value === 'number' && !Number.isInteger(value)) return null; + if (typeof value === 'string' && value.trim() === '') return null; + if ( + typeof value !== 'string' && + typeof value !== 'number' && + typeof value !== 'bigint' + ) { + return null; + } + const parsed = BigInt(value); + return parsed >= 0n ? parsed : null; + } catch { + return null; + } +} + +function parseNonNegativeInt(value: unknown): number | null { + if (value === undefined || value === null || value === '') return null; + if (typeof value === 'bigint') { + return value >= 0n && value <= BigInt(Number.MAX_SAFE_INTEGER) + ? Number(value) + : null; + } + const n = Number(value); + return Number.isSafeInteger(n) && n >= 0 ? n : null; +} + +/** Midnight UTC of the calendar day `date` falls on. */ +export function toUtcDay(date: Date): Date { + return new Date( + Date.UTC(date.getUTCFullYear(), date.getUTCMonth(), date.getUTCDate()) + ); +} + +/** Pure validation of a raw event into a typed, persistable snapshot. */ +export function parseSnapshotTakenEvent( + event: IndexerChainEvent +): KeySnapshotParseResult { + const raw = event as Record; + + const rawKey = raw.creatorId ?? raw.keyId ?? raw.key_id; + if (typeof rawKey !== 'string' || rawKey.trim() === '') { + return { ok: false, reason: 'missing or invalid creatorId' }; + } + + const price = parseNonNegativeBigInt(raw.price); + if (price === null) { + return { ok: false, reason: 'missing or invalid price' }; + } + const volume = parseNonNegativeBigInt(raw.volume); + if (volume === null) { + return { ok: false, reason: 'missing or invalid volume' }; + } + const holderCount = parseNonNegativeInt(raw.holderCount ?? raw.holder_count); + if (holderCount === null) { + return { ok: false, reason: 'missing or invalid holder_count' }; + } + const marketCap = parseNonNegativeBigInt(raw.marketCap ?? raw.market_cap); + if (marketCap === null) { + return { ok: false, reason: 'missing or invalid market_cap' }; + } + + const takenAt = parseTimestamp(raw.takenAt ?? raw.timestamp); + if (takenAt === null) { + return { ok: false, reason: 'missing or invalid timestamp' }; + } + + const ledger = parseNonNegativeInt(raw.ledger); + if (ledger === null) { + return { ok: false, reason: 'missing or invalid ledger' }; + } + const eventIndex = Number(event.eventIndex); + if ( + typeof event.txHash !== 'string' || + event.txHash === '' || + !Number.isInteger(eventIndex) || + eventIndex < 0 + ) { + return { ok: false, reason: 'missing or invalid txHash/eventIndex' }; + } + + return { + ok: true, + value: { + creatorId: rawKey.trim(), + snapshotDate: toUtcDay(takenAt), + price, + volume, + holderCount, + marketCap, + takenAt, + ledger, + txHash: event.txHash, + eventIndex, + }, + }; +} + +/** Drop every cached snapshot response for a key (all ranges, literal DELs). */ +export async function invalidateKeySnapshotCache( + creatorId: string +): Promise { + await cacheInvalidate( + ...KEY_SNAPSHOT_RANGES.map(range => keySnapshotCacheKey(creatorId, range)) + ); +} + +/** + * Upserts one snapshot. Inserts a new day, or overwrites an existing day only + * when this event is newer by (ledger, eventIndex). Returns the number of rows + * written (0 when an equal or newer event for that day is already stored). + */ +export async function upsertKeySnapshot( + snapshot: ParsedKeySnapshot +): Promise { + return prisma.$executeRaw` + INSERT INTO "key_snapshots" ( + "id", "creatorId", "snapshotDate", "price", "volume", "holderCount", + "marketCap", "takenAt", "ledger", "txHash", "eventIndex", "updatedAt" + ) VALUES ( + ${randomUUID()}, ${snapshot.creatorId}, ${snapshot.snapshotDate}::date, + ${snapshot.price}, ${snapshot.volume}, ${snapshot.holderCount}, + ${snapshot.marketCap}, ${snapshot.takenAt}, ${snapshot.ledger}, + ${snapshot.txHash}, ${snapshot.eventIndex}, NOW() + ) + ON CONFLICT ("creatorId", "snapshotDate") DO UPDATE SET + "price" = EXCLUDED."price", + "volume" = EXCLUDED."volume", + "holderCount" = EXCLUDED."holderCount", + "marketCap" = EXCLUDED."marketCap", + "takenAt" = EXCLUDED."takenAt", + "ledger" = EXCLUDED."ledger", + "txHash" = EXCLUDED."txHash", + "eventIndex" = EXCLUDED."eventIndex", + "updatedAt" = NOW() + WHERE ("key_snapshots"."ledger", "key_snapshots"."eventIndex") + < (EXCLUDED."ledger", EXCLUDED."eventIndex") + `; +} + +/** + * Processes one SnapshotTaken event. + * + * Returns `true` when a row was inserted or updated, `false` when the event + * was stale/duplicate or malformed (and sent to the DLQ). + * Database errors are rethrown on purpose; see the header comment. + */ +export async function processSnapshotTakenEvent( + event: IndexerChainEvent +): Promise { + const eventId = `${event.txHash}:${event.eventIndex}`; + const parsed = parseSnapshotTakenEvent(event); + + if (!parsed.ok) { + logger.warn( + { + type: 'key_snapshot_event_rejected', + eventId, + reason: parsed.reason, + }, + 'Malformed SnapshotTaken event moved to DLQ' + ); + await moveToDLQ({ + jobType: KEY_SNAPSHOT_INDEXER_DLQ_JOB_TYPE, + payload: JSON.parse( + JSON.stringify(event, (_k, v) => + typeof v === 'bigint' ? v.toString() : v + ) + ), + retryCount: 0, + failureReason: parsed.reason, + }); + return false; + } + + const snapshot = parsed.value; + const written = await upsertKeySnapshot(snapshot); + + // Invalidate only after the row is durable so a reader that repopulates + // the cache immediately afterwards sees the new snapshot. + await invalidateKeySnapshotCache(snapshot.creatorId); + + if (written > 0) { + logger.info( + { + type: 'key_snapshot_indexed', + eventId, + creator_id: snapshot.creatorId, + snapshot_date: snapshot.snapshotDate.toISOString().slice(0, 10), + ledger: snapshot.ledger, + }, + 'Key analytics snapshot indexed' + ); + } else { + logger.debug( + { + type: 'key_snapshot_event_stale_skipped', + eventId, + creator_id: snapshot.creatorId, + }, + 'SnapshotTaken event not newer than stored snapshot; skipping' + ); + } + return written > 0; +} + +/** + * Processes a batch of SnapshotTaken events. Other events are ignored. + * Does not advance the ledger cursor; the pipeline dispatcher does that. + */ +export async function processSnapshotTakenEvents( + events: IndexerChainEvent[] +): Promise { + await processIndexerChainEvents(events, async event => { + if (!isSnapshotTakenEvent(event)) { + return; + } + await processSnapshotTakenEvent(event); + }); +} diff --git a/src/modules/keys/__tests__/key-snapshots.test.ts b/src/modules/keys/__tests__/key-snapshots.test.ts new file mode 100644 index 00000000..a37fe992 --- /dev/null +++ b/src/modules/keys/__tests__/key-snapshots.test.ts @@ -0,0 +1,233 @@ +// GET /api/v1/keys/:keyId/snapshots (#1112): route, read service and cache +// invalidation wired end to end against an in-memory Redis and mocked Prisma. + +const redisStore = new Map(); + +jest.mock('../../../utils/redis.utils', () => ({ + cacheGetJson: jest.fn(async (key: string) => { + const v = redisStore.get(key); + return v ? JSON.parse(v) : null; + }), + cacheSetJson: jest.fn(async (key: string, value: unknown) => { + redisStore.set(key, JSON.stringify(value)); + }), + cacheInvalidate: jest.fn(async (...keys: string[]) => { + for (const k of keys) redisStore.delete(k); + }), +})); + +jest.mock('../../../utils/prisma.utils', () => ({ + prisma: { + creatorProfile: { findFirst: jest.fn(), findUnique: jest.fn() }, + keyAnalyticsSnapshot: { findMany: jest.fn() }, + $executeRaw: jest.fn(), + }, +})); + +jest.mock('../../../utils/indexer-dlq.utils', () => ({ + moveToDLQ: jest.fn().mockResolvedValue({}), +})); + +jest.mock('../../../utils/logger.utils', () => ({ + logger: { + info: jest.fn(), + warn: jest.fn(), + error: jest.fn(), + debug: jest.fn(), + isLevelEnabled: jest.fn().mockReturnValue(false), + }, +})); + +import request from 'supertest'; +import express from 'express'; +import keysRouter from '../keys.routes'; +import { prisma } from '../../../utils/prisma.utils'; +import { cacheSetJson } from '../../../utils/redis.utils'; +import { processSnapshotTakenEvent } from '../../indexer/key-snapshot-indexer.service'; +import { rangeStartDate } from '../key-snapshots.service'; + +const app = express(); +app.use(express.json()); +app.use('/api/v1/keys', keysRouter); + +const db = prisma as any; +const KEYS: Record = { 'key-1': 'key-1', alice: 'key-1' }; + +function row(date: string, overrides: Record = {}) { + return { + snapshotDate: new Date(`${date}T00:00:00.000Z`), + price: 12500000n, + volume: 980000000n, + holderCount: 42, + marketCap: 525000000n, + ...overrides, + }; +} + +beforeEach(() => { + jest.clearAllMocks(); + redisStore.clear(); + db.creatorProfile.findFirst.mockImplementation(async (args: any) => { + const term = args.where.OR[0].id; + const id = KEYS[term]; + return id ? { id } : null; + }); + db.keyAnalyticsSnapshot.findMany.mockResolvedValue([]); + db.$executeRaw.mockResolvedValue(1); +}); + +describe('rangeStartDate', () => { + const now = new Date('2026-10-27T08:00:00.000Z'); + + it.each([ + ['7d', '2026-10-21'], + ['30d', '2026-09-28'], + ['90d', '2026-07-30'], + ] as const)('%s starts %s (today counted)', (range, expected) => { + expect(rangeStartDate(range, now)?.toISOString().slice(0, 10)).toBe( + expected + ); + }); + + it('all is unbounded', () => { + expect(rangeStartDate('all', now)).toBeNull(); + }); +}); + +describe('GET /api/v1/keys/:keyId/snapshots', () => { + it('returns daily snapshots with all four metrics, oldest first', async () => { + db.keyAnalyticsSnapshot.findMany.mockResolvedValue([ + row('2026-10-25'), + row('2026-10-26', { price: 13000000n, holderCount: 45 }), + ]); + + const res = await request(app).get( + '/api/v1/keys/key-1/snapshots?range=7d' + ); + + expect(res.status).toBe(200); + expect(res.body.data).toEqual([ + { + date: '2026-10-25', + price: '12500000', + volume: '980000000', + holder_count: 42, + market_cap: '525000000', + }, + { + date: '2026-10-26', + price: '13000000', + volume: '980000000', + holder_count: 45, + market_cap: '525000000', + }, + ]); + const args = db.keyAnalyticsSnapshot.findMany.mock.calls[0][0]; + expect(args.orderBy).toEqual({ snapshotDate: 'asc' }); + }); + + it.each(['7d', '30d', '90d'])( + 'filters %s from the current date backward', + async range => { + await request(app).get(`/api/v1/keys/key-1/snapshots?range=${range}`); + const args = db.keyAnalyticsSnapshot.findMany.mock.calls[0][0]; + expect(args.where.creatorId).toBe('key-1'); + expect(args.where.snapshotDate.gte).toBeInstanceOf(Date); + } + ); + + it('range=all applies no date filter', async () => { + await request(app).get('/api/v1/keys/key-1/snapshots?range=all'); + const args = db.keyAnalyticsSnapshot.findMany.mock.calls[0][0]; + expect(args.where).toEqual({ creatorId: 'key-1' }); + }); + + it('defaults to 30d when range is omitted', async () => { + await request(app).get('/api/v1/keys/key-1/snapshots'); + expect(cacheSetJson).toHaveBeenCalledWith( + 'key-snapshots:key-1:30d', + [], + 300 + ); + }); + + it('returns an empty array for a key with no snapshots yet', async () => { + const res = await request(app).get( + '/api/v1/keys/key-1/snapshots?range=all' + ); + expect(res.status).toBe(200); + expect(res.body.data).toEqual([]); + }); + + it('resolves a handle to the canonical key id', async () => { + await request(app).get('/api/v1/keys/alice/snapshots?range=7d'); + const args = db.keyAnalyticsSnapshot.findMany.mock.calls[0][0]; + expect(args.where.creatorId).toBe('key-1'); + }); + + it('returns 404 for an unknown key', async () => { + const res = await request(app).get('/api/v1/keys/nope/snapshots'); + expect(res.status).toBe(404); + }); + + it('returns 422 for an unsupported range', async () => { + const res = await request(app).get( + '/api/v1/keys/key-1/snapshots?range=1y' + ); + expect(res.status).toBe(422); + expect(db.keyAnalyticsSnapshot.findMany).not.toHaveBeenCalled(); + }); + + it('caches for 5 minutes and serves repeat requests from cache under 200ms', async () => { + db.keyAnalyticsSnapshot.findMany.mockResolvedValue([row('2026-10-26')]); + await request(app).get('/api/v1/keys/key-1/snapshots?range=7d'); + expect(cacheSetJson).toHaveBeenCalledWith( + 'key-snapshots:key-1:7d', + expect.any(Array), + 300 + ); + + const started = Date.now(); + const res = await request(app).get( + '/api/v1/keys/key-1/snapshots?range=7d' + ); + const elapsed = Date.now() - started; + + expect(res.status).toBe(200); + expect(res.body.data).toHaveLength(1); + expect(db.keyAnalyticsSnapshot.findMany).toHaveBeenCalledTimes(1); + expect(elapsed).toBeLessThan(200); + }); + + it('a new SnapshotTaken event invalidates every cached range for the key', async () => { + db.keyAnalyticsSnapshot.findMany.mockResolvedValue([row('2026-10-25')]); + for (const range of ['7d', '30d', '90d', 'all']) { + await request(app).get(`/api/v1/keys/key-1/snapshots?range=${range}`); + } + expect(redisStore.size).toBe(4); + + await processSnapshotTakenEvent({ + txHash: 'tx-9', + eventIndex: 0, + eventType: 'SnapshotTaken', + ledger: 9000, + creatorId: 'key-1', + price: '14000000', + volume: '1000', + holder_count: 50, + market_cap: '700000000', + timestamp: '2026-10-26T23:59:00.000Z', + }); + expect(redisStore.size).toBe(0); + + db.keyAnalyticsSnapshot.findMany.mockResolvedValue([ + row('2026-10-25'), + row('2026-10-26', { price: 14000000n }), + ]); + const res = await request(app).get( + '/api/v1/keys/key-1/snapshots?range=7d' + ); + expect(res.body.data).toHaveLength(2); + expect(res.body.data[1].price).toBe('14000000'); + }); +}); diff --git a/src/modules/keys/key-snapshots.service.ts b/src/modules/keys/key-snapshots.service.ts new file mode 100644 index 00000000..6512bd20 --- /dev/null +++ b/src/modules/keys/key-snapshots.service.ts @@ -0,0 +1,89 @@ +// src/modules/keys/key-snapshots.service.ts +// +// Read side of GET /keys/:key_id/snapshots (#1112). Serves daily analytics +// snapshots from `key_snapshots`, oldest first (chart order), fronted by a +// 5-minute cache that the SnapshotTaken indexer invalidates per key. + +import { prisma } from '../../utils/prisma.utils'; +import { cacheGetJson, cacheSetJson } from '../../utils/redis.utils'; +import { + KEY_SNAPSHOT_CACHE_TTL_SECONDS, + KEY_SNAPSHOT_RANGE_DAYS, + keySnapshotCacheKey, + type KeySnapshotRange, +} from '../../constants/redis.constants'; + +export interface KeySnapshotDto { + /** UTC day, YYYY-MM-DD. */ + date: string; + /** Stroops, as strings to avoid JS precision loss. */ + price: string; + volume: string; + holder_count: number; + market_cap: string; +} + +/** + * First UTC day included in `range`, counting today as day one + * (7d = today and the six days before). `null` for `all`. + */ +export function rangeStartDate( + range: KeySnapshotRange, + now: Date = new Date() +): Date | null { + if (range === 'all') return null; + const days = KEY_SNAPSHOT_RANGE_DAYS[range]; + return new Date( + Date.UTC( + now.getUTCFullYear(), + now.getUTCMonth(), + now.getUTCDate() - (days - 1) + ) + ); +} + +/** + * Daily snapshots for a key within `range`, oldest first. Returns an empty + * array for a key with no snapshots yet; that result is cached too. + * + * `creatorId` must already be the canonical CreatorProfile id (not a + * handle) so the cache key matches what the indexer invalidates. + */ +export async function getKeySnapshots( + creatorId: string, + range: KeySnapshotRange, + now: Date = new Date() +): Promise { + const cacheKey = keySnapshotCacheKey(creatorId, range); + const cached = await cacheGetJson(cacheKey); + if (cached !== null) { + return cached; + } + + const start = rangeStartDate(range, now); + const rows = await prisma.keyAnalyticsSnapshot.findMany({ + where: { + creatorId, + ...(start ? { snapshotDate: { gte: start } } : {}), + }, + orderBy: { snapshotDate: 'asc' }, + select: { + snapshotDate: true, + price: true, + volume: true, + holderCount: true, + marketCap: true, + }, + }); + + const result: KeySnapshotDto[] = rows.map(row => ({ + date: row.snapshotDate.toISOString().slice(0, 10), + price: row.price.toString(), + volume: row.volume.toString(), + holder_count: row.holderCount, + market_cap: row.marketCap.toString(), + })); + + await cacheSetJson(cacheKey, result, KEY_SNAPSHOT_CACHE_TTL_SECONDS); + return result; +} diff --git a/src/modules/keys/keys.routes.ts b/src/modules/keys/keys.routes.ts index cd7db29b..a8032768 100644 --- a/src/modules/keys/keys.routes.ts +++ b/src/modules/keys/keys.routes.ts @@ -32,7 +32,12 @@ import { getKeyMetadata, KeyMetadataNotFoundError, } from './key-metadata-sync.service'; -import { TWAP_WINDOWS } from '../../constants/redis.constants'; +import { + TWAP_WINDOWS, + KEY_SNAPSHOT_RANGES, + KEY_SNAPSHOT_DEFAULT_RANGE, +} from '../../constants/redis.constants'; +import { getKeySnapshots } from './key-snapshots.service'; import { cacheControl } from '../../middlewares/cache-control.middleware'; import { envConfig } from '../../config'; import { getKeyProposals, getProposalForVoting } from './key-proposals.service'; @@ -1050,6 +1055,54 @@ router.get('/:keyId/twap', async (req, res, next) => { } }); +// ── GET /:keyId/snapshots (#1112) ───────────────────────────── +// Daily analytics snapshots (price, volume, holder_count, market_cap) from +// the indexed SnapshotTaken events (key_snapshots table), oldest first. +// range=7d|30d|90d|all (default 30d) counts back from today, UTC. +// 5-minute cache, invalidated per key by the indexer. A key with no +// snapshots yet returns an empty array; an unknown key returns 404. + +const keySnapshotsQuerySchema = z.object({ + range: z + .enum(KEY_SNAPSHOT_RANGES, { + errorMap: () => ({ + message: 'Invalid range param. Must be 7d, 30d, 90d, or all', + }), + }) + .default(KEY_SNAPSHOT_DEFAULT_RANGE), +}); + +router.get('/:keyId/snapshots', async (req, res, next) => { + const keyId = String(req.params.keyId); + const parsed = keySnapshotsQuerySchema.safeParse(req.query); + if (!parsed.success) { + sendError( + res, + 422, + ErrorCode.UNPROCESSABLE_ENTITY, + 'Invalid range param. Must be 7d, 30d, 90d, or all', + zodIssuesToDetails(parsed.error.issues) + ); + return; + } + + try { + const creator = await prisma.creatorProfile.findFirst({ + where: { OR: [{ id: keyId }, { handle: keyId }] }, + select: { id: true }, + }); + if (!creator) { + sendNotFound(res, 'Key'); + return; + } + + const snapshots = await getKeySnapshots(creator.id, parsed.data.range); + sendSuccess(res, snapshots); + } catch (error) { + next(error); + } +}); + // ── GET /:keyId/depth and /:keyId/depth/spread (#1152) ──────── // Order book depth from the indexed DepthUpdated events (key_depth table), // 5s cache invalidated per key by the indexer. Keys with no open orders