Implemented creator revenue share indexer and distribution history REST API

docs/api-endpoints.md

@@ -192,6 +192,27 @@ display/bio/avatar fields plus a `null` `source`. The avatar is resolved from
 its IPFS CID via `PINATA_GATEWAY_URL`. See
 [social-metadata-sync.md](./social-metadata-sync.md) for the sync details.
 
+## Creator Revenue Distribution Endpoints
+
+### GET /keys/:keyId/revenue-distributions
+
+Return indexed revenue distributions for a creator key, newest first.
+
+- **Auth:** None
+- **Query Params:** `limit` (1-100, default 50), `cursor` (distribution cursor)
+- **Response:** Each entry includes `distributionId`, `distributedAt`, `amount`,
+  and `holderCount`, along with cursor pagination metadata.
+- **Cache:** 60 seconds; invalidated when a new distribution is indexed.
+
+### GET /keys/:keyId/revenue-distributions/:wallet
+
+Return the wallet's claimable and claimed amounts for each indexed distribution.
+
+- **Auth:** None
+- **Response:** Distribution entries include `claimableAmount` and
+  `claimedAmount` in addition to the distribution date, amount, and holder count.
+- **Cache:** 60 seconds; invalidated when a distribution or claim is indexed.
+
 ---
 
 ## Creator Profile Endpoints

prisma/schema/migrations/20261021000200_add_creator_revenue_distributions/migration.sql

@@ -0,0 +1,65 @@
+CREATE TABLE "revenue_distributions" (
+    "id" TEXT NOT NULL,
+    "keyId" TEXT NOT NULL,
+    "chainDistributionId" TEXT,
+    "amount" DECIMAL(20,7) NOT NULL,
+    "holderCount" INTEGER NOT NULL,
+    "ledger" INTEGER NOT NULL,
+    "txHash" TEXT NOT NULL,
+    "eventIndex" INTEGER NOT NULL DEFAULT 0,
+    "distributedAt" TIMESTAMP(3) NOT NULL,
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+
+    CONSTRAINT "revenue_distributions_pkey" PRIMARY KEY ("id")
+);
+
+CREATE UNIQUE INDEX "revenue_distributions_ledger_txHash_eventIndex_key"
+    ON "revenue_distributions"("ledger", "txHash", "eventIndex");
+CREATE UNIQUE INDEX "revenue_distributions_keyId_chainDistributionId_key"
+    ON "revenue_distributions"("keyId", "chainDistributionId");
+CREATE INDEX "revenue_distributions_keyId_distributedAt_id_idx"
+    ON "revenue_distributions"("keyId", "distributedAt" DESC, "id" DESC);
+
+CREATE TABLE "revenue_distribution_shares" (
+    "id" TEXT NOT NULL,
+    "distributionId" TEXT NOT NULL,
+    "wallet" TEXT NOT NULL,
+    "amount" DECIMAL(20,7) NOT NULL,
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+
+    CONSTRAINT "revenue_distribution_shares_pkey" PRIMARY KEY ("id"),
+    CONSTRAINT "revenue_distribution_shares_distributionId_fkey"
+        FOREIGN KEY ("distributionId") REFERENCES "revenue_distributions"("id")
+        ON DELETE CASCADE ON UPDATE CASCADE
+);
+
+CREATE UNIQUE INDEX "revenue_distribution_shares_distributionId_wallet_key"
+    ON "revenue_distribution_shares"("distributionId", "wallet");
+CREATE INDEX "revenue_distribution_shares_wallet_idx"
+    ON "revenue_distribution_shares"("wallet");
+
+CREATE TABLE "revenue_share_claims" (
+    "id" TEXT NOT NULL,
+    "distributionId" TEXT NOT NULL,
+    "shareId" TEXT NOT NULL,
+    "wallet" TEXT NOT NULL,
+    "amount" DECIMAL(20,7) NOT NULL,
+    "ledger" INTEGER NOT NULL,
+    "txHash" TEXT NOT NULL,
+    "eventIndex" INTEGER NOT NULL DEFAULT 0,
+    "claimedAt" TIMESTAMP(3) NOT NULL,
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+
+    CONSTRAINT "revenue_share_claims_pkey" PRIMARY KEY ("id"),
+    CONSTRAINT "revenue_share_claims_distributionId_fkey"
+        FOREIGN KEY ("distributionId") REFERENCES "revenue_distributions"("id")
+        ON DELETE CASCADE ON UPDATE CASCADE,
+    CONSTRAINT "revenue_share_claims_shareId_fkey"
+        FOREIGN KEY ("shareId") REFERENCES "revenue_distribution_shares"("id")
+        ON DELETE CASCADE ON UPDATE CASCADE
+);
+
+CREATE UNIQUE INDEX "revenue_share_claims_ledger_txHash_eventIndex_key"
+    ON "revenue_share_claims"("ledger", "txHash", "eventIndex");
+CREATE INDEX "revenue_share_claims_distributionId_wallet_idx"
+    ON "revenue_share_claims"("distributionId", "wallet");

prisma/schema/revenue.prisma

@@ -70,3 +70,62 @@ model ProtocolRevenueShare {
   @@unique([distributionId, wallet])
   @@index([wallet])
 }
+
+/// A creator-key revenue distribution indexed from RevenueDistributed.
+model RevenueDistribution {
+  id                String   @id @default(cuid())
+  keyId             String
+  chainDistributionId String?
+  amount            Decimal  @db.Decimal(20, 7)
+  holderCount       Int
+  ledger            Int
+  txHash            String
+  eventIndex        Int      @default(0)
+  distributedAt     DateTime
+  createdAt         DateTime @default(now())
+
+  shares            RevenueDistributionShare[]
+  claims            RevenueShareClaim[]
+
+  @@unique([ledger, txHash, eventIndex])
+  @@unique([keyId, chainDistributionId])
+  @@index([keyId, distributedAt(sort: Desc), id(sort: Desc)])
+  @@map("revenue_distributions")
+}
+
+/// A wallet's claimable allocation at the time a key distribution was made.
+model RevenueDistributionShare {
+  id                String   @id @default(cuid())
+  distributionId    String
+  wallet            String
+  amount            Decimal  @db.Decimal(20, 7)
+  createdAt         DateTime @default(now())
+
+  distribution      RevenueDistribution @relation(fields: [distributionId], references: [id], onDelete: Cascade)
+  claims            RevenueShareClaim[]
+
+  @@unique([distributionId, wallet])
+  @@index([wallet])
+  @@map("revenue_distribution_shares")
+}
+
+/// An indexed on-chain RevenueShareClaimed event.
+model RevenueShareClaim {
+  id                String   @id @default(cuid())
+  distributionId    String
+  shareId           String
+  wallet            String
+  amount            Decimal  @db.Decimal(20, 7)
+  ledger            Int
+  txHash            String
+  eventIndex        Int      @default(0)
+  claimedAt         DateTime
+  createdAt         DateTime @default(now())
+
+  distribution      RevenueDistribution @relation(fields: [distributionId], references: [id], onDelete: Cascade)
+  share             RevenueDistributionShare @relation(fields: [shareId], references: [id], onDelete: Cascade)
+
+  @@unique([ledger, txHash, eventIndex])
+  @@index([distributionId, wallet])
+  @@map("revenue_share_claims")
+}

src/modules/indexer/indexer-pipeline-dispatch.integration.test.ts

@@ -8,6 +8,7 @@ import {
    SOCIAL_METADATA_UPDATED_EVENT_TYPE,
 } from './social-metadata-indexer.service';
 import { processDividendEvents } from './dividend-indexer.service';
+import { processRevenueShareEvents } from './revenue-share-indexer.service';
 import { processGovernanceProposalEvents } from './governance-proposal-indexer.service';
 import { upsertPriceSnapshot } from './price-snapshot.service';
 import { prisma } from '../../utils/prisma.utils';
@@ -21,6 +22,9 @@ jest.mock('./social-metadata-indexer.service', () => ({
 jest.mock('./dividend-indexer.service', () => ({
    processDividendEvents: jest.fn(),
 }));
+jest.mock('./revenue-share-indexer.service', () => ({
+   processRevenueShareEvents: jest.fn(),
+}));
 jest.mock('./lp-indexer.service', () => {
    const actual = jest.requireActual('./lp-indexer.service');
    return {
@@ -81,6 +85,7 @@ jest.mock('../../utils/logger.utils', () => ({
 
 const mockSocialHandler = processSocialMetadataEvents as jest.Mock;
 const mockDividendHandler = processDividendEvents as jest.Mock;
+const mockRevenueShareHandler = processRevenueShareEvents as jest.Mock;
 const mockGovernanceHandler = processGovernanceProposalEvents as jest.Mock;
 const mockUpsertPriceSnapshot = upsertPriceSnapshot as jest.Mock;
 const mockPrisma = prisma as unknown as {
@@ -196,6 +201,50 @@ describe('processIndexerEvents', () => {
       expect(mockPrisma.indexedLedger.upsert).toHaveBeenCalledTimes(1);
    });
 
+   it('routes creator revenue distribution and claim events before advancing the cursor', async () => {
+      const distributionEvent = socialEvent({
+         txHash: 'tx-revenue-distributed',
+         eventType: 'RevenueDistributed',
+         ledger: 812,
+      });
+      const claimEvent = socialEvent({
+         txHash: 'tx-revenue-claimed',
+         eventType: 'RevenueShareClaimed',
+         ledger: 813,
+      });
+
+      await processIndexerEvents([distributionEvent, claimEvent]);
+
+      expect(mockRevenueShareHandler).toHaveBeenCalledWith([
+         distributionEvent,
+         claimEvent,
+      ]);
+      expect(mockPrisma.indexedLedger.upsert).toHaveBeenCalledTimes(1);
+      expect(mockPrisma.indexedLedger.upsert).toHaveBeenCalledWith(
+         expect.objectContaining({
+            update: expect.objectContaining({ ledger: 813 }),
+         })
+      );
+   });
+
+   it('does not advance the cursor if creator revenue indexing fails', async () => {
+      mockRevenueShareHandler.mockRejectedValueOnce(
+         new Error('revenue write failed')
+      );
+
+      await expect(
+         processIndexerEvents([
+            socialEvent({
+               txHash: 'tx-revenue-failure',
+               eventType: 'RevenueDistributed',
+               ledger: 814,
+            }),
+         ])
+      ).rejects.toThrow('revenue write failed');
+
+      expect(mockPrisma.indexedLedger.upsert).not.toHaveBeenCalled();
+   });
+
    it('still advances the cursor for batches with only unknown event types', async () => {
       await processIndexerEvents([
          {

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

@@ -372,6 +372,13 @@ export async function processIndexerEvents(
          event.eventType === 'DividendClaimed' ||
          event.eventType === 'DIVIDEND_CLAIMED'
    );
+   const revenueShareEvents = events.filter(
+      event =>
+         event.eventType === 'RevenueDistributed' ||
+         event.eventType === 'REVENUE_DISTRIBUTED' ||
+         event.eventType === 'RevenueShareClaimed' ||
+         event.eventType === 'REVENUE_SHARE_CLAIMED'
+   );
    const socialEvents = events.filter(
       event => event.eventType === SOCIAL_METADATA_UPDATED_EVENT_TYPE
    );
@@ -482,6 +489,12 @@ export async function processIndexerEvents(
       await processDividendEvents(dividendEvents, { advanceCursor: false });
    }
 
+   if (revenueShareEvents.length > 0) {
+      const { processRevenueShareEvents } =
+         await import('./revenue-share-indexer.service');
+      await processRevenueShareEvents(revenueShareEvents);
+   }
+
    const { isLpEventType, processLpEvents } =
       await import('./lp-indexer.service');
    const lpEvents = events.filter(event => isLpEventType(event.eventType));