Implemented key leaderboard indexer and rankings REST API

docs/api-endpoints.md

@@ -740,7 +740,9 @@ Cached for 60 seconds.
    "success": true,
    "data": {
       "total_balance": "1200000000",
-      "per_source_breakdown": [{ "source": "trading", "balance": "1500000000" }],
+      "per_source_breakdown": [
+         { "source": "trading", "balance": "1500000000" }
+      ],
       "last_updated": "2026-10-21T10:00:00.000Z"
    }
 }
@@ -766,6 +768,26 @@ Paginated `FeeReceived` history (`source`, `amount`, `tx_hash`, `ledger`,
 
 ## Keys Endpoints
 
+### GET /keys/:keyId/leaderboard
+
+Returns wallets ranked by the selected `LeaderboardUpdated` metric. Results
+are cached for 60 seconds and the cache is invalidated when a new event is
+indexed for the key. Keys without indexed leaderboard rows return an empty
+`items` array.
+
+- **Auth:** None
+- **Query Params:** `metric` (`volume`, `pnl`, or `holding_days`, required),
+  `limit` (integer, default `20`, maximum `100`)
+- **Response:** `{ "success": true, "data": { "items": [{ "rank": 1, "wallet": "...", "score": "..." }] } }`
+
+### GET /keys/:keyId/leaderboard/:wallet
+
+Returns the wallet's ordinal rank and exact score for each leaderboard metric.
+Each metric is `null` when the wallet has no indexed entry for the key.
+
+- **Auth:** None
+- **Response:** `{ "success": true, "data": { "keyId": "...", "wallet": "...", "metrics": { "volume": { "rank": 1, "score": "..." }, "pnl": null, "holding_days": null } } }`
+
 ### GET /keys/:keyId/price
 
 Returns the current bonding-curve buy and sell price for a given quantity of

docs/api-inventory.md

@@ -194,6 +194,16 @@ Early-access whitelist status check and management endpoints (#1084).
 | `POST`   | `/keys/:keyId/whitelist`          | Bulk add wallets to key whitelist (creator JWT required).                         |
 | `DELETE` | `/keys/:keyId/whitelist/:wallet`  | Remove a wallet from key whitelist (creator JWT required).                        |
 
+## Key Leaderboard
+
+Wallet rankings are indexed from `LeaderboardUpdated` contract events and
+cached for 60 seconds. The cache is invalidated when a new event is indexed.
+
+| Method | Path                                       | Description                                            |
+| :----- | :----------------------------------------- | :----------------------------------------------------- |
+| `GET`  | `/keys/:key_id/leaderboard?metric=&limit=` | Ranked wallets for `volume`, `pnl`, or `holding_days`. |
+| `GET`  | `/keys/:key_id/leaderboard/:wallet`        | Wallet rank and score for all three metrics.           |
+
 ---
 
 ## Root and Miscellaneous

prisma/schema/key-leaderboard.prisma

@@ -0,0 +1,22 @@
+// Current per-wallet key leaderboard metrics, indexed from LeaderboardUpdated
+// contract events. The ledger/event index watermark makes replays and
+// out-of-order delivery converge on the latest snapshot.
+model KeyLeaderboard {
+  id          String   @id @default(cuid())
+  /// Key ID is not a foreign key so events preceding key registration are retained.
+  creatorId   String
+  wallet      String
+  volume      BigInt
+  pnl         BigInt
+  holdingDays BigInt
+  ledger      Int
+  eventIndex  Int
+  txHash      String
+  updatedAt   DateTime @updatedAt
+
+  @@unique([creatorId, wallet])
+  @@index([creatorId, volume(sort: Desc), wallet], map: "key_leaderboards_creatorId_volume_idx")
+  @@index([creatorId, pnl(sort: Desc), wallet], map: "key_leaderboards_creatorId_pnl_idx")
+  @@index([creatorId, holdingDays(sort: Desc), wallet], map: "key_leaderboards_creatorId_holdingDays_idx")
+  @@map("key_leaderboards")
+}

prisma/schema/migrations/20261022122500_add_key_leaderboards/migration.sql

@@ -0,0 +1,33 @@
+-- CreateTable
+CREATE TABLE "key_leaderboards" (
+    "id" TEXT NOT NULL,
+    "creatorId" TEXT NOT NULL,
+    "wallet" TEXT NOT NULL,
+    "volume" BIGINT NOT NULL,
+    "pnl" BIGINT NOT NULL,
+    "holdingDays" BIGINT NOT NULL,
+    "ledger" INTEGER NOT NULL,
+    "eventIndex" INTEGER NOT NULL,
+    "txHash" TEXT NOT NULL,
+    "updatedAt" TIMESTAMP(3) NOT NULL,
+
+    CONSTRAINT "key_leaderboards_pkey" PRIMARY KEY ("id"),
+    CONSTRAINT "key_leaderboards_volume_check" CHECK ("volume" >= 0),
+    CONSTRAINT "key_leaderboards_holding_days_check" CHECK ("holdingDays" >= 0)
+);
+
+-- CreateIndex
+CREATE UNIQUE INDEX "key_leaderboards_creatorId_wallet_key"
+ON "key_leaderboards"("creatorId", "wallet");
+
+-- CreateIndex
+CREATE INDEX "key_leaderboards_creatorId_volume_idx"
+ON "key_leaderboards"("creatorId", "volume" DESC, "wallet");
+
+-- CreateIndex
+CREATE INDEX "key_leaderboards_creatorId_pnl_idx"
+ON "key_leaderboards"("creatorId", "pnl" DESC, "wallet");
+
+-- CreateIndex
+CREATE INDEX "key_leaderboards_creatorId_holdingDays_idx"
+ON "key_leaderboards"("creatorId", "holdingDays" DESC, "wallet");

src/modules/indexer/indexer-pipeline.service.ts

@@ -423,6 +423,17 @@ export async function processIndexerEvents(
       await processDepthUpdatedEvents(depthEvents);
    }
 
+   // LeaderboardUpdated events. Await before advancing the shared
+   // cursor so a failed write causes the event batch to be redelivered.
+   const leaderboardEvents = events.filter(event =>
+      ['LEADERBOARD_UPDATED', 'LeaderboardUpdated'].includes(event.eventType)
+   );
+   if (leaderboardEvents.length > 0) {
+      const { processLeaderboardUpdatedEvents } =
+         await import('./key-leaderboard-indexer.service');
+      await processLeaderboardUpdatedEvents(leaderboardEvents);
+   }
+
    const whitelistEvents = events.filter(
       event =>
          event.eventType === 'WalletWhitelisted' ||

src/modules/indexer/key-leaderboard-indexer.service.test.ts

@@ -0,0 +1,190 @@
+jest.mock('../../utils/prisma.utils', () => ({
+   prisma: {
+      keyLeaderboard: {
+         updateMany: jest.fn(),
+         create: 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(),
+      debug: jest.fn(),
+      info: jest.fn(),
+      error: jest.fn(),
+   },
+}));
+
+import { prisma } from '../../utils/prisma.utils';
+import { cacheInvalidate } from '../../utils/redis.utils';
+import { moveToDLQ } from '../../utils/indexer-dlq.utils';
+import {
+   parseLeaderboardUpdatedEvent,
+   processLeaderboardUpdatedEvent,
+   processLeaderboardUpdatedEvents,
+   KEY_LEADERBOARD_INDEXER_DLQ_JOB_TYPE,
+} from './key-leaderboard-indexer.service';
+import { IndexerChainEvent } from '../../utils/indexer-event-processor.utils';
+
+const mockTable = (
+   prisma as unknown as {
+      keyLeaderboard: { updateMany: jest.Mock; create: jest.Mock };
+   }
+).keyLeaderboard;
+const mockInvalidate = cacheInvalidate as jest.Mock;
+const mockDlq = moveToDLQ as jest.Mock;
+
+function leaderboardEvent(
+   overrides: Record<string, unknown> = {}
+): IndexerChainEvent {
+   return {
+      txHash: 'tx-1',
+      eventIndex: 0,
+      eventType: 'LeaderboardUpdated',
+      ledger: 100,
+      creatorId: 'key-1',
+      wallet: 'GABC',
+      volume: '100000',
+      pnl: '-250',
+      holdingDays: '12',
+      ...overrides,
+   };
+}
+
+beforeEach(() => {
+   jest.clearAllMocks();
+   mockTable.updateMany.mockResolvedValue({ count: 0 });
+   mockTable.create.mockResolvedValue({});
+});
+
+describe('parseLeaderboardUpdatedEvent', () => {
+   it('parses exact integer scores and event aliases', () => {
+      expect(
+         parseLeaderboardUpdatedEvent(
+            leaderboardEvent({
+               eventType: 'LEADERBOARD_UPDATED',
+               key_id: 'key-2',
+               creatorId: undefined,
+               wallet_address: 'GWALLET',
+               wallet: undefined,
+               holding_days: '25',
+               holdingDays: undefined,
+               volume: '9007199254740993',
+            })
+         )
+      ).toEqual({
+         ok: true,
+         value: {
+            creatorId: 'key-2',
+            wallet: 'GWALLET',
+            volume: 9007199254740993n,
+            pnl: -250n,
+            holdingDays: 25n,
+            ledger: 100,
+            eventIndex: 0,
+            txHash: 'tx-1',
+         },
+      });
+   });
+
+   it.each([
+      ['missing key', { creatorId: '' }],
+      ['missing wallet', { wallet: '' }],
+      ['negative volume', { volume: '-1' }],
+      ['invalid pnl', { pnl: 'unknown' }],
+      ['negative holding days', { holdingDays: '-1' }],
+      ['missing ledger', { ledger: undefined }],
+      ['missing event identity', { txHash: '' }],
+   ])('rejects %s', (_name, overrides) => {
+      expect(parseLeaderboardUpdatedEvent(leaderboardEvent(overrides)).ok).toBe(
+         false
+      );
+   });
+});
+
+describe('processLeaderboardUpdatedEvent', () => {
+   it('upserts a new snapshot and invalidates the key cache', async () => {
+      await expect(
+         processLeaderboardUpdatedEvent(leaderboardEvent())
+      ).resolves.toBe(true);
+
+      expect(mockTable.create).toHaveBeenCalledWith({
+         data: {
+            creatorId: 'key-1',
+            wallet: 'GABC',
+            volume: 100000n,
+            pnl: -250n,
+            holdingDays: 12n,
+            ledger: 100,
+            eventIndex: 0,
+            txHash: 'tx-1',
+         },
+      });
+      expect(mockInvalidate).toHaveBeenCalledWith('key:leaderboard:key-1:*');
+   });
+
+   it('overwrites only with a newer on-chain position', async () => {
+      mockTable.updateMany.mockResolvedValueOnce({ count: 1 });
+
+      await expect(
+         processLeaderboardUpdatedEvent(leaderboardEvent({ ledger: 101 }))
+      ).resolves.toBe(true);
+
+      expect(mockTable.updateMany).toHaveBeenCalledWith(
+         expect.objectContaining({
+            where: {
+               creatorId: 'key-1',
+               wallet: 'GABC',
+               OR: [
+                  { ledger: { lt: 101 } },
+                  { ledger: 101, eventIndex: { lt: 0 } },
+               ],
+            },
+         })
+      );
+      expect(mockTable.create).not.toHaveBeenCalled();
+   });
+
+   it('moves malformed events to the DLQ', async () => {
+      await processLeaderboardUpdatedEvent(
+         leaderboardEvent({ holdingDays: '-2' })
+      );
+
+      expect(mockDlq).toHaveBeenCalledWith(
+         expect.objectContaining({
+            jobType: KEY_LEADERBOARD_INDEXER_DLQ_JOB_TYPE,
+            retryCount: 0,
+         })
+      );
+      expect(mockTable.create).not.toHaveBeenCalled();
+      expect(mockInvalidate).not.toHaveBeenCalled();
+   });
+
+   it('propagates persistence errors so the pipeline cursor stays put', async () => {
+      mockTable.create.mockRejectedValueOnce(new Error('database unavailable'));
+
+      await expect(
+         processLeaderboardUpdatedEvent(leaderboardEvent())
+      ).rejects.toThrow('database unavailable');
+      expect(mockInvalidate).not.toHaveBeenCalled();
+   });
+
+   it('processes supported event spellings and ignores unrelated events', async () => {
+      await processLeaderboardUpdatedEvents([
+         leaderboardEvent({ txHash: 'first' }),
+         leaderboardEvent({ txHash: 'ignored', eventType: 'KEY_BOUGHT' }),
+         leaderboardEvent({
+            txHash: 'second',
+            eventType: 'LEADERBOARD_UPDATED',
+         }),
+      ]);
+
+      expect(mockTable.create).toHaveBeenCalledTimes(2);
+   });
+});