diff --git a/docs/api-endpoints.md b/docs/api-endpoints.md index 1422ba50..fba876bc 100644 --- a/docs/api-endpoints.md +++ b/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 diff --git a/prisma/schema/migrations/20261021000200_add_creator_revenue_distributions/migration.sql b/prisma/schema/migrations/20261021000200_add_creator_revenue_distributions/migration.sql new file mode 100644 index 00000000..8f28ceb4 --- /dev/null +++ b/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"); diff --git a/prisma/schema/revenue.prisma b/prisma/schema/revenue.prisma index b7931edb..1b4c1e82 100644 --- a/prisma/schema/revenue.prisma +++ b/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") +} diff --git a/src/modules/indexer/indexer-pipeline-dispatch.integration.test.ts b/src/modules/indexer/indexer-pipeline-dispatch.integration.test.ts index 2282345d..54f2969e 100644 --- a/src/modules/indexer/indexer-pipeline-dispatch.integration.test.ts +++ b/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([ { diff --git a/src/modules/indexer/indexer-pipeline.service.ts b/src/modules/indexer/indexer-pipeline.service.ts index 0d883990..328d72c6 100644 --- a/src/modules/indexer/indexer-pipeline.service.ts +++ b/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)); diff --git a/src/modules/indexer/revenue-share-indexer.service.test.ts b/src/modules/indexer/revenue-share-indexer.service.test.ts new file mode 100644 index 00000000..a5e6782a --- /dev/null +++ b/src/modules/indexer/revenue-share-indexer.service.test.ts @@ -0,0 +1,203 @@ +import { prisma } from '../../utils/prisma.utils'; +import { cacheInvalidate } from '../../utils/redis.utils'; +import { + parseRevenueShareContractLog, + processRevenueShareEvents, +} from './revenue-share-indexer.service'; + +jest.mock('../../utils/prisma.utils', () => ({ + prisma: { + revenueDistribution: { findFirst: jest.fn(), findMany: jest.fn() }, + keyOwnership: { findMany: jest.fn() }, + revenueShareClaim: { findFirst: jest.fn(), create: jest.fn() }, + $transaction: jest.fn(), + }, +})); + +jest.mock('../../utils/redis.utils', () => ({ + cacheInvalidate: jest.fn(), +})); + +jest.mock('../../utils/logger.utils', () => ({ + logger: { + info: jest.fn(), + warn: jest.fn(), + error: jest.fn(), + debug: jest.fn(), + }, +})); + +const mockPrisma = prisma as unknown as { + revenueDistribution: { findFirst: jest.Mock; findMany: jest.Mock }; + keyOwnership: { findMany: jest.Mock }; + revenueShareClaim: { findFirst: jest.Mock; create: jest.Mock }; + $transaction: jest.Mock; +}; +const mockCacheInvalidate = cacheInvalidate as jest.Mock; + +describe('creator revenue share indexer', () => { + beforeEach(() => { + jest.clearAllMocks(); + mockPrisma.revenueDistribution.findFirst.mockResolvedValue(null); + mockPrisma.revenueDistribution.findMany.mockResolvedValue([]); + mockPrisma.keyOwnership.findMany.mockResolvedValue([ + { ownerAddress: 'GWALLET1', balance: 1 }, + { ownerAddress: 'GWALLET2', balance: 3 }, + ]); + mockPrisma.$transaction.mockImplementation( + async (callback: (transaction: unknown) => Promise) => + callback({ + revenueDistribution: { + create: jest + .fn() + .mockResolvedValue({ id: 'distribution-db-1' }), + }, + revenueDistributionShare: { + createMany: jest.fn().mockResolvedValue({ count: 2 }), + }, + }) + ); + mockCacheInvalidate.mockResolvedValue(undefined); + }); + + it('records each distribution and snapshots proportional wallet shares', async () => { + const tx = { + revenueDistribution: { + create: jest.fn().mockResolvedValue({ id: 'distribution-db-1' }), + }, + revenueDistributionShare: { + createMany: jest.fn().mockResolvedValue({ count: 2 }), + }, + }; + mockPrisma.$transaction.mockImplementationOnce( + async (callback: (transaction: unknown) => Promise) => + callback(tx) + ); + + await processRevenueShareEvents([ + { + eventType: 'RevenueDistributed', + keyId: 'key-1', + distributionId: 'chain-dist-1', + amount: '12', + distributedAt: '2026-10-20T10:00:00.000Z', + holderCount: 2, + ledger: 900, + txHash: 'tx-distributed', + eventIndex: 0, + }, + ]); + + expect(tx.revenueDistribution.create).toHaveBeenCalledWith({ + data: expect.objectContaining({ + keyId: 'key-1', + chainDistributionId: 'chain-dist-1', + amount: '12', + holderCount: 2, + ledger: 900, + }), + }); + expect(tx.revenueDistributionShare.createMany).toHaveBeenCalledWith({ + data: [ + { + distributionId: 'distribution-db-1', + wallet: 'GWALLET1', + amount: '3.0000000', + }, + { + distributionId: 'distribution-db-1', + wallet: 'GWALLET2', + amount: '9.0000000', + }, + ], + }); + expect(mockCacheInvalidate).toHaveBeenCalledWith( + 'key:revenue-distributions:key-1:*' + ); + }); + + it('indexes claim events idempotently and invalidates the key cache', async () => { + mockPrisma.revenueShareClaim.findFirst.mockResolvedValue(null); + mockPrisma.revenueDistribution.findMany.mockResolvedValue([ + { + id: 'distribution-db-1', + shares: [ + { + id: 'share-1', + amount: 5, + claims: [{ amount: 1 }], + }, + ], + }, + ]); + mockPrisma.revenueShareClaim.create.mockResolvedValue({}); + + await processRevenueShareEvents([ + { + eventType: 'RevenueShareClaimed', + keyId: 'key-1', + wallet: 'GWALLET1', + amount: '2', + claimedAt: '2026-10-20T11:00:00.000Z', + ledger: 901, + txHash: 'tx-claimed', + eventIndex: 3, + }, + ]); + + expect(mockPrisma.revenueShareClaim.create).toHaveBeenCalledWith({ + data: expect.objectContaining({ + distributionId: 'distribution-db-1', + shareId: 'share-1', + wallet: 'GWALLET1', + amount: '2', + }), + }); + expect(mockCacheInvalidate).toHaveBeenCalledWith( + 'key:revenue-distributions:key-1:*' + ); + }); + + it('fails a malformed event so the shared indexer cursor can retry it', async () => { + await expect( + processRevenueShareEvents([ + { + eventType: 'RevenueDistributed', + keyId: 'key-1', + amount: 'not-a-number', + ledger: 902, + txHash: 'tx-invalid', + eventIndex: 0, + }, + ]) + ).rejects.toThrow('missing valid key, amount, date, or ledger'); + expect(mockPrisma.$transaction).not.toHaveBeenCalled(); + }); + + it('parses Soroban RevenueShareClaimed logs', () => { + expect( + parseRevenueShareContractLog({ + contractId: 'contract-1', + topics: [ + 'revenue_share_claimed', + 'key-1', + 'GWALLET1', + '2.5', + 'dist-1', + ], + ledger: 903, + txHash: 'tx-raw-claim', + timestamp: '2026-10-20T12:00:00.000Z', + eventIndex: 1, + }) + ).toEqual( + expect.objectContaining({ + eventType: 'RevenueShareClaimed', + keyId: 'key-1', + wallet: 'GWALLET1', + amount: '2.5', + distributionId: 'dist-1', + }) + ); + }); +}); diff --git a/src/modules/indexer/revenue-share-indexer.service.ts b/src/modules/indexer/revenue-share-indexer.service.ts new file mode 100644 index 00000000..87cd3565 --- /dev/null +++ b/src/modules/indexer/revenue-share-indexer.service.ts @@ -0,0 +1,507 @@ +import { prisma } from '../../utils/prisma.utils'; +import { logger } from '../../utils/logger.utils'; +import { + processIndexerChainEvents, + IndexerChainEvent, +} from '../../utils/indexer-event-processor.utils'; +import { cacheInvalidate } from '../../utils/redis.utils'; +import { revenueDistributionCachePattern } from '../revenue/revenue-distributions.service'; + +const REVENUE_DISTRIBUTION_EVENT_TYPES = new Set([ + 'RevenueDistributed', + 'REVENUE_DISTRIBUTED', +]); +const REVENUE_CLAIM_EVENT_TYPES = new Set([ + 'RevenueShareClaimed', + 'REVENUE_SHARE_CLAIMED', +]); + +function readString( + event: IndexerChainEvent, + ...names: string[] +): string | null { + for (const name of names) { + const value = event[name]; + if ( + (typeof value === 'string' || typeof value === 'number') && + String(value).trim() !== '' + ) { + return String(value); + } + } + return null; +} + +function readDate(value: unknown): Date | null { + const date = + typeof value === 'number' + ? new Date(value < 100_000_000_000 ? value * 1000 : value) + : value instanceof Date + ? new Date(value.getTime()) + : typeof value === 'string' + ? new Date(value) + : null; + return date && Number.isFinite(date.getTime()) ? date : null; +} + +function readAmount(value: unknown): string | null { + if ( + (typeof value !== 'string' && + typeof value !== 'number' && + typeof value !== 'bigint') || + String(value).trim() === '' + ) { + return null; + } + const amount = Number(value); + return Number.isFinite(amount) && amount >= 0 ? String(value) : null; +} + +function isUniqueConstraintError(error: unknown): boolean { + return ( + typeof error === 'object' && + error !== null && + 'code' in error && + error.code === 'P2002' + ); +} + +function eventId(event: IndexerChainEvent): string { + return `${event.txHash}:${event.eventIndex}`; +} + +async function processRevenueDistributedEvent( + event: IndexerChainEvent +): Promise { + const keyId = readString( + event, + 'keyId', + 'key_id', + 'creatorId', + 'creator_id' + ); + const amount = readAmount( + event.totalAmount ?? + event.total_amount ?? + event.totalDistributed ?? + event.total_distributed ?? + event.amount ?? + event.revenueAmount ?? + event.revenue_amount + ); + const distributedAt = readDate( + event.distributedAt ?? + event.distributed_at ?? + event.timestamp ?? + event.createdAt ?? + event.ledgerClosedAt ?? + event.ledger_closed_at + ); + const ledger = event.ledger; + if ( + !keyId || + amount === null || + !distributedAt || + typeof ledger !== 'number' || + !Number.isInteger(ledger) || + typeof event.txHash !== 'string' || + !event.txHash + ) { + throw new Error( + `RevenueDistributed event ${eventId(event)} is missing valid key, amount, date, or ledger` + ); + } + + const chainDistributionId = readString( + event, + 'distributionId', + 'distribution_id', + 'snapshotId', + 'snapshot_id' + ); + const existing = await prisma.revenueDistribution.findFirst({ + where: { + OR: [ + { ledger, txHash: event.txHash, eventIndex: event.eventIndex }, + ...(chainDistributionId ? [{ keyId, chainDistributionId }] : []), + ], + }, + select: { id: true }, + }); + if (existing) { + return; + } + + const holderRows = await prisma.keyOwnership.findMany({ + where: { creatorId: keyId, balance: { gt: 0 } }, + select: { ownerAddress: true, balance: true }, + }); + const suppliedRecipients = event.recipients ?? event.shares; + const shares = + Array.isArray(suppliedRecipients) && suppliedRecipients.length > 0 + ? suppliedRecipients.map((recipient: Record) => { + const wallet = + recipient.wallet ?? + recipient.address ?? + recipient.ownerAddress; + const recipientAmount = readAmount( + recipient.amount ?? recipient.share + ); + if ( + typeof wallet !== 'string' || + wallet.length === 0 || + recipientAmount === null + ) { + throw new Error( + `RevenueDistributed event ${eventId(event)} has an invalid recipient` + ); + } + return { wallet, amount: recipientAmount }; + }) + : (() => { + const totalBalance = holderRows.reduce( + (total, holder) => total + Number(holder.balance), + 0 + ); + const totalAmount = Number(amount); + return totalBalance > 0 + ? holderRows.map(holder => ({ + wallet: holder.ownerAddress, + amount: ( + (totalAmount * Number(holder.balance)) / + totalBalance + ).toFixed(7), + })) + : []; + })(); + const holderCountValue = + event.holderCount ?? + event.holder_count ?? + event.holdersCount ?? + event.holders_count; + const holderCount = + holderCountValue === undefined || holderCountValue === null + ? shares.length || holderRows.length + : Number(holderCountValue); + if (!Number.isInteger(holderCount) || holderCount < 0) { + throw new Error( + `RevenueDistributed event ${eventId(event)} has an invalid holder count` + ); + } + + try { + await prisma.$transaction(async tx => { + const distribution = await tx.revenueDistribution.create({ + data: { + keyId, + chainDistributionId, + amount, + holderCount, + ledger, + txHash: event.txHash, + eventIndex: event.eventIndex, + distributedAt, + }, + }); + if (shares.length > 0) { + await tx.revenueDistributionShare.createMany({ + data: shares.map(share => ({ + distributionId: distribution.id, + wallet: share.wallet, + amount: share.amount, + })), + }); + } + }); + } catch (error) { + if (isUniqueConstraintError(error)) { + return; + } + throw error; + } + + await cacheInvalidate(revenueDistributionCachePattern(keyId)); + logger.info( + { + keyId, + distributionId: chainDistributionId, + ledger, + txHash: event.txHash, + }, + 'Creator revenue distribution recorded' + ); +} + +async function processRevenueShareClaimedEvent( + event: IndexerChainEvent +): Promise { + const keyId = readString( + event, + 'keyId', + 'key_id', + 'creatorId', + 'creator_id' + ); + const wallet = readString( + event, + 'wallet', + 'walletAddress', + 'wallet_address', + 'claimant', + 'claimantAddress', + 'claimant_address' + ); + const chainDistributionId = readString( + event, + 'distributionId', + 'distribution_id', + 'snapshotId', + 'snapshot_id' + ); + const amount = readAmount( + event.amount ?? + event.claimedAmount ?? + event.claimed_amount ?? + event.amountXlm + ); + const claimedAt = readDate( + event.claimedAt ?? + event.claimed_at ?? + event.timestamp ?? + event.ledgerClosedAt ?? + event.ledger_closed_at + ); + const ledger = event.ledger; + if ( + !keyId || + !wallet || + amount === null || + Number(amount) <= 0 || + !claimedAt || + typeof ledger !== 'number' || + !Number.isInteger(ledger) || + typeof event.txHash !== 'string' || + !event.txHash + ) { + throw new Error( + `RevenueShareClaimed event ${eventId(event)} is missing valid key, wallet, amount, date, or ledger` + ); + } + + const existingClaim = await prisma.revenueShareClaim.findFirst({ + where: { ledger, txHash: event.txHash, eventIndex: event.eventIndex }, + select: { id: true }, + }); + if (existingClaim) { + return; + } + + const distributions = await prisma.revenueDistribution.findMany({ + where: { + keyId, + ...(chainDistributionId + ? { + OR: [{ id: chainDistributionId }, { chainDistributionId }], + } + : { shares: { some: { wallet } } }), + }, + orderBy: [{ distributedAt: 'asc' }, { id: 'asc' }], + include: { + shares: { + where: { wallet }, + include: { + claims: { select: { amount: true } }, + }, + }, + }, + }); + const pending = distributions + .map(distribution => { + const share = distribution.shares[0]; + if (!share) { + return null; + } + const claimed = share.claims.reduce( + (total, claim) => total + Number(claim.amount), + 0 + ); + const remaining = Math.max(0, Number(share.amount) - claimed); + return { distribution, share, remaining }; + }) + .filter( + (candidate): candidate is NonNullable => + candidate !== null + ) + .filter(candidate => candidate.remaining + 0.0000001 >= Number(amount)); + const exactMatches = pending.filter( + candidate => Math.abs(candidate.remaining - Number(amount)) < 0.0000001 + ); + const selected = chainDistributionId + ? pending.length === 1 + ? pending[0] + : undefined + : exactMatches.length === 1 + ? exactMatches[0] + : pending.length === 1 + ? pending[0] + : undefined; + const distribution = selected?.distribution; + const share = selected?.share; + if (!distribution || !share) { + throw new Error( + `RevenueShareClaimed event ${eventId(event)} references an unknown or ambiguous distribution or wallet share` + ); + } + + const alreadyClaimed = share.claims.reduce( + (total, claim) => total + Number(claim.amount), + 0 + ); + if (alreadyClaimed + Number(amount) > Number(share.amount) + 0.0000001) { + throw new Error( + `RevenueShareClaimed event ${eventId(event)} exceeds the wallet's distribution share` + ); + } + + try { + await prisma.revenueShareClaim.create({ + data: { + distributionId: distribution.id, + shareId: share.id, + wallet, + amount, + ledger, + txHash: event.txHash, + eventIndex: event.eventIndex, + claimedAt, + }, + }); + } catch (error) { + if (isUniqueConstraintError(error)) { + return; + } + throw error; + } + + await cacheInvalidate(revenueDistributionCachePattern(keyId)); + logger.info( + { + keyId, + wallet, + distributionId: chainDistributionId, + ledger, + txHash: event.txHash, + }, + 'Creator revenue share claim recorded' + ); +} + +export async function processRevenueShareEvents( + events: IndexerChainEvent[] +): Promise { + await processIndexerChainEvents(events, async event => { + if (REVENUE_DISTRIBUTION_EVENT_TYPES.has(event.eventType)) { + await processRevenueDistributedEvent(event); + } else if (REVENUE_CLAIM_EVENT_TYPES.has(event.eventType)) { + await processRevenueShareClaimedEvent(event); + } + }); +} + +export interface RevenueShareContractLog { + contractId: string; + topics: string[]; + data?: Record; + ledger: number; + txHash: string; + eventIndex?: number; + timestamp?: string | number | Date; +} + +export function parseRevenueShareContractLog( + log: RevenueShareContractLog +): IndexerChainEvent | null { + const topic = log.topics[0]?.toLowerCase(); + if (!topic) { + return null; + } + const data = log.data ?? {}; + const common = { + keyId: + readString( + data as IndexerChainEvent, + 'keyId', + 'key_id', + 'creatorId', + 'creator_id' + ) ?? + log.topics[1] ?? + log.contractId, + ledger: log.ledger, + txHash: log.txHash, + eventIndex: log.eventIndex ?? 0, + }; + if (['revenuedistributed', 'revenue_distributed'].includes(topic)) { + return { + eventType: 'RevenueDistributed', + ...common, + distributionId: + data.distributionId ?? data.distribution_id ?? log.topics[2], + amount: + data.amount ?? + data.totalAmount ?? + data.total_amount ?? + data.totalDistributed ?? + data.total_distributed ?? + log.topics[3], + holderCount: + data.holderCount ?? + data.holder_count ?? + data.holdersCount ?? + data.holders_count ?? + log.topics[4], + distributedAt: + data.distributedAt ?? + data.distributed_at ?? + data.timestamp ?? + data.ledgerClosedAt ?? + data.ledger_closed_at ?? + log.timestamp, + recipients: data.recipients ?? data.shares, + }; + } + if (['revenueshareclaimed', 'revenue_share_claimed'].includes(topic)) { + return { + eventType: 'RevenueShareClaimed', + ...common, + wallet: + data.wallet ?? + data.walletAddress ?? + data.wallet_address ?? + data.claimant ?? + data.claimantAddress ?? + data.claimant_address ?? + log.topics[2], + amount: + data.amount ?? + data.claimedAmount ?? + data.claimed_amount ?? + data.amountXlm ?? + log.topics[3], + distributionId: + data.distributionId ?? + data.distribution_id ?? + data.snapshotId ?? + data.snapshot_id ?? + log.topics[4], + claimedAt: + data.claimedAt ?? + data.claimed_at ?? + data.timestamp ?? + data.ledgerClosedAt ?? + data.ledger_closed_at ?? + log.timestamp, + }; + } + return null; +} diff --git a/src/modules/keys/keys.routes.ts b/src/modules/keys/keys.routes.ts index 56b4fae3..2bae2e9a 100644 --- a/src/modules/keys/keys.routes.ts +++ b/src/modules/keys/keys.routes.ts @@ -46,6 +46,7 @@ import { import { KeySearchQueryTooShortError, searchKeys } from './key-search.service'; import { KEY_SEARCH_MIN_QUERY_LENGTH } from '../../constants/notifications.constants'; import dividendRouter from '../dividends/dividend.routes'; +import revenueDistributionsRouter from '../revenue/revenue-distributions.routes'; import whitelistRouter from '../whitelist/whitelist.routes'; import { requireJwtAuth, @@ -2026,6 +2027,9 @@ router.get('/:keyId/simulate', async (req, res, next) => { // Mount dividend routes router.use('/', dividendRouter); +// Creator key revenue distribution history and wallet claims +router.use('/', revenueDistributionsRouter); + // Mount whitelist routes router.use('/', whitelistRouter); diff --git a/src/modules/revenue/revenue-distributions.routes.ts b/src/modules/revenue/revenue-distributions.routes.ts new file mode 100644 index 00000000..e1735d16 --- /dev/null +++ b/src/modules/revenue/revenue-distributions.routes.ts @@ -0,0 +1,64 @@ +import { Router } from 'express'; +import { z } from 'zod'; +import { + sendSuccess, + sendValidationError, + zodIssuesToDetails, +} from '../../utils/api-response.utils'; +import { + getRevenueDistributions, + getWalletRevenueDistributions, +} from './revenue-distributions.service'; + +const router = Router(); +const paginationQuerySchema = z.object({ + limit: z.coerce.number().int().positive().max(100).optional().default(50), + cursor: z.string().min(1).optional(), +}); +const stellarWalletSchema = z.string().regex(/^G[A-Z2-7]{55}$/); + +router.get('/:keyId/revenue-distributions', async (req, res, next) => { + const parsed = paginationQuerySchema.safeParse(req.query); + if (!parsed.success) { + sendValidationError( + res, + 'Invalid query parameters', + zodIssuesToDetails(parsed.error.issues) + ); + return; + } + try { + sendSuccess( + res, + await getRevenueDistributions({ + keyId: req.params.keyId, + ...parsed.data, + }) + ); + } catch (error) { + next(error); + } +}); + +router.get('/:keyId/revenue-distributions/:wallet', async (req, res, next) => { + const parsed = stellarWalletSchema.safeParse(req.params.wallet); + if (!parsed.success) { + sendValidationError(res, 'Invalid wallet address', [ + { field: 'wallet', message: 'Invalid Stellar wallet address' }, + ]); + return; + } + try { + sendSuccess( + res, + await getWalletRevenueDistributions({ + keyId: req.params.keyId, + wallet: parsed.data, + }) + ); + } catch (error) { + next(error); + } +}); + +export default router; diff --git a/src/modules/revenue/revenue-distributions.service.test.ts b/src/modules/revenue/revenue-distributions.service.test.ts new file mode 100644 index 00000000..06dcee59 --- /dev/null +++ b/src/modules/revenue/revenue-distributions.service.test.ts @@ -0,0 +1,120 @@ +import { prisma } from '../../utils/prisma.utils'; +import { cacheGetJson, cacheSetJson } from '../../utils/redis.utils'; +import { + getRevenueDistributions, + getWalletRevenueDistributions, + REVENUE_DISTRIBUTIONS_CACHE_TTL_SECONDS, +} from './revenue-distributions.service'; + +jest.mock('../../utils/prisma.utils', () => ({ + prisma: { revenueDistribution: { findMany: jest.fn() } }, +})); + +jest.mock('../../utils/redis.utils', () => ({ + cacheGetJson: jest.fn(), + cacheSetJson: jest.fn(), +})); + +const mockPrisma = prisma as unknown as { + revenueDistribution: { findMany: jest.Mock }; +}; +const mockCacheGet = cacheGetJson as jest.Mock; +const mockCacheSet = cacheSetJson as jest.Mock; + +describe('revenue distribution read service', () => { + beforeEach(() => { + jest.clearAllMocks(); + mockCacheGet.mockResolvedValue(null); + mockCacheSet.mockResolvedValue(undefined); + }); + + it('returns paginated distributions and caches them for 60 seconds', async () => { + mockPrisma.revenueDistribution.findMany.mockResolvedValue([ + { + id: 'dist-db-1', + chainDistributionId: 'dist-chain-1', + distributedAt: new Date('2026-10-20T10:00:00.000Z'), + amount: 25, + holderCount: 3, + }, + { + id: 'dist-db-2', + chainDistributionId: null, + distributedAt: new Date('2026-10-19T10:00:00.000Z'), + amount: 20, + holderCount: 2, + }, + ]); + + const result = await getRevenueDistributions({ + keyId: 'key-1', + limit: 1, + }); + + expect(result).toEqual({ + entries: [ + { + distributionId: 'dist-chain-1', + distributedAt: '2026-10-20T10:00:00.000Z', + amount: 25, + holderCount: 3, + }, + ], + pagination: { + limit: 1, + hasMore: true, + nextCursor: 'dist-db-1', + }, + }); + expect(mockCacheSet).toHaveBeenCalledWith( + 'key:revenue-distributions:key-1:list:1:', + result, + REVENUE_DISTRIBUTIONS_CACHE_TTL_SECONDS + ); + }); + + it('returns the wallet claimed and remaining claimable amount per distribution', async () => { + mockPrisma.revenueDistribution.findMany.mockResolvedValue([ + { + id: 'dist-db-1', + chainDistributionId: 'dist-chain-1', + distributedAt: new Date('2026-10-20T10:00:00.000Z'), + amount: 25, + holderCount: 3, + shares: [{ amount: 10, claims: [{ amount: 3 }, { amount: 2 }] }], + }, + { + id: 'dist-db-2', + chainDistributionId: null, + distributedAt: new Date('2026-10-19T10:00:00.000Z'), + amount: 20, + holderCount: 2, + shares: [], + }, + ]); + + const result = await getWalletRevenueDistributions({ + keyId: 'key-1', + wallet: 'GWALLET1', + }); + + expect(result).toEqual([ + { + distributionId: 'dist-chain-1', + distributedAt: '2026-10-20T10:00:00.000Z', + amount: 25, + holderCount: 3, + claimableAmount: 5, + claimedAmount: 5, + }, + { + distributionId: 'dist-db-2', + distributedAt: '2026-10-19T10:00:00.000Z', + amount: 20, + holderCount: 2, + claimableAmount: 0, + claimedAmount: 0, + }, + ]); + }); +}); diff --git a/src/modules/revenue/revenue-distributions.service.ts b/src/modules/revenue/revenue-distributions.service.ts new file mode 100644 index 00000000..f617c77f --- /dev/null +++ b/src/modules/revenue/revenue-distributions.service.ts @@ -0,0 +1,126 @@ +import { prisma } from '../../utils/prisma.utils'; +import { cacheGetJson, cacheSetJson } from '../../utils/redis.utils'; + +export const REVENUE_DISTRIBUTIONS_CACHE_TTL_SECONDS = 60; +const REVENUE_DISTRIBUTIONS_CACHE_PREFIX = 'key:revenue-distributions'; + +export function revenueDistributionCachePattern(keyId: string): string { + return `${REVENUE_DISTRIBUTIONS_CACHE_PREFIX}:${keyId}:*`; +} + +export interface RevenueDistributionEntry { + distributionId: string; + distributedAt: string; + amount: number; + holderCount: number; +} + +export interface RevenueDistributionPage { + entries: RevenueDistributionEntry[]; + pagination: { + limit: number; + hasMore: boolean; + nextCursor?: string; + }; +} + +export async function getRevenueDistributions(input: { + keyId: string; + limit: number; + cursor?: string; +}): Promise { + const cacheKey = `${REVENUE_DISTRIBUTIONS_CACHE_PREFIX}:${input.keyId}:list:${input.limit}:${input.cursor ?? ''}`; + const cached = await cacheGetJson(cacheKey); + if (cached) { + return cached; + } + + const distributions = await prisma.revenueDistribution.findMany({ + where: { keyId: input.keyId }, + orderBy: [{ distributedAt: 'desc' }, { id: 'desc' }], + take: input.limit + 1, + skip: input.cursor ? 1 : 0, + cursor: input.cursor ? { id: input.cursor } : undefined, + }); + const hasMore = distributions.length > input.limit; + const page = distributions.slice(0, input.limit); + const result: RevenueDistributionPage = { + entries: page.map(distribution => ({ + distributionId: distribution.chainDistributionId ?? distribution.id, + distributedAt: distribution.distributedAt.toISOString(), + amount: Number(distribution.amount), + holderCount: distribution.holderCount, + })), + pagination: { + limit: input.limit, + hasMore, + nextCursor: + hasMore && page.length > 0 ? page[page.length - 1].id : undefined, + }, + }; + + await cacheSetJson( + cacheKey, + result, + REVENUE_DISTRIBUTIONS_CACHE_TTL_SECONDS + ); + return result; +} + +export interface WalletRevenueDistributionEntry extends RevenueDistributionEntry { + claimableAmount: number; + claimedAmount: number; +} + +export async function getWalletRevenueDistributions(input: { + keyId: string; + wallet: string; +}): Promise { + const cacheKey = `${REVENUE_DISTRIBUTIONS_CACHE_PREFIX}:${input.keyId}:wallet:${input.wallet}`; + const cached = + await cacheGetJson(cacheKey); + if (cached) { + return cached; + } + + const distributions = await prisma.revenueDistribution.findMany({ + where: { keyId: input.keyId }, + orderBy: [{ distributedAt: 'desc' }, { id: 'desc' }], + include: { + shares: { + where: { wallet: input.wallet }, + include: { + claims: { + where: { wallet: input.wallet }, + select: { amount: true }, + }, + }, + }, + }, + }); + + const result = distributions.map(distribution => { + const share = distribution.shares[0]; + const claimableAmount = share ? Number(share.amount) : 0; + const claimedAmount = + share?.claims.reduce( + (total, claim) => total + Number(claim.amount), + 0 + ) ?? 0; + return { + distributionId: distribution.chainDistributionId ?? distribution.id, + distributedAt: distribution.distributedAt.toISOString(), + amount: Number(distribution.amount), + holderCount: distribution.holderCount, + claimableAmount: Math.max(0, claimableAmount - claimedAmount), + claimedAmount, + }; + }); + + await cacheSetJson( + cacheKey, + result, + REVENUE_DISTRIBUTIONS_CACHE_TTL_SECONDS + ); + return result; +}