feat(keys): index SnapshotTaken events and serve GET /keys/:key_id/sn…

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")
+}

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");

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<KeySnapshotRange, 'all'>,
+   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}`;

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(

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<string, unknown> = {}
+): 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);
+   });
+});