diff --git a/prisma/schema/instalment-schedule.prisma b/prisma/schema/instalment-schedule.prisma new file mode 100644 index 00000000..c1d35902 --- /dev/null +++ b/prisma/schema/instalment-schedule.prisma @@ -0,0 +1,27 @@ +// prisma/schema/instalment-schedule.prisma +// Invoice payment schedule and InstalmentPaid event read model (#1177). + +/// One row per invoice schedule instalment, updated/inserted as InstalmentPaid +/// contract events are indexed. Unpaid rows have paidAt = null; overdue status +/// is derived at request time by comparing dueDate against the current clock. +model InstalmentPayment { + id String @id @default(cuid()) + invoiceId String + instalmentNumber Int + totalInstalments Int? + amount Decimal @db.Decimal(30, 7) + dueDate DateTime + paidAt DateTime? + payer String? + txHash String? + eventIndex Int? + ledger Int? + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + + @@unique([invoiceId, instalmentNumber]) + @@unique([txHash, eventIndex]) + @@index([invoiceId, dueDate(sort: Asc)]) + @@index([invoiceId, instalmentNumber(sort: Asc)]) + @@map("instalment_payments") +} diff --git a/prisma/schema/key-collaboration.prisma b/prisma/schema/key-collaboration.prisma index e9f27fc1..092858cd 100644 --- a/prisma/schema/key-collaboration.prisma +++ b/prisma/schema/key-collaboration.prisma @@ -46,7 +46,7 @@ model KeyCoCreator { @@map("key_cocreators") } -/// One row per indexed TipSent event (#1132). +/// One row per indexed TipSent event (#1132, #1173). model KeyTip { id String @id @default(cuid()) /// Key ID (CreatorProfile.id) that received the tip. @@ -55,6 +55,8 @@ model KeyTip { tipper String /// Tip amount in XLM (converted from stroops). amount Decimal @db.Decimal(20, 7) + /// Whether the tipper chose to tip anonymously (#1173). + anonymous Boolean @default(false) /// IPFS CID of the tip message, when one was attached. messageHash String? ledger Int? diff --git a/prisma/schema/migrations/20261021233500_add_instalment_payments_and_anonymous_tips/migration.sql b/prisma/schema/migrations/20261021233500_add_instalment_payments_and_anonymous_tips/migration.sql new file mode 100644 index 00000000..05f4950a --- /dev/null +++ b/prisma/schema/migrations/20261021233500_add_instalment_payments_and_anonymous_tips/migration.sql @@ -0,0 +1,33 @@ +-- AlterTable +ALTER TABLE "key_tips" ADD COLUMN "anonymous" BOOLEAN NOT NULL DEFAULT false; + +-- CreateTable +CREATE TABLE "instalment_payments" ( + "id" TEXT NOT NULL, + "invoiceId" TEXT NOT NULL, + "instalmentNumber" INTEGER NOT NULL, + "totalInstalments" INTEGER, + "amount" DECIMAL(30,7) NOT NULL, + "dueDate" TIMESTAMP(3) NOT NULL, + "paidAt" TIMESTAMP(3), + "payer" TEXT, + "txHash" TEXT, + "eventIndex" INTEGER, + "ledger" INTEGER, + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updatedAt" TIMESTAMP(3) NOT NULL, + + CONSTRAINT "instalment_payments_pkey" PRIMARY KEY ("id") +); + +-- CreateIndex +CREATE UNIQUE INDEX "instalment_payments_invoiceId_instalmentNumber_key" ON "instalment_payments"("invoiceId", "instalmentNumber"); + +-- CreateIndex +CREATE UNIQUE INDEX "instalment_payments_txHash_eventIndex_key" ON "instalment_payments"("txHash", "eventIndex"); + +-- CreateIndex +CREATE INDEX "instalment_payments_invoiceId_dueDate_idx" ON "instalment_payments"("invoiceId", "dueDate" ASC); + +-- CreateIndex +CREATE INDEX "instalment_payments_invoiceId_instalmentNumber_idx" ON "instalment_payments"("invoiceId", "instalmentNumber" ASC); diff --git a/src/constants/redis.constants.ts b/src/constants/redis.constants.ts index 563d8849..4e633a98 100644 --- a/src/constants/redis.constants.ts +++ b/src/constants/redis.constants.ts @@ -109,6 +109,7 @@ export const contentAccessCachePattern = (keyId: string): string => // key_tips / key_tip_totals tables with a 60s TTL. IPFS tip message // content fetched on demand is cached longer (immutable per CID). export const KEY_TIPS_CACHE_TTL_SECONDS = 60; +export const KEY_TIPS_LEADERBOARD_CACHE_TTL_SECONDS = 300; export const KEY_TIP_CONTENT_CACHE_TTL_SECONDS = 3600; export const keyTipsListCacheKey = ( @@ -120,8 +121,35 @@ export const keyTipsListCacheKey = ( export const keyTipsTotalCacheKey = (keyId: string): string => `key:tips:${keyId}:total`; +export const keyTipsLeaderboardCacheKey = ( + keyId: string, + limit: number +): string => `key:tips:${keyId}:leaderboard:${limit}`; + export const keyTipContentCacheKey = (messageHash: string): string => `key:tips:content:${messageHash}`; export const keyTipsCachePattern = (keyId: string): string => `key:tips:${keyId}:*`; + +// ── Key community polls endpoints (#1174) ─────────────────────── +// GET /keys/:key_id/polls and /polls/:poll_id are cached for 30s and +// invalidated on PollVoteCast and PollClosed events. +export const KEY_POLLS_CACHE_TTL_SECONDS = 30; + +export const keyPollsListCacheKey = ( + keyId: string, + status: string, + page: number, + limit: number +): string => `key:polls:${keyId}:list:${status}:${page}:${limit}`; + +export const keyPollDetailCacheKey = ( + keyId: string, + pollId: string, + wallet: string +): string => `key:polls:${keyId}:detail:${pollId}:${wallet}`; + +export const keyPollsCachePattern = (keyId: string): string => + `key:polls:${keyId}:*`; + diff --git a/src/modules/governance/governance-delegation-indexer.service.ts b/src/modules/governance/governance-delegation-indexer.service.ts index 472dfca7..b4e6d76e 100644 --- a/src/modules/governance/governance-delegation-indexer.service.ts +++ b/src/modules/governance/governance-delegation-indexer.service.ts @@ -34,8 +34,20 @@ import { delegateCachePattern, delegatorsCachePattern, delegationHistoryCachePattern, + votingWeightCachePattern, } from './governance-delegation.service'; +export const DELEGATION_SET_EVENT_TYPES: ReadonlySet = new Set([ + 'DELEGATION_SET', + 'VotingDelegated', + 'VOTING_DELEGATED', +]); + +export const DELEGATION_REVOKED_EVENT_TYPES: ReadonlySet = new Set([ + 'DELEGATION_REVOKED', + 'DelegationRevoked', +]); + // ── Typed event interfaces ──────────────────────────────────── /** @@ -51,7 +63,7 @@ import { * occurredAt — ISO-8601 timestamp */ export interface DelegationSetEvent extends IndexerChainEvent { - eventType: 'DELEGATION_SET'; + eventType: 'DELEGATION_SET' | 'VotingDelegated' | 'VOTING_DELEGATED'; delegatorWallet: string; delegateeWallet: string; keyId: string; @@ -70,7 +82,7 @@ export interface DelegationSetEvent extends IndexerChainEvent { * occurredAt — ISO-8601 timestamp */ export interface DelegationRevokedEvent extends IndexerChainEvent { - eventType: 'DELEGATION_REVOKED'; + eventType: 'DELEGATION_REVOKED' | 'DelegationRevoked'; delegatorWallet: string; keyId: string; occurredAt: string; @@ -89,7 +101,7 @@ const DELEGATION_SET_REQUIRED_FIELDS: (keyof DelegationSetEvent)[] = [ ]; /** - * Process a batch of DELEGATION_SET events. + * Process a batch of DELEGATION_SET / VotingDelegated events. * * For each event: * 1. Validates required fields. @@ -106,7 +118,7 @@ export async function processDelegationSetEvents( events: IndexerChainEvent[] ): Promise { await processIndexerChainEvents(events, async event => { - if (event.eventType !== 'DELEGATION_SET') return; + if (!DELEGATION_SET_EVENT_TYPES.has(event.eventType)) return; const e = event as DelegationSetEvent; @@ -250,10 +262,13 @@ export async function processDelegationSetEvents( delegatorsCachePattern(e.delegateeWallet), delegationHistoryCachePattern(e.delegatorWallet), delegationHistoryCachePattern(e.delegateeWallet), + votingWeightCachePattern(e.delegatorWallet), + votingWeightCachePattern(e.delegateeWallet), ]; if (previousDelegatee && previousDelegatee !== e.delegateeWallet) { patternsToInvalidate.push(delegatorsCachePattern(previousDelegatee)); patternsToInvalidate.push(delegationHistoryCachePattern(previousDelegatee)); + patternsToInvalidate.push(votingWeightCachePattern(previousDelegatee)); } await cacheInvalidate(...patternsToInvalidate); @@ -282,7 +297,7 @@ const DELEGATION_REVOKED_REQUIRED_FIELDS: (keyof DelegationRevokedEvent)[] = [ ]; /** - * Process a batch of DELEGATION_REVOKED events. + * Process a batch of DELEGATION_REVOKED / DelegationRevoked events. * * For each event: * 1. Validates required fields. @@ -298,7 +313,7 @@ export async function processDelegationRevokedEvents( events: IndexerChainEvent[] ): Promise { await processIndexerChainEvents(events, async event => { - if (event.eventType !== 'DELEGATION_REVOKED') return; + if (!DELEGATION_REVOKED_EVENT_TYPES.has(event.eventType)) return; const e = event as DelegationRevokedEvent; @@ -421,7 +436,9 @@ export async function processDelegationRevokedEvents( delegateCachePattern(e.delegatorWallet), delegatorsCachePattern(previousDelegatee), delegationHistoryCachePattern(e.delegatorWallet), - delegationHistoryCachePattern(previousDelegatee) + delegationHistoryCachePattern(previousDelegatee), + votingWeightCachePattern(e.delegatorWallet), + votingWeightCachePattern(previousDelegatee) ); logger.info( diff --git a/src/modules/governance/governance-delegation.service.ts b/src/modules/governance/governance-delegation.service.ts index e08646e3..cdc0d636 100644 --- a/src/modules/governance/governance-delegation.service.ts +++ b/src/modules/governance/governance-delegation.service.ts @@ -10,7 +10,11 @@ // getDelegationHistory — full event log for a wallet across all keys import { prisma } from '../../utils/prisma.utils'; -import { cacheGetJson, cacheSetJson } from '../../utils/redis.utils'; +import { + cacheGetJson, + cacheInvalidate, + cacheSetJson, +} from '../../utils/redis.utils'; import { buildOffsetPaginationMeta } from '../../utils/pagination.utils'; // ── Cache TTLs ──────────────────────────────────────────────── @@ -18,6 +22,7 @@ import { buildOffsetPaginationMeta } from '../../utils/pagination.utils'; const DELEGATE_CACHE_TTL_SECONDS = 60; const DELEGATORS_CACHE_TTL_SECONDS = 60; const HISTORY_CACHE_TTL_SECONDS = 60; +export const VOTING_WEIGHT_CACHE_TTL_SECONDS = 60; // ── Error classes ───────────────────────────────────────────── @@ -66,6 +71,20 @@ export interface DelegationHistoryItem { occurredAt: string; } +export interface DelegatorWeightEntry { + wallet: string; + amount: number; +} + +export interface KeyGovernanceVotingWeight { + wallet: string; + key_id: string; + own_weight: number; + delegated_weight: number; + total: number; + delegators: DelegatorWeightEntry[]; +} + // ── Cache invalidation helpers (used by indexer) ────────────── export function delegateCachePattern(wallet: string): string { @@ -80,6 +99,28 @@ export function delegationHistoryCachePattern(wallet: string): string { return `governance:delegation:history:${wallet}:*`; } +export function votingWeightCacheKey(wallet: string, keyId: string): string { + return `governance:voting-weight:${wallet}:${keyId}`; +} + +export function votingWeightCachePattern(wallet: string): string { + return `governance:voting-weight:${wallet}:*`; +} + +export async function invalidateVotingWeightCache( + wallet: string, + keyId?: string +): Promise { + if (keyId) { + await cacheInvalidate( + votingWeightCacheKey(wallet, keyId), + votingWeightCachePattern(wallet) + ); + return; + } + await cacheInvalidate(votingWeightCachePattern(wallet)); +} + // ── Current delegate ────────────────────────────────────────── export interface GetCurrentDelegateQuery { @@ -298,6 +339,108 @@ export async function getDelegatedVoteWeight( return Number(agg._sum.balance ?? 0); } +/** + * Returns a wallet's total governance voting weight for a key including + * delegated weight from active delegations (#1176). + * + * - `own_weight`: wallet's current holder balance (`KeyOwnership`) + * - `delegated_weight`: sum of active delegators' balances for `keyId` + * - `total`: `own_weight + delegated_weight` + * - `delegators`: `{ wallet, amount }[]` for each active delegating wallet + * - Cached for 60s and invalidated on VotingDelegated / DelegationRevoked + * - Returns `total: 0` when the wallet has no holdings and no delegations + */ +export async function getKeyGovernanceVotingWeight( + wallet: string, + keyId: string +): Promise { + const normalizedWallet = wallet.trim(); + const normalizedKeyId = keyId.trim(); + const cacheKey = votingWeightCacheKey(normalizedWallet, normalizedKeyId); + + try { + const cached = await cacheGetJson(cacheKey); + if (cached !== null) { + return cached; + } + } catch { + // Cache read failure falls through to database. + } + + const [ownOwnership, activeDelegations] = await Promise.all([ + prisma.keyOwnership.findUnique({ + where: { + ownerAddress_creatorId: { + ownerAddress: normalizedWallet, + creatorId: normalizedKeyId, + }, + }, + select: { balance: true }, + }), + prisma.voteDelegation.findMany({ + where: { + delegateeWallet: normalizedWallet, + keyId: normalizedKeyId, + isActive: true, + }, + orderBy: { occurredAt: 'desc' }, + select: { delegatorWallet: true }, + }), + ]); + + const ownWeight = ownOwnership + ? Math.max(0, Number(ownOwnership.balance)) + : 0; + + let delegators: DelegatorWeightEntry[] = []; + if (activeDelegations.length > 0) { + const delegatorWallets = activeDelegations.map(d => d.delegatorWallet); + const delegatorOwnerships = await prisma.keyOwnership.findMany({ + where: { + ownerAddress: { in: delegatorWallets }, + creatorId: normalizedKeyId, + }, + select: { + ownerAddress: true, + balance: true, + }, + }); + + const balanceByWallet = new Map(); + for (const row of delegatorOwnerships) { + balanceByWallet.set( + row.ownerAddress, + Math.max(0, Number(row.balance)) + ); + } + + delegators = activeDelegations.map(d => ({ + wallet: d.delegatorWallet, + amount: balanceByWallet.get(d.delegatorWallet) ?? 0, + })); + } + + const delegatedWeight = delegators.reduce((sum, d) => sum + d.amount, 0); + const total = ownWeight + delegatedWeight; + + const result: KeyGovernanceVotingWeight = { + wallet: normalizedWallet, + key_id: normalizedKeyId, + own_weight: ownWeight, + delegated_weight: delegatedWeight, + total, + delegators, + }; + + try { + await cacheSetJson(cacheKey, result, VOTING_WEIGHT_CACHE_TTL_SECONDS); + } catch { + // Cache write failure is non-critical. + } + + return result; +} + // ── Row mappers ─────────────────────────────────────────────── function mapDelegationRow(row: { diff --git a/src/modules/governance/governance-voting-weight.test.ts b/src/modules/governance/governance-voting-weight.test.ts new file mode 100644 index 00000000..ddc98683 --- /dev/null +++ b/src/modules/governance/governance-voting-weight.test.ts @@ -0,0 +1,193 @@ +import express from 'express'; +import request from 'supertest'; +import governanceRouter from './governance.routes'; +import { getKeyGovernanceVotingWeight } from './governance-delegation.service'; +import { + processDelegationRevokedEvents, + processDelegationSetEvents, +} from './governance-delegation-indexer.service'; +import { prisma } from '../../utils/prisma.utils'; +import { + cacheGetJson, + cacheInvalidate, + cacheSetJson, +} from '../../utils/redis.utils'; + +const mockCache = new Map(); + +jest.mock('../../utils/logger.utils', () => ({ + logger: { + info: jest.fn(), + warn: jest.fn(), + error: jest.fn(), + debug: jest.fn(), + }, +})); + +jest.mock('../../utils/redis.utils', () => ({ + cacheGetJson: jest.fn(async (key: string) => { + const raw = mockCache.get(key); + return raw === undefined ? null : JSON.parse(raw); + }), + cacheSetJson: jest.fn(async (key: string, value: unknown) => { + mockCache.set(key, JSON.stringify(value)); + }), + cacheInvalidate: jest.fn(async (...patterns: string[]) => { + for (const pattern of patterns) { + const prefix = pattern.replace(/\*$/, ''); + for (const k of Array.from(mockCache.keys())) { + if (k.startsWith(prefix)) mockCache.delete(k); + } + } + }), +})); + +jest.mock('../../utils/prisma.utils', () => ({ + prisma: { + keyOwnership: { + findUnique: jest.fn(), + findMany: jest.fn(), + }, + voteDelegation: { + findMany: jest.fn(), + findUnique: jest.fn(), + upsert: jest.fn(), + update: jest.fn(), + }, + voteDelegationHistory: { + findUnique: jest.fn(), + create: jest.fn(), + }, + activity: { + create: jest.fn(), + }, + $transaction: jest.fn(async (ops: Array>) => + Promise.all(ops) + ), + }, +})); + +const app = express(); +app.use(express.json()); +app.use('/governance', governanceRouter); + +const WALLET = 'GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAWHF'; +const DELEGATOR_1 = 'GBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBWHF'; +const DELEGATOR_2 = 'GCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCWHF'; + +beforeEach(() => { + jest.clearAllMocks(); + mockCache.clear(); +}); + +describe('GET /governance/voting-weight/:wallet/:key_id (#1176)', () => { + it('computes own_weight, delegated_weight, total, and delegators list accurately', async () => { + (prisma.keyOwnership.findUnique as jest.Mock).mockResolvedValue({ + balance: '40', + }); + (prisma.voteDelegation.findMany as jest.Mock).mockResolvedValue([ + { delegatorWallet: DELEGATOR_1 }, + { delegatorWallet: DELEGATOR_2 }, + ]); + (prisma.keyOwnership.findMany as jest.Mock).mockResolvedValue([ + { ownerAddress: DELEGATOR_1, balance: '25' }, + { ownerAddress: DELEGATOR_2, balance: '35' }, + ]); + + const res = await request(app).get( + `/governance/voting-weight/${WALLET}/key-1` + ); + + expect(res.status).toBe(200); + expect(res.body.success).toBe(true); + expect(res.body.data).toEqual({ + wallet: WALLET, + key_id: 'key-1', + own_weight: 40, + delegated_weight: 60, + total: 100, + delegators: [ + { wallet: DELEGATOR_1, amount: 25 }, + { wallet: DELEGATOR_2, amount: 35 }, + ], + }); + expect(cacheSetJson).toHaveBeenCalledWith( + `governance:voting-weight:${WALLET}:key-1`, + expect.any(Object), + 60 + ); + }); + + it('returns total: 0 for wallets with no holdings and no delegations', async () => { + (prisma.keyOwnership.findUnique as jest.Mock).mockResolvedValue(null); + (prisma.voteDelegation.findMany as jest.Mock).mockResolvedValue([]); + + const result = await getKeyGovernanceVotingWeight(WALLET, 'key-empty'); + + expect(result).toEqual({ + wallet: WALLET, + key_id: 'key-empty', + own_weight: 0, + delegated_weight: 0, + total: 0, + delegators: [], + }); + }); + + it('invalidates cached voting weight on VotingDelegated and DelegationRevoked events', async () => { + (prisma.voteDelegationHistory.findUnique as jest.Mock).mockResolvedValue( + null + ); + (prisma.voteDelegation.findUnique as jest.Mock).mockResolvedValue(null); + + await processDelegationSetEvents([ + { + eventType: 'VotingDelegated', + delegatorWallet: DELEGATOR_1, + delegateeWallet: WALLET, + keyId: 'key-1', + occurredAt: '2026-10-21T12:00:00.000Z', + ledger: 800, + txHash: 'tx-del-1', + eventIndex: 0, + }, + ]); + + expect(cacheInvalidate).toHaveBeenCalledWith( + expect.any(String), + expect.any(String), + expect.any(String), + expect.any(String), + `governance:voting-weight:${DELEGATOR_1}:*`, + `governance:voting-weight:${WALLET}:*` + ); + + (prisma.voteDelegation.findUnique as jest.Mock).mockResolvedValue({ + id: 'del-row-1', + delegateeWallet: WALLET, + isActive: true, + }); + + await processDelegationRevokedEvents([ + { + eventType: 'DelegationRevoked', + delegatorWallet: DELEGATOR_1, + keyId: 'key-1', + occurredAt: '2026-10-21T12:05:00.000Z', + ledger: 801, + txHash: 'tx-del-2', + eventIndex: 0, + }, + ]); + + expect(cacheGetJson).not.toThrow(); + expect(cacheInvalidate).toHaveBeenLastCalledWith( + expect.any(String), + expect.any(String), + expect.any(String), + expect.any(String), + `governance:voting-weight:${DELEGATOR_1}:*`, + `governance:voting-weight:${WALLET}:*` + ); + }); +}); diff --git a/src/modules/governance/governance.routes.ts b/src/modules/governance/governance.routes.ts index 8b191c70..96c8f09f 100644 --- a/src/modules/governance/governance.routes.ts +++ b/src/modules/governance/governance.routes.ts @@ -43,10 +43,50 @@ import { NonceAlreadyConsumedError, NonceMismatchError, } from '../nonce'; +import { cacheControl } from '../../middlewares/cache-control.middleware'; +import { + getKeyGovernanceVotingWeight, + VOTING_WEIGHT_CACHE_TTL_SECONDS, +} from './governance-delegation.service'; const router = Router(); router.use('/timelock', timelockRouter); +/** + * GET /governance/voting-weight/:wallet/:key_id + * + * Returns a wallet's total governance voting weight for a key including + * delegated weight (`own_weight`, `delegated_weight`, `total`, `delegators`) (#1176). + * Cached for 60s and invalidated on VotingDelegated and DelegationRevoked events. + */ +router.get( + '/voting-weight/:wallet/:key_id', + cacheControl({ + maxAge: VOTING_WEIGHT_CACHE_TTL_SECONDS, + type: 'public', + mustRevalidate: true, + }), + async (req, res, next) => { + const wallet = String(req.params.wallet ?? '').trim(); + const keyId = String(req.params.key_id ?? '').trim(); + + if (!wallet || !keyId) { + sendValidationError(res, 'Invalid path parameters', [ + ...(wallet ? [] : [{ field: 'wallet', message: 'wallet is required' }]), + ...(keyId ? [] : [{ field: 'key_id', message: 'key_id is required' }]), + ]); + return; + } + + try { + const result = await getKeyGovernanceVotingWeight(wallet, keyId); + sendSuccess(res, result); + } catch (error) { + next(error); + } + } +); + /** * GET /governance/proposals/escalating * diff --git a/src/modules/indexer/indexer-pipeline.service.ts b/src/modules/indexer/indexer-pipeline.service.ts index 99f53198..d4e146a3 100644 --- a/src/modules/indexer/indexer-pipeline.service.ts +++ b/src/modules/indexer/indexer-pipeline.service.ts @@ -600,6 +600,26 @@ export async function processIndexerEvents( await processHealthScoreEvents(healthScoreEvents); } + // Community poll vote & close events (#1174). + const { POLL_EVENT_TYPES, processPollEvents } = + await import('./poll-indexer.service'); + const pollEvents = events.filter(event => + POLL_EVENT_TYPES.has(event.eventType) + ); + if (pollEvents.length > 0) { + await processPollEvents(pollEvents); + } + + // Invoice instalment payment schedule (#1177). + const { INSTALMENT_PAID_EVENT_TYPES, processInstalmentPaidEvents } = + await import('./instalment-schedule-indexer.service'); + const instalmentEvents = events.filter(event => + INSTALMENT_PAID_EVENT_TYPES.has(event.eventType) + ); + if (instalmentEvents.length > 0) { + await processInstalmentPaidEvents(instalmentEvents); + } + if (options.advanceCursor !== false) await advanceIndexerCursorForBatch(events); } diff --git a/src/modules/indexer/instalment-schedule-indexer.service.test.ts b/src/modules/indexer/instalment-schedule-indexer.service.test.ts new file mode 100644 index 00000000..d02a4ebd --- /dev/null +++ b/src/modules/indexer/instalment-schedule-indexer.service.test.ts @@ -0,0 +1,107 @@ +import { processInstalmentPaidEvents } from './instalment-schedule-indexer.service'; +import { prisma } from '../../utils/prisma.utils'; +import { cacheInvalidate } from '../../utils/redis.utils'; + +jest.mock('../../utils/logger.utils', () => ({ + logger: { + info: jest.fn(), + warn: jest.fn(), + error: jest.fn(), + debug: jest.fn(), + }, +})); + +jest.mock('../../utils/redis.utils', () => ({ + cacheInvalidate: jest.fn().mockResolvedValue(undefined), +})); + +jest.mock('../../utils/prisma.utils', () => ({ + prisma: { + instalmentPayment: { + findUnique: jest.fn(), + upsert: jest.fn(), + }, + }, +})); + +const base = { + txHash: 'tx-inst-1', + eventIndex: 1, + ledger: 900, + timestamp: 1_800_000_000, +}; + +beforeEach(() => { + jest.clearAllMocks(); +}); + +describe('processInstalmentPaidEvents (#1177)', () => { + it('indexes InstalmentPaid event into instalment_payments and invalidates cache', async () => { + (prisma.instalmentPayment.findUnique as jest.Mock).mockResolvedValue( + null + ); + (prisma.instalmentPayment.upsert as jest.Mock).mockResolvedValue({ + id: 'inst-1', + }); + + await processInstalmentPaidEvents([ + { + ...base, + eventType: 'InstalmentPaid', + invoice_id: 'inv-1', + instalment_number: 2, + total_instalments: 4, + amount: '500.0000000', + due_date: '2026-10-25T00:00:00.000Z', + payer: 'GPAYER', + }, + ]); + + expect(prisma.instalmentPayment.upsert).toHaveBeenCalledWith( + expect.objectContaining({ + where: { + invoiceId_instalmentNumber: { + invoiceId: 'inv-1', + instalmentNumber: 2, + }, + }, + create: expect.objectContaining({ + invoiceId: 'inv-1', + instalmentNumber: 2, + totalInstalments: 4, + amount: '500.0000000', + payer: 'GPAYER', + txHash: 'tx-inst-1', + eventIndex: 1, + ledger: 900, + }), + }) + ); + expect(cacheInvalidate).toHaveBeenCalledWith( + 'invoice:schedule:inv-1', + 'invoice:schedule:next:inv-1' + ); + }); + + it('skips duplicate replayed event by (txHash, eventIndex)', async () => { + (prisma.instalmentPayment.findUnique as jest.Mock).mockResolvedValue({ + id: 'existing-id', + }); + + await processInstalmentPaidEvents([ + { + ...base, + eventType: 'InstalmentPaid', + invoice_id: 'inv-1', + instalment_number: 1, + amount: '100', + }, + ]); + + expect(prisma.instalmentPayment.upsert).not.toHaveBeenCalled(); + expect(cacheInvalidate).toHaveBeenCalledWith( + 'invoice:schedule:inv-1', + 'invoice:schedule:next:inv-1' + ); + }); +}); diff --git a/src/modules/indexer/instalment-schedule-indexer.service.ts b/src/modules/indexer/instalment-schedule-indexer.service.ts new file mode 100644 index 00000000..c5fb4087 --- /dev/null +++ b/src/modules/indexer/instalment-schedule-indexer.service.ts @@ -0,0 +1,137 @@ +// src/modules/indexer/instalment-schedule-indexer.service.ts +// Indexes InstalmentPaid contract events into instalment_payments (#1177). +// Each event is idempotent by chain position (txHash + eventIndex) and +// invalidates the 60s invoice schedule cache. + +import { z } from 'zod'; +import { prisma } from '../../utils/prisma.utils'; +import { logger } from '../../utils/logger.utils'; +import { + IndexerChainEvent, + processIndexerChainEvents, +} from '../../utils/indexer-event-processor.utils'; +import { + chainPositionSchema, + eventIdentifierSchema, + eventOccurredAt, +} from '../../utils/indexer-event-fields.utils'; +import { invalidateInvoiceScheduleCache } from '../invoice/invoice-schedule.service'; + +export const INSTALMENT_PAID_EVENT_TYPES: ReadonlySet = new Set([ + 'InstalmentPaid', + 'INSTALMENT_PAID', +]); + +const instalmentPaidPayload = z.object({ + invoiceId: eventIdentifierSchema, + instalmentNumber: z.coerce.number().int().min(1).max(10_000), + totalInstalments: z.coerce.number().int().min(1).max(10_000).optional(), + amount: z.union([z.string().min(1), z.number().positive(), z.bigint()]), + payer: eventIdentifierSchema.nullable().optional(), +}); + +function parseOptionalDate(raw: unknown): Date | null { + if (raw === undefined || raw === null || raw === '') return null; + const numeric = + typeof raw === 'number' || + typeof raw === 'bigint' || + (typeof raw === 'string' && /^\d+$/.test(raw)); + const date = numeric + ? new Date(Number(raw) * 1000) + : new Date(String(raw)); + return Number.isNaN(date.getTime()) ? null : date; +} + +export async function processInstalmentPaidEvents( + events: IndexerChainEvent[] +): Promise { + await processIndexerChainEvents(events, async event => { + if (!INSTALMENT_PAID_EVENT_TYPES.has(event.eventType)) { + return; + } + + const raw = event as Record; + const position = chainPositionSchema.safeParse(event); + const payload = instalmentPaidPayload.safeParse({ + invoiceId: raw.invoice_id ?? raw.invoiceId, + instalmentNumber: + raw.instalment_number ?? + raw.instalmentNumber ?? + raw.instalment_index ?? + raw.index ?? + 1, + totalInstalments: + raw.total_instalments ?? raw.totalInstalments ?? undefined, + amount: raw.amount, + payer: raw.payer ?? raw.actor ?? raw.wallet, + }); + + if (!position.success || !payload.success) { + logger.warn( + { + eventId: `${event.txHash}:${event.eventIndex}`, + eventType: event.eventType, + }, + 'Skipping InstalmentPaid event with missing or invalid fields' + ); + return; + } + + // Idempotency check by (txHash, eventIndex). + const existingByTx = await prisma.instalmentPayment.findUnique({ + where: { + txHash_eventIndex: { + txHash: position.data.txHash, + eventIndex: position.data.eventIndex, + }, + }, + select: { id: true }, + }); + + if (existingByTx) { + await invalidateInvoiceScheduleCache(payload.data.invoiceId); + return; + } + + const occurredAt = eventOccurredAt(event); + const paidAt = + parseOptionalDate(raw.paid_at ?? raw.paidAt) ?? occurredAt; + const dueDate = + parseOptionalDate(raw.due_date ?? raw.dueDate) ?? occurredAt; + const amountStr = String(payload.data.amount); + + await prisma.instalmentPayment.upsert({ + where: { + invoiceId_instalmentNumber: { + invoiceId: payload.data.invoiceId, + instalmentNumber: payload.data.instalmentNumber, + }, + }, + update: { + amount: amountStr, + paidAt, + payer: payload.data.payer ?? null, + txHash: position.data.txHash, + eventIndex: position.data.eventIndex, + ledger: position.data.ledger, + ...(payload.data.totalInstalments !== undefined + ? { totalInstalments: payload.data.totalInstalments } + : {}), + }, + create: { + invoiceId: payload.data.invoiceId, + instalmentNumber: payload.data.instalmentNumber, + totalInstalments: payload.data.totalInstalments ?? null, + amount: amountStr, + dueDate, + paidAt, + payer: payload.data.payer ?? null, + txHash: position.data.txHash, + eventIndex: position.data.eventIndex, + ledger: position.data.ledger, + }, + }); + + await invalidateInvoiceScheduleCache(payload.data.invoiceId); + }); +} diff --git a/src/modules/indexer/poll-indexer.service.ts b/src/modules/indexer/poll-indexer.service.ts new file mode 100644 index 00000000..f3f187cb --- /dev/null +++ b/src/modules/indexer/poll-indexer.service.ts @@ -0,0 +1,152 @@ +// src/modules/indexer/poll-indexer.service.ts +// Indexes PollVoteCast and PollClosed contract events (#1174) and invalidates +// the 30s key community poll caches. + +import { Decimal } from '@prisma/client/runtime/library'; +import { prisma } from '../../utils/prisma.utils'; +import { logger } from '../../utils/logger.utils'; +import { + IndexerChainEvent, + processIndexerChainEvents, +} from '../../utils/indexer-event-processor.utils'; +import { eventOccurredAt } from '../../utils/indexer-event-fields.utils'; +import { applyVoteWeight } from '../../utils/vote-tally.utils'; +import { invalidateKeyPollsCache } from '../keys/key-polls.service'; + +export const POLL_EVENT_TYPES: ReadonlySet = new Set([ + 'PollVoteCast', + 'POLL_VOTE_CAST', + 'PollClosed', + 'POLL_CLOSED', +]); + +function readString( + raw: Record, + ...names: string[] +): string | null { + for (const name of names) { + const value = raw[name]; + if (typeof value === 'string' && value.trim() !== '') { + return value.trim(); + } + } + return null; +} + +export async function processPollEvents( + events: IndexerChainEvent[] +): Promise { + await processIndexerChainEvents(events, async event => { + if (!POLL_EVENT_TYPES.has(event.eventType)) { + return; + } + + const raw = event as Record; + const keyId = readString(raw, 'keyId', 'key_id', 'creatorId', 'creator_id'); + const pollId = readString( + raw, + 'pollId', + 'poll_id', + 'proposalId', + 'proposal_id' + ); + + if (!keyId || !pollId) { + logger.warn( + { + eventId: `${event.txHash}:${event.eventIndex}`, + eventType: event.eventType, + }, + 'Skipping poll event with missing keyId or pollId' + ); + return; + } + + const isClosedEvent = + event.eventType === 'PollClosed' || event.eventType === 'POLL_CLOSED'; + + if (isClosedEvent) { + const occurredAt = eventOccurredAt(event); + const existing = await prisma.governanceProposal.findFirst({ + where: { keyId, proposalId: pollId }, + select: { id: true, status: true }, + }); + if (existing && existing.status === 'active') { + await prisma.governanceProposal.update({ + where: { id: existing.id }, + data: { + status: 'closed', + closedAt: occurredAt, + }, + }); + } + await invalidateKeyPollsCache(keyId); + return; + } + + // PollVoteCast / POLL_VOTE_CAST + const voter = readString(raw, 'voter', 'wallet', 'actor'); + const optionIndex = Number( + raw.optionIndex ?? raw.option_index ?? raw.option ?? 0 + ); + const weightStr = String(raw.weight ?? raw.vote_weight ?? '1'); + + if (voter && Number.isInteger(optionIndex) && optionIndex >= 0) { + const proposal = await prisma.governanceProposal.findFirst({ + where: { keyId, proposalId: pollId }, + }); + + if (proposal) { + const existingVote = await prisma.governanceVote.findUnique({ + where: { + keyId_proposalId_voter: { + keyId, + proposalId: pollId, + voter, + }, + }, + }); + + if (!existingVote) { + const options = Array.isArray(proposal.options) + ? (proposal.options as string[]) + : []; + const optionLabel = + options[optionIndex] ?? `option_${optionIndex}`; + const updated = applyVoteWeight( + { + totalVotingWeight: proposal.totalVotingWeight, + results: (proposal.results ?? {}) as Record< + string, + string + >, + }, + weightStr, + optionLabel + ); + + await prisma.$transaction([ + prisma.governanceProposal.update({ + where: { id: proposal.id }, + data: { + totalVotingWeight: updated.totalVotingWeight, + results: updated.results, + }, + }), + prisma.governanceVote.create({ + data: { + keyId, + proposalId: pollId, + voter, + optionIndex, + weight: new Decimal(weightStr), + }, + }), + ]); + } + } + } + + await invalidateKeyPollsCache(keyId); + }); +} diff --git a/src/modules/indexer/tip-indexer.service.ts b/src/modules/indexer/tip-indexer.service.ts index 774f7ae9..e4ce1c6b 100644 --- a/src/modules/indexer/tip-indexer.service.ts +++ b/src/modules/indexer/tip-indexer.service.ts @@ -39,6 +39,7 @@ export interface ParsedTipEvent { tipper: string; /** Amount in XLM (converted from stroops). */ amountXlm: string; + anonymous?: boolean; messageHash: string | null; tippedAt: Date; ledger: number; @@ -96,7 +97,9 @@ export function parseTipEvent(event: IndexerChainEvent): TipParseResult { return { ok: false, reason: 'missing or invalid key/creator id' }; } - const tipper = readStringField(raw, 'tipper', 'sender', 'wallet', 'actor'); + const anonymous = Boolean(raw.anonymous ?? raw.is_anonymous ?? false); + const rawTipper = readStringField(raw, 'tipper', 'sender', 'wallet', 'actor'); + const tipper = rawTipper ?? (anonymous ? 'anonymous' : null); if (!tipper) { return { ok: false, reason: 'missing or invalid tipper' }; } @@ -151,6 +154,7 @@ export function parseTipEvent(event: IndexerChainEvent): TipParseResult { keyId, tipper, amountXlm: stroopsToXlmDecimalString(amount), + ...(anonymous ? { anonymous: true } : {}), messageHash, tippedAt, ledger, @@ -201,6 +205,7 @@ export async function processTipEvent( keyId: value.keyId, tipper: value.tipper, amount: value.amountXlm, + ...(value.anonymous ? { anonymous: true } : {}), messageHash: value.messageHash, ledger: value.ledger, txHash: value.txHash, diff --git a/src/modules/invoice/invoice-schedule.routes.ts b/src/modules/invoice/invoice-schedule.routes.ts new file mode 100644 index 00000000..d3d73c68 --- /dev/null +++ b/src/modules/invoice/invoice-schedule.routes.ts @@ -0,0 +1,84 @@ +// src/modules/invoice/invoice-schedule.routes.ts +// GET /invoices/:invoice_id/schedule and GET /invoices/:invoice_id/schedule/next (#1177). + +import { Router } from 'express'; +import { z } from 'zod'; +import { cacheControl } from '../../middlewares/cache-control.middleware'; +import { + sendNotFound, + sendSuccess, + sendValidationError, + zodIssuesToDetails, +} from '../../utils/api-response.utils'; +import { + getInvoiceSchedule, + getNextInvoiceInstalment, + INVOICE_SCHEDULE_CACHE_TTL_SECONDS, +} from './invoice-schedule.service'; + +export const invoiceScheduleRouter = Router(); + +const paramsSchema = z.object({ + invoice_id: z.string().min(1).max(128), +}); + +const scheduleCache = () => + cacheControl({ + maxAge: INVOICE_SCHEDULE_CACHE_TTL_SECONDS, + type: 'public', + mustRevalidate: true, + }); + +invoiceScheduleRouter.get( + '/:invoice_id/schedule/next', + scheduleCache(), + async (req, res, next) => { + const params = paramsSchema.safeParse(req.params); + if (!params.success) { + sendValidationError( + res, + 'Invalid invoice id', + zodIssuesToDetails(params.error.issues) + ); + return; + } + try { + const nextInstalment = await getNextInvoiceInstalment( + params.data.invoice_id + ); + if (!nextInstalment) { + sendNotFound(res, 'Invoice'); + return; + } + sendSuccess(res, nextInstalment); + } catch (error) { + next(error); + } + } +); + +invoiceScheduleRouter.get( + '/:invoice_id/schedule', + scheduleCache(), + async (req, res, next) => { + const params = paramsSchema.safeParse(req.params); + if (!params.success) { + sendValidationError( + res, + 'Invalid invoice id', + zodIssuesToDetails(params.error.issues) + ); + return; + } + try { + const schedule = await getInvoiceSchedule(params.data.invoice_id); + if (!schedule) { + sendNotFound(res, 'Invoice'); + return; + } + sendSuccess(res, schedule); + } catch (error) { + next(error); + } + } +); diff --git a/src/modules/invoice/invoice-schedule.service.test.ts b/src/modules/invoice/invoice-schedule.service.test.ts new file mode 100644 index 00000000..0f7b1856 --- /dev/null +++ b/src/modules/invoice/invoice-schedule.service.test.ts @@ -0,0 +1,150 @@ +import express from 'express'; +import request from 'supertest'; +import { invoiceScheduleRouter } from './invoice-schedule.routes'; +import { + buildInvoiceSchedule, + computeInstalmentStatus, + getInvoiceSchedule, + getNextInvoiceInstalment, +} from './invoice-schedule.service'; +import { prisma } from '../../utils/prisma.utils'; +import { cacheSetJson } from '../../utils/redis.utils'; + +const mockCache = new Map(); + +jest.mock('../../utils/redis.utils', () => ({ + cacheGetJson: jest.fn(async (key: string) => { + const raw = mockCache.get(key); + return raw === undefined ? null : JSON.parse(raw); + }), + cacheSetJson: jest.fn(async (key: string, value: unknown) => { + mockCache.set(key, JSON.stringify(value)); + }), + cacheInvalidate: jest.fn(async (...keys: string[]) => { + for (const k of keys) mockCache.delete(k); + }), +})); + +jest.mock('../../utils/prisma.utils', () => ({ + prisma: { + invoice: { + findUnique: jest.fn(), + }, + instalmentPayment: { + findMany: jest.fn(), + }, + }, +})); + +const app = express(); +app.use(express.json()); +app.use('/invoices', invoiceScheduleRouter); + +const NOW = new Date('2026-10-21T12:00:00.000Z'); +const PAST_DUE = new Date('2026-10-15T12:00:00.000Z'); +const FUTURE_DUE = new Date('2026-10-28T12:00:00.000Z'); +const PAID_AT = new Date('2026-10-10T09:00:00.000Z'); + +beforeEach(() => { + jest.clearAllMocks(); + mockCache.clear(); +}); + +describe('invoice-schedule.service & routes (#1177)', () => { + it('computes paid, overdue, and pending status accurately at request time', () => { + expect(computeInstalmentStatus(PAST_DUE, PAID_AT, NOW)).toBe('paid'); + expect(computeInstalmentStatus(PAST_DUE, null, NOW)).toBe('overdue'); + expect(computeInstalmentStatus(FUTURE_DUE, null, NOW)).toBe('pending'); + }); + + it('builds schedule with paid, overdue, and pending counts', () => { + const schedule = buildInvoiceSchedule( + 'inv-1', + [ + { + id: 'inst-1', + instalmentNumber: 1, + amount: '100', + dueDate: new Date('2026-10-10T00:00:00.000Z'), + paidAt: PAID_AT, + txHash: 'tx-1', + }, + { + id: 'inst-2', + instalmentNumber: 2, + amount: '100', + dueDate: PAST_DUE, + paidAt: null, + txHash: null, + }, + { + id: 'inst-3', + instalmentNumber: 3, + amount: '100', + dueDate: FUTURE_DUE, + paidAt: null, + txHash: null, + }, + ], + NOW + ); + + expect(schedule.paid_count).toBe(1); + expect(schedule.overdue_count).toBe(1); + expect(schedule.pending_count).toBe(1); + expect(schedule.instalments.map(i => i.status)).toEqual([ + 'paid', + 'overdue', + 'pending', + ]); + }); + + it('returns schedule and next due instalment via REST and caches for 60s', async () => { + (prisma.invoice.findUnique as jest.Mock).mockResolvedValue({ + id: 'inv-1', + }); + (prisma.instalmentPayment.findMany as jest.Mock).mockResolvedValue([ + { + id: 'inst-1', + instalmentNumber: 1, + amount: '250.0000000', + dueDate: PAST_DUE, + paidAt: PAID_AT, + txHash: 'tx-1', + }, + { + id: 'inst-2', + instalmentNumber: 2, + amount: '250.0000000', + dueDate: FUTURE_DUE, + paidAt: null, + txHash: null, + }, + ]); + + const scheduleRes = await request(app).get('/invoices/inv-1/schedule'); + expect(scheduleRes.status).toBe(200); + expect(scheduleRes.body.data.instalments).toHaveLength(2); + expect(cacheSetJson).toHaveBeenCalledWith( + 'invoice:schedule:inv-1', + expect.any(Object), + 60 + ); + + const nextRes = await request(app).get('/invoices/inv-1/schedule/next'); + expect(nextRes.status).toBe(200); + expect(nextRes.body.data.instalment_number).toBe(2); + expect(nextRes.body.data.next_instalment.instalment_number).toBe(2); + }); + + it('returns 404 when invoice does not exist', async () => { + (prisma.invoice.findUnique as jest.Mock).mockResolvedValue(null); + (prisma.instalmentPayment.findMany as jest.Mock).mockResolvedValue([]); + + expect(await getInvoiceSchedule('unknown')).toBeNull(); + expect(await getNextInvoiceInstalment('unknown')).toBeNull(); + + const res = await request(app).get('/invoices/unknown/schedule'); + expect(res.status).toBe(404); + }); +}); diff --git a/src/modules/invoice/invoice-schedule.service.ts b/src/modules/invoice/invoice-schedule.service.ts new file mode 100644 index 00000000..7b3e0c49 --- /dev/null +++ b/src/modules/invoice/invoice-schedule.service.ts @@ -0,0 +1,234 @@ +// src/modules/invoice/invoice-schedule.service.ts +// Invoice payment schedule and next-instalment read model (#1177). +// +// - GET /invoices/:invoice_id/schedule — all instalments with status, due_date, paid_at +// - GET /invoices/:invoice_id/schedule/next — next due unpaid instalment +// - Overdue status is computed at request time for unpaid instalments past due_date +// - Cached in Redis for 60s and invalidated on InstalmentPaid events + +import { prisma } from '../../utils/prisma.utils'; +import { + cacheGetJson, + cacheInvalidate, + cacheSetJson, +} from '../../utils/redis.utils'; + +export const INVOICE_SCHEDULE_CACHE_TTL_SECONDS = 60; + +export type InstalmentStatus = 'paid' | 'pending' | 'overdue'; + +export interface InvoiceInstalmentItem { + id: string; + invoice_id: string; + instalment_number: number; + amount: string; + status: InstalmentStatus; + due_date: string; + paid_at: string | null; + tx_hash: string | null; +} + +export interface InvoicePaymentSchedule { + invoice_id: string; + total_instalments: number; + paid_count: number; + pending_count: number; + overdue_count: number; + instalments: InvoiceInstalmentItem[]; +} + +export interface InvoiceNextInstalmentResponse { + invoice_id: string; + next_instalment: InvoiceInstalmentItem | null; + id?: string; + instalment_number?: number; + amount?: string; + status?: InstalmentStatus; + due_date?: string; + paid_at?: string | null; + tx_hash?: string | null; +} + +export function invoiceScheduleCacheKey(invoiceId: string): string { + return `invoice:schedule:${invoiceId}`; +} + +export function invoiceScheduleNextCacheKey(invoiceId: string): string { + return `invoice:schedule:next:${invoiceId}`; +} + +export async function invalidateInvoiceScheduleCache( + invoiceId: string +): Promise { + await cacheInvalidate( + invoiceScheduleCacheKey(invoiceId), + invoiceScheduleNextCacheKey(invoiceId) + ); +} + +/** + * Computes instalment status at request time: + * - 'paid' when paidAt is present + * - 'overdue' when unpaid and dueDate is strictly before `now` + * - 'pending' when unpaid and dueDate is at or after `now` + */ +export function computeInstalmentStatus( + dueDate: Date | string, + paidAt: Date | string | null | undefined, + now: Date = new Date() +): InstalmentStatus { + if (paidAt !== null && paidAt !== undefined) { + return 'paid'; + } + const dueMs = + dueDate instanceof Date ? dueDate.getTime() : new Date(dueDate).getTime(); + return dueMs < now.getTime() ? 'overdue' : 'pending'; +} + +function recomputeScheduleStatuses( + schedule: InvoicePaymentSchedule, + now: Date = new Date() +): InvoicePaymentSchedule { + let paidCount = 0; + let pendingCount = 0; + let overdueCount = 0; + + const instalments = schedule.instalments.map(item => { + const status = computeInstalmentStatus(item.due_date, item.paid_at, now); + if (status === 'paid') paidCount += 1; + else if (status === 'overdue') overdueCount += 1; + else pendingCount += 1; + return { + ...item, + status, + }; + }); + + return { + invoice_id: schedule.invoice_id, + total_instalments: instalments.length, + paid_count: paidCount, + pending_count: pendingCount, + overdue_count: overdueCount, + instalments, + }; +} + +export function buildInvoiceSchedule( + invoiceId: string, + rows: Array<{ + id: string; + instalmentNumber: number; + amount: unknown; + dueDate: Date; + paidAt: Date | null; + txHash?: string | null; + }>, + now: Date = new Date() +): InvoicePaymentSchedule { + const sorted = [...rows].sort((a, b) => { + const diff = a.dueDate.getTime() - b.dueDate.getTime(); + if (diff !== 0) return diff; + return a.instalmentNumber - b.instalmentNumber; + }); + + const baseInstalments: InvoiceInstalmentItem[] = sorted.map(row => ({ + id: row.id, + invoice_id: invoiceId, + instalment_number: row.instalmentNumber, + amount: Number(row.amount).toFixed(7), + status: computeInstalmentStatus(row.dueDate, row.paidAt, now), + due_date: row.dueDate.toISOString(), + paid_at: row.paidAt ? row.paidAt.toISOString() : null, + tx_hash: row.txHash ?? null, + })); + + return recomputeScheduleStatuses( + { + invoice_id: invoiceId, + total_instalments: baseInstalments.length, + paid_count: 0, + pending_count: 0, + overdue_count: 0, + instalments: baseInstalments, + }, + now + ); +} + +/** + * Returns the full payment schedule for an invoice, or null when the invoice + * does not exist and has no indexed instalments. + */ +export async function getInvoiceSchedule( + invoiceId: string, + now: Date = new Date() +): Promise { + const cacheKey = invoiceScheduleCacheKey(invoiceId); + try { + const cached = await cacheGetJson(cacheKey); + if (cached !== null) { + return recomputeScheduleStatuses(cached, now); + } + } catch { + // Cache read failure falls through to database. + } + + const [invoice, rows] = await Promise.all([ + prisma.invoice.findUnique({ + where: { id: invoiceId }, + select: { id: true }, + }), + prisma.instalmentPayment.findMany({ + where: { invoiceId }, + orderBy: [{ dueDate: 'asc' }, { instalmentNumber: 'asc' }], + }), + ]); + + if (!invoice && rows.length === 0) { + return null; + } + + const schedule = buildInvoiceSchedule(invoiceId, rows, now); + try { + await cacheSetJson( + cacheKey, + schedule, + INVOICE_SCHEDULE_CACHE_TTL_SECONDS + ); + } catch { + // Cache write failure is non-critical. + } + return schedule; +} + +/** + * Returns the next upcoming unpaid instalment (earliest due unpaid instalment), + * or `{ invoice_id, next_instalment: null }` if all instalments are paid. + * Returns null if the invoice does not exist. + */ +export async function getNextInvoiceInstalment( + invoiceId: string, + now: Date = new Date() +): Promise { + const schedule = await getInvoiceSchedule(invoiceId, now); + if (!schedule) { + return null; + } + + const nextItem = + schedule.instalments.find(item => item.status !== 'paid') ?? null; + + if (!nextItem) { + return { + invoice_id: invoiceId, + next_instalment: null, + }; + } + + return { + ...nextItem, + invoice_id: invoiceId, + next_instalment: nextItem, + }; +} diff --git a/src/modules/invoice/invoice.routes.ts b/src/modules/invoice/invoice.routes.ts index 9a9bf840..f4669024 100644 --- a/src/modules/invoice/invoice.routes.ts +++ b/src/modules/invoice/invoice.routes.ts @@ -6,6 +6,7 @@ import { httpGetInvoiceComparison } from './invoice.controllers'; import { invoiceProtocolRouter } from './invoice-protocol.routes'; import { invoiceMarketRouter } from './invoice-market.routes'; import { invoiceEarlyIncentiveRouter } from './invoice-early-incentive.routes'; +import { invoiceScheduleRouter } from './invoice-schedule.routes'; const invoiceRouter = Router(); @@ -27,5 +28,6 @@ invoiceRouter.get( invoiceRouter.use(invoiceProtocolRouter); invoiceRouter.use(invoiceMarketRouter); invoiceRouter.use(invoiceEarlyIncentiveRouter); +invoiceRouter.use(invoiceScheduleRouter); export default invoiceRouter; diff --git a/src/modules/keys/key-polls.service.test.ts b/src/modules/keys/key-polls.service.test.ts new file mode 100644 index 00000000..6df2de5f --- /dev/null +++ b/src/modules/keys/key-polls.service.test.ts @@ -0,0 +1,207 @@ +import { + getKeyPollDetail, + invalidateKeyPollsCache, + listKeyPolls, + PollNotFoundError, +} from './key-polls.service'; +import { processPollEvents } from '../indexer/poll-indexer.service'; +import { prisma } from '../../utils/prisma.utils'; +import { + cacheGetJson, + cacheInvalidate, + cacheSetJson, +} from '../../utils/redis.utils'; + +const mockCache = new Map(); + +jest.mock('../../utils/logger.utils', () => ({ + logger: { + info: jest.fn(), + warn: jest.fn(), + error: jest.fn(), + debug: jest.fn(), + }, +})); + +jest.mock('../../utils/redis.utils', () => ({ + cacheGetJson: jest.fn(async (key: string) => { + const val = mockCache.get(key); + return val === undefined ? null : JSON.parse(val); + }), + cacheSetJson: jest.fn(async (key: string, value: unknown) => { + mockCache.set(key, JSON.stringify(value)); + }), + cacheInvalidate: jest.fn(async (...patterns: string[]) => { + for (const pattern of patterns) { + const prefix = pattern.replace(/\*$/, ''); + for (const k of Array.from(mockCache.keys())) { + if (k.startsWith(prefix)) mockCache.delete(k); + } + } + }), +})); + +jest.mock('../../utils/prisma.utils', () => ({ + prisma: { + governanceProposal: { + findMany: jest.fn(), + count: jest.fn(), + findFirst: jest.fn(), + update: jest.fn(), + }, + governanceVote: { + findUnique: jest.fn(), + create: jest.fn(), + }, + $transaction: jest.fn(async (ops: Array>) => + Promise.all(ops) + ), + }, +})); + +const NOW = new Date('2026-10-21T12:00:00.000Z'); +const FUTURE = new Date('2026-10-21T13:00:00.000Z'); +const PAST = new Date('2026-10-21T10:00:00.000Z'); +const WALLET = 'GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAWHF'; + +beforeEach(() => { + jest.clearAllMocks(); + mockCache.clear(); +}); + +describe('key-polls.service (#1174)', () => { + it('lists active and closed polls with tally and time remaining', async () => { + (prisma.governanceProposal.findMany as jest.Mock).mockResolvedValue([ + { + keyId: 'key-1', + proposalId: 'poll-1', + title: 'Next Roadmap Item', + options: ['Feature A', 'Feature B'], + totalVotingWeight: '100', + results: { 'Feature A': '75', 'Feature B': '25' }, + snapshotLedger: 500, + expiresAt: FUTURE, + closedAt: null, + status: 'active', + }, + { + keyId: 'key-1', + proposalId: 'poll-2', + title: 'Past Poll', + options: ['Yes', 'No'], + totalVotingWeight: '50', + results: { Yes: '50', No: '0' }, + snapshotLedger: 400, + expiresAt: PAST, + closedAt: PAST, + status: 'passed', + }, + ]); + (prisma.governanceProposal.count as jest.Mock).mockResolvedValue(2); + + const page = await listKeyPolls('key-1', { page: 1, limit: 10, now: NOW }); + + expect(page.polls).toHaveLength(2); + expect(page.polls[0]).toMatchObject({ + poll_id: 'poll-1', + status: 'active', + time_remaining_seconds: 3600, + tally: { 'Feature A': '75', 'Feature B': '25' }, + }); + expect(page.polls[1]).toMatchObject({ + poll_id: 'poll-2', + status: 'closed', + time_remaining_seconds: 0, + }); + expect(cacheSetJson).toHaveBeenCalledWith( + 'key:polls:key-1:list:all:1:10', + expect.any(Object), + 30 + ); + }); + + it('returns single poll detail with per-option results and wallet vote', async () => { + (prisma.governanceProposal.findFirst as jest.Mock).mockResolvedValue({ + keyId: 'key-1', + proposalId: 'poll-1', + title: 'Next Roadmap Item', + options: ['Feature A', 'Feature B'], + totalVotingWeight: '100', + results: { 'Feature A': '75', 'Feature B': '25' }, + snapshotLedger: 500, + expiresAt: FUTURE, + closedAt: null, + status: 'active', + }); + (prisma.governanceVote.findUnique as jest.Mock).mockResolvedValue({ + voter: WALLET, + optionIndex: 0, + weight: '75', + }); + + const detail = await getKeyPollDetail('key-1', 'poll-1', WALLET, NOW); + + expect(detail.option_results).toEqual([ + { index: 0, option: 'Feature A', votes: '75', percentage: 75 }, + { index: 1, option: 'Feature B', votes: '25', percentage: 25 }, + ]); + expect(detail.wallet_vote).toEqual({ + voter: WALLET, + option_index: 0, + option: 'Feature A', + weight: '75', + }); + expect(cacheGetJson).toHaveBeenCalled(); + }); + + it('throws PollNotFoundError (404) for unknown poll IDs', async () => { + (prisma.governanceProposal.findFirst as jest.Mock).mockResolvedValue(null); + + await expect( + getKeyPollDetail('key-1', 'unknown-poll') + ).rejects.toBeInstanceOf(PollNotFoundError); + }); + + it('invalidates poll cache on PollVoteCast and PollClosed events', async () => { + (prisma.governanceProposal.findFirst as jest.Mock).mockResolvedValue({ + id: 'row-1', + keyId: 'key-1', + proposalId: 'poll-1', + options: ['Yes', 'No'], + totalVotingWeight: '10', + results: { Yes: '10', No: '0' }, + status: 'active', + }); + (prisma.governanceVote.findUnique as jest.Mock).mockResolvedValue(null); + + await processPollEvents([ + { + eventType: 'PollVoteCast', + key_id: 'key-1', + poll_id: 'poll-1', + voter: WALLET, + option_index: 1, + weight: '5', + txHash: 'tx-poll-vote', + eventIndex: 0, + ledger: 600, + }, + { + eventType: 'PollClosed', + key_id: 'key-1', + poll_id: 'poll-1', + txHash: 'tx-poll-close', + eventIndex: 1, + ledger: 601, + }, + ]); + + expect(cacheInvalidate).toHaveBeenCalledWith( + 'key:polls:key-1:*', + 'key:poll:weight:key-1:*' + ); + + await invalidateKeyPollsCache('key-1'); + expect(cacheInvalidate).toHaveBeenCalledTimes(3); + }); +}); diff --git a/src/modules/keys/key-polls.service.ts b/src/modules/keys/key-polls.service.ts new file mode 100644 index 00000000..d91b1909 --- /dev/null +++ b/src/modules/keys/key-polls.service.ts @@ -0,0 +1,295 @@ +// src/modules/keys/key-polls.service.ts +// +// Read model for key community polls (#1174), backed by GovernanceProposal +// and GovernanceVote rows indexed from on-chain poll events. +// +// - GET /keys/:key_id/polls — paginated active & closed polls with tally and time remaining +// - GET /keys/:key_id/polls/:poll_id — single poll with per-option results and optional wallet vote +// - GET /keys/:key_id/polls/:poll_id/weight/:wallet — wallet vote weight from snapshot (key-poll-weight.service.ts) +// +// Responses are cached in Redis for 30s and invalidated on PollVoteCast and +// PollClosed events. + +import { prisma } from '../../utils/prisma.utils'; +import { + cacheGetJson, + cacheInvalidate, + cacheSetJson, +} from '../../utils/redis.utils'; +import { + KEY_POLLS_CACHE_TTL_SECONDS, + keyPollDetailCacheKey, + keyPollsCachePattern, + keyPollsListCacheKey, +} from '../../constants/redis.constants'; +import { buildPaginationMetadata } from '../../utils/indexer-event-fields.utils'; +import { PaginationMetadata } from '../../utils/api-response.utils'; +import { PollNotFoundError } from './key-poll-weight.service'; + +export { PollNotFoundError }; + +export type PollFilterStatus = 'active' | 'closed'; + +export interface KeyPollOptionResult { + index: number; + option: string; + votes: string; + percentage: number; +} + +export interface KeyPollWalletVote { + voter: string; + option_index: number; + option: string; + weight: string; +} + +export interface KeyPollItem { + poll_id: string; + key_id: string; + title: string; + options: string[]; + status: 'active' | 'closed'; + raw_status: string; + tally: Record; + results: Record; + option_results: KeyPollOptionResult[]; + total_voting_weight: string; + snapshot_block: number; + expires_at: string; + closed_at: string | null; + time_remaining_seconds: number; +} + +export interface KeyPollDetail extends KeyPollItem { + wallet_vote: KeyPollWalletVote | null; +} + +export interface KeyPollsPage { + key_id: string; + polls: KeyPollItem[]; + items: KeyPollItem[]; + pagination: PaginationMetadata; +} + +export async function invalidateKeyPollsCache(keyId: string): Promise { + await cacheInvalidate( + keyPollsCachePattern(keyId), + `key:poll:weight:${keyId}:*` + ); +} + +function buildOptionResults( + options: string[], + rawResults: Record, + totalVotingWeight: string +): { + normalizedResults: Record; + optionResults: KeyPollOptionResult[]; +} { + const totalNum = Number(totalVotingWeight) || 0; + const normalizedResults: Record = {}; + const optionResults: KeyPollOptionResult[] = options.map((opt, index) => { + const votes = String(rawResults?.[opt] ?? '0'); + normalizedResults[opt] = votes; + const votesNum = Number(votes) || 0; + const percentage = + totalNum > 0 ? Math.round((votesNum / totalNum) * 10000) / 100 : 0; + return { + index, + option: opt, + votes, + percentage, + }; + }); + + return { normalizedResults, optionResults }; +} + +export function mapProposalToPollItem( + row: { + keyId: string; + proposalId: string; + title: string; + options: unknown; + totalVotingWeight: string; + results: unknown; + snapshotLedger: number; + expiresAt: Date; + closedAt: Date | null; + status: string; + }, + now: Date = new Date() +): KeyPollItem { + const options = Array.isArray(row.options) ? (row.options as string[]) : []; + const rawResults = + row.results && typeof row.results === 'object' + ? (row.results as Record) + : {}; + + const isExpired = row.expiresAt.getTime() <= now.getTime(); + const isClosed = row.status !== 'active' || row.closedAt !== null || isExpired; + const timeRemainingSeconds = isClosed + ? 0 + : Math.max(0, Math.floor((row.expiresAt.getTime() - now.getTime()) / 1000)); + + const { normalizedResults, optionResults } = buildOptionResults( + options, + rawResults, + row.totalVotingWeight + ); + + return { + poll_id: row.proposalId, + key_id: row.keyId, + title: row.title, + options, + status: isClosed ? 'closed' : 'active', + raw_status: row.status, + tally: normalizedResults, + results: normalizedResults, + option_results: optionResults, + total_voting_weight: row.totalVotingWeight, + snapshot_block: row.snapshotLedger, + expires_at: row.expiresAt.toISOString(), + closed_at: row.closedAt ? row.closedAt.toISOString() : null, + time_remaining_seconds: timeRemainingSeconds, + }; +} + +/** + * Lists paginated active and closed community polls for a key (#1174). + */ +export async function listKeyPolls( + keyId: string, + opts: { + status?: PollFilterStatus; + page?: number; + limit?: number; + now?: Date; + } = {} +): Promise { + const page = Math.max(1, Math.trunc(opts.page ?? 1)); + const limit = Math.min(Math.max(1, Math.trunc(opts.limit ?? 20)), 100); + const statusKey = opts.status ?? 'all'; + const cacheKey = keyPollsListCacheKey(keyId, statusKey, page, limit); + + try { + const cached = await cacheGetJson(cacheKey); + if (cached !== null) { + return cached; + } + } catch { + // Fall through to DB on Redis failure. + } + + const now = opts.now ?? new Date(); + const where: Record = { keyId }; + if (opts.status === 'active') { + where.status = 'active'; + where.expiresAt = { gt: now }; + } else if (opts.status === 'closed') { + where.OR = [{ status: { not: 'active' } }, { expiresAt: { lte: now } }]; + } + + const skip = (page - 1) * limit; + const [rows, totalCount] = await Promise.all([ + prisma.governanceProposal.findMany({ + where, + orderBy: + opts.status === 'closed' + ? { closedAt: 'desc' } + : { createdAt: 'desc' }, + skip, + take: limit, + }), + prisma.governanceProposal.count({ where }), + ]); + + const polls = rows.map(row => mapProposalToPollItem(row, now)); + const result: KeyPollsPage = { + key_id: keyId, + polls, + items: polls, + pagination: buildPaginationMetadata({ page, limit, totalCount }), + }; + + try { + await cacheSetJson(cacheKey, result, KEY_POLLS_CACHE_TTL_SECONDS); + } catch { + // Cache write failure is non-critical. + } + + return result; +} + +/** + * Returns a single poll with per-option results and the caller/query wallet's + * cast vote when available (#1174). Throws PollNotFoundError if not found. + */ +export async function getKeyPollDetail( + keyId: string, + pollId: string, + wallet?: string, + now: Date = new Date() +): Promise { + const normalizedWallet = wallet?.trim() || undefined; + const cacheKey = keyPollDetailCacheKey( + keyId, + pollId, + normalizedWallet ?? '_anon' + ); + + try { + const cached = await cacheGetJson(cacheKey); + if (cached !== null) { + return cached; + } + } catch { + // Fall through to DB on Redis failure. + } + + const row = await prisma.governanceProposal.findFirst({ + where: { keyId, proposalId: pollId }, + }); + + if (!row) { + throw new PollNotFoundError(pollId); + } + + const item = mapProposalToPollItem(row, now); + let walletVote: KeyPollWalletVote | null = null; + + if (normalizedWallet) { + const voteRow = await prisma.governanceVote.findUnique({ + where: { + keyId_proposalId_voter: { + keyId, + proposalId: pollId, + voter: normalizedWallet, + }, + }, + }); + if (voteRow) { + walletVote = { + voter: voteRow.voter, + option_index: voteRow.optionIndex, + option: item.options[voteRow.optionIndex] ?? '', + weight: String(voteRow.weight), + }; + } + } + + const detail: KeyPollDetail = { + ...item, + wallet_vote: walletVote, + }; + + try { + await cacheSetJson(cacheKey, detail, KEY_POLLS_CACHE_TTL_SECONDS); + } catch { + // Cache write failure is non-critical. + } + + return detail; +} diff --git a/src/modules/keys/key-proposal-votes.service.ts b/src/modules/keys/key-proposal-votes.service.ts index b33bd9a5..c8441d28 100644 --- a/src/modules/keys/key-proposal-votes.service.ts +++ b/src/modules/keys/key-proposal-votes.service.ts @@ -239,6 +239,13 @@ export async function castKeyProposalVote( }); }); + try { + const { invalidateKeyPollsCache } = await import('./key-polls.service'); + await invalidateKeyPollsCache(keyId); + } catch { + // Cache invalidation is best-effort. + } + return { proposalId, optionIndex, diff --git a/src/modules/keys/keys.routes.ts b/src/modules/keys/keys.routes.ts index 934b9b7d..9d19d37e 100644 --- a/src/modules/keys/keys.routes.ts +++ b/src/modules/keys/keys.routes.ts @@ -107,6 +107,7 @@ import { PollVoteWeight, POLL_WEIGHT_CACHE_SECONDS, } from './key-poll-weight.service'; +import { getKeyPollDetail, listKeyPolls } from './key-polls.service'; import { getKeyHoldingCapacity } from './key-holding-capacity.service'; import { StellarAddressSchema } from '../wallet/wallet.schemas'; import { @@ -1482,6 +1483,81 @@ router.post( } ); +/** + * GET /api/v1/keys/:keyId/polls?status=active|closed&page=&limit= + * Paginated active and closed community polls with tally and time remaining (#1174). + */ +const listPollsQuerySchema = z.object({ + status: z.enum(['active', 'closed']).optional(), + page: z.coerce.number().int().min(1).default(1), + limit: z.coerce.number().int().min(1).max(100).default(20), +}); + +router.get( + '/:keyId/polls', + cacheControl({ maxAge: 30, type: 'public', mustRevalidate: true }), + async (req, res, next) => { + const parsed = listPollsQuerySchema.safeParse(req.query); + if (!parsed.success) { + sendValidationError( + res, + 'Invalid polls query parameters', + zodIssuesToDetails(parsed.error.issues) + ); + return; + } + + try { + const keyId = String(req.params.keyId); + const result = await listKeyPolls(keyId, parsed.data); + sendSuccess(res, result); + } catch (error) { + next(error); + } + } +); + +/** + * GET /api/v1/keys/:keyId/polls/:pollId?wallet= + * Single community poll with per-option results and optional wallet vote (#1174). + */ +const pollDetailQuerySchema = z.object({ + wallet: StellarAddressSchema.optional(), +}); + +router.get( + '/:keyId/polls/:pollId', + cacheControl({ maxAge: 30, type: 'public', mustRevalidate: true }), + async (req, res, next) => { + const parsed = pollDetailQuerySchema.safeParse(req.query); + if (!parsed.success) { + sendValidationError( + res, + 'Invalid poll query parameters', + zodIssuesToDetails(parsed.error.issues) + ); + return; + } + + try { + const keyId = String(req.params.keyId); + const pollId = String(req.params.pollId); + const detail = await getKeyPollDetail( + keyId, + pollId, + parsed.data.wallet + ); + sendSuccess(res, detail); + } catch (error) { + if (error instanceof PollNotFoundError) { + sendNotFound(res, 'Poll'); + return; + } + next(error); + } + } +); + /** * GET /api/v1/keys/:keyId/polls/:pollId/weight/:wallet * Return a wallet's estimated vote weight for a key poll, read from the holder diff --git a/src/modules/tips/tips-leaderboard.test.ts b/src/modules/tips/tips-leaderboard.test.ts new file mode 100644 index 00000000..8203dd73 --- /dev/null +++ b/src/modules/tips/tips-leaderboard.test.ts @@ -0,0 +1,165 @@ +import express from 'express'; +import request from 'supertest'; +import tipsRouter from './tips.routes'; +import { getTipLeaderboard, invalidateKeyTipsCache } from './tips.service'; +import { prisma } from '../../utils/prisma.utils'; +import { + cacheGetJson, + cacheInvalidate, + cacheSetJson, +} from '../../utils/redis.utils'; + +const mockCacheStore = new Map(); +const mockTtls: number[] = []; + +jest.mock('../../utils/logger.utils', () => ({ + logger: { + info: jest.fn(), + warn: jest.fn(), + error: jest.fn(), + debug: jest.fn(), + }, +})); + +jest.mock('../../utils/redis.utils', () => ({ + cacheGetJson: jest.fn(async (key: string) => { + const raw = mockCacheStore.get(key); + return raw === undefined ? null : JSON.parse(raw); + }), + cacheSetJson: jest.fn( + async (key: string, value: unknown, ttlSeconds: number) => { + mockCacheStore.set(key, JSON.stringify(value)); + mockTtls.push(ttlSeconds); + } + ), + cacheInvalidate: jest.fn(async (pattern: string) => { + const prefix = pattern.replace(/\*$/, ''); + for (const k of Array.from(mockCacheStore.keys())) { + if (k.startsWith(prefix)) { + mockCacheStore.delete(k); + } + } + }), +})); + +jest.mock('../whitelist/whitelist.service', () => ({ + resolveCreatorId: jest.fn(async (keyId: string) => keyId), +})); + +jest.mock('../../utils/prisma.utils', () => ({ + prisma: { + keyTip: { + findMany: jest.fn(), + }, + }, +})); + +const app = express(); +app.use(express.json()); +app.use('/keys', tipsRouter); + +const WALLET_A = 'GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAWHF'; +const WALLET_B = 'GBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBWHF'; + +beforeEach(() => { + jest.clearAllMocks(); + mockCacheStore.clear(); + mockTtls.length = 0; +}); + +describe('GET /keys/:key_id/tips/leaderboard (#1173)', () => { + it('returns tippers ranked by total amount with truncated wallets', async () => { + (prisma.keyTip.findMany as jest.Mock).mockResolvedValue([ + { tipper: WALLET_B, amount: '10.0000000', anonymous: false }, + { tipper: WALLET_A, amount: '25.5000000', anonymous: false }, + { tipper: WALLET_A, amount: '14.5000000', anonymous: false }, + ]); + + const res = await request(app).get( + '/keys/key-1/tips/leaderboard?limit=5' + ); + + expect(res.status).toBe(200); + expect(res.body.success).toBe(true); + expect(res.body.data).toEqual([ + { + rank: 1, + wallet: 'GAAA…AWHF', + total_tipped: '40.0000000', + tip_count: 2, + anonymous: false, + }, + { + rank: 2, + wallet: 'GBBB…BWHF', + total_tipped: '10.0000000', + tip_count: 1, + anonymous: false, + }, + ]); + expect(cacheSetJson).toHaveBeenCalledWith( + 'key:tips:key-1:leaderboard:5', + expect.any(Array), + 300 + ); + }); + + it('handles anonymous tips with wallet: null and anonymous: true', async () => { + (prisma.keyTip.findMany as jest.Mock).mockResolvedValue([ + { tipper: 'anonymous', amount: '50.0000000', anonymous: true }, + { tipper: WALLET_A, amount: '20.0000000', anonymous: false }, + ]); + + const leaderboard = await getTipLeaderboard('key-1', 10); + + expect(leaderboard).toEqual([ + { + rank: 1, + wallet: null, + total_tipped: '50.0000000', + tip_count: 1, + anonymous: true, + }, + { + rank: 2, + wallet: 'GAAA…AWHF', + total_tipped: '20.0000000', + tip_count: 1, + anonymous: false, + }, + ]); + }); + + it('returns an empty list for keys with no tips', async () => { + (prisma.keyTip.findMany as jest.Mock).mockResolvedValue([]); + + const res = await request(app).get('/keys/key-empty/tips/leaderboard'); + + expect(res.status).toBe(200); + expect(res.body.data).toEqual([]); + }); + + it('serves cached responses in under 200ms and invalidates on new TipSent event', async () => { + (prisma.keyTip.findMany as jest.Mock).mockResolvedValue([ + { tipper: WALLET_A, amount: '15.0000000', anonymous: false }, + ]); + + await request(app).get('/keys/key-1/tips/leaderboard?limit=10'); + const start = Date.now(); + const cachedRes = await request(app).get( + '/keys/key-1/tips/leaderboard?limit=10' + ); + const elapsed = Date.now() - start; + + expect(cachedRes.status).toBe(200); + expect(elapsed).toBeLessThan(200); + expect(prisma.keyTip.findMany).toHaveBeenCalledTimes(1); + expect(cacheGetJson).toHaveBeenCalledTimes(2); + + await invalidateKeyTipsCache('key-1'); + expect(cacheInvalidate).toHaveBeenCalledWith('key:tips:key-1:*'); + + await request(app).get('/keys/key-1/tips/leaderboard?limit=10'); + expect(prisma.keyTip.findMany).toHaveBeenCalledTimes(2); + }); +}); diff --git a/src/modules/tips/tips.routes.ts b/src/modules/tips/tips.routes.ts index b7deff9b..20b78e62 100644 --- a/src/modules/tips/tips.routes.ts +++ b/src/modules/tips/tips.routes.ts @@ -23,8 +23,12 @@ import { } from '../../utils/api-response.utils'; import { ErrorCode } from '../../constants/error.constants'; import { cacheControl } from '../../middlewares/cache-control.middleware'; -import { KEY_TIPS_CACHE_TTL_SECONDS } from '../../constants/redis.constants'; import { + KEY_TIPS_CACHE_TTL_SECONDS, + KEY_TIPS_LEADERBOARD_CACHE_TTL_SECONDS, +} from '../../constants/redis.constants'; +import { + getTipLeaderboard, getTipMessageContent, getTipTotals, listTips, @@ -44,6 +48,10 @@ const ListTipsQuerySchema = z.object({ cursor: z.string().min(1).optional(), }); +const TipLeaderboardQuerySchema = z.object({ + limit: z.coerce.number().int().min(1).max(100).default(10), +}); + const tipsCache = () => cacheControl({ maxAge: KEY_TIPS_CACHE_TTL_SECONDS, @@ -51,6 +59,44 @@ const tipsCache = () => mustRevalidate: true, }); +const tipsLeaderboardCache = () => + cacheControl({ + maxAge: KEY_TIPS_LEADERBOARD_CACHE_TTL_SECONDS, + type: 'public', + mustRevalidate: true, + }); + +/** + * GET /keys/:key_id/tips/leaderboard?limit=N + * Public. Top tippers ranked by total amount tipped (#1173). + */ +const httpGetTipLeaderboard: AsyncController = async (req, res, next) => { + try { + const keyId = (req.params.key_id ?? req.params.keyId) as string; + if (!keyId) { + return sendValidationError(res, 'Missing required parameters', [ + { field: 'keyId', message: 'Key ID is required' }, + ]); + } + + const parsedQuery = TipLeaderboardQuerySchema.safeParse(req.query); + if (!parsedQuery.success) { + return sendValidationError( + res, + 'Invalid leaderboard query', + zodIssuesToDetails(parsedQuery.error.issues) + ); + } + + return sendSuccess( + res, + await getTipLeaderboard(keyId, parsedQuery.data.limit) + ); + } catch (error) { + next(error); + } +}; + /** * GET /keys/:key_id/tips?limit=&cursor= * Public. Cursor-paginated tip history: tipper, amount, message_hash, @@ -135,6 +181,13 @@ const httpGetTipContent: AsyncController = async (req, res) => { } }; +tipsRouter.get( + ['/:key_id/tips/leaderboard', '/:keyId/tips/leaderboard'], + normalizeKeyId, + tipsLeaderboardCache(), + httpGetTipLeaderboard +); + tipsRouter.get( ['/:key_id/tips/total', '/:keyId/tips/total'], normalizeKeyId, diff --git a/src/modules/tips/tips.service.ts b/src/modules/tips/tips.service.ts index 64e80b9a..8390bb82 100644 --- a/src/modules/tips/tips.service.ts +++ b/src/modules/tips/tips.service.ts @@ -22,8 +22,10 @@ import { logger } from '../../utils/logger.utils'; import { KEY_TIP_CONTENT_CACHE_TTL_SECONDS, KEY_TIPS_CACHE_TTL_SECONDS, + KEY_TIPS_LEADERBOARD_CACHE_TTL_SECONDS, keyTipContentCacheKey, keyTipsCachePattern, + keyTipsLeaderboardCacheKey, keyTipsListCacheKey, keyTipsTotalCacheKey, } from '../../constants/redis.constants'; @@ -31,6 +33,7 @@ import { normalizeIpfsCid, resolvePinataGatewayUrl, } from '../../utils/pinata-gateway.utils'; +import { truncateWallet } from '../../utils/wallet-display.utils'; import { resolveCreatorId } from '../whitelist/whitelist.service'; /** One tip in the history. */ @@ -60,6 +63,15 @@ export interface TipTotals { count: number; } +/** One entry on the per-key tip leaderboard (#1173). */ +export interface TipLeaderboardEntry { + rank: number; + wallet: string | null; + total_tipped: string; + tip_count: number; + anonymous: boolean; +} + /** Maximum IPFS message size fetched for the content endpoint (256 KiB). */ export const MAX_TIP_MESSAGE_BYTES = 256 * 1024; @@ -67,11 +79,15 @@ export const MAX_TIP_MESSAGE_BYTES = 256 * 1024; export const TIP_MESSAGE_FETCH_TIMEOUT_MS = 5000; /** - * Drops every cached tip response for a key (list + total). Called by the - * indexer after every TipSent event. + * Drops every cached tip response for a key (list + total + leaderboard). + * Called by the indexer after every TipSent event. */ export async function invalidateKeyTipsCache(keyId: string): Promise { - await cacheInvalidate(keyTipsCachePattern(keyId)); + const resolvedKeyId = await resolveTipKeyId(keyId); + await cacheInvalidate(keyTipsCachePattern(resolvedKeyId)); + if (resolvedKeyId !== keyId) { + await cacheInvalidate(keyTipsCachePattern(keyId)); + } } /** @@ -189,6 +205,138 @@ export async function getTipTotals(keyId: string): Promise { return result; } +function isAnonymousTipRow(row: { + tipper?: string | null; + anonymous?: boolean | null; +}): boolean { + if (row.anonymous === true) return true; + const trimmed = (row.tipper ?? '').trim(); + if (!trimmed) return true; + const lower = trimmed.toLowerCase(); + return lower === 'anonymous' || lower === 'anon' || lower === 'null'; +} + +/** + * Returns the top tippers for a creator key ranked by total amount (#1173). + * + * - Anonymous tips are represented with `wallet: null` and `anonymous: true`. + * - Non-anonymous tippers have their wallet truncated (`GABC…WXYZ`). + * - Cached in Redis for 5 minutes (300s) and invalidated on TipSent events. + * - Keys with no tips return an empty array `[]`. + */ +export async function getTipLeaderboard( + keyId: string, + limit = 10 +): Promise { + const resolvedKeyId = await resolveTipKeyId(keyId); + const clampedLimit = Math.min(Math.max(Math.trunc(limit) || 10, 1), 100); + const cacheKey = keyTipsLeaderboardCacheKey(resolvedKeyId, clampedLimit); + + try { + const cached = await cacheGetJson(cacheKey); + if (cached !== null) { + logger.debug({ keyId: resolvedKeyId }, 'Tip leaderboard cache hit'); + return cached; + } + } catch { + // Cache read failures fall through to the database. + } + + const rows = await prisma.keyTip.findMany({ + where: { keyId: resolvedKeyId }, + select: { + tipper: true, + amount: true, + anonymous: true, + }, + }); + + if (rows.length === 0) { + try { + await cacheSetJson( + cacheKey, + [], + KEY_TIPS_LEADERBOARD_CACHE_TTL_SECONDS + ); + } catch { + // Cache write failures are non-critical. + } + return []; + } + + const buckets = new Map< + string, + { + rawWallet: string | null; + anonymous: boolean; + totalStroops: bigint; + tipCount: number; + } + >(); + + for (const row of rows) { + const anon = isAnonymousTipRow(row); + const trimmedTipper = (row.tipper ?? '').trim(); + const hasDistinctWallet = + Boolean(trimmedTipper) && + trimmedTipper.toLowerCase() !== 'anonymous' && + trimmedTipper.toLowerCase() !== 'anon' && + trimmedTipper.toLowerCase() !== 'null'; + + const bucketKey = anon + ? hasDistinctWallet + ? `anon:${trimmedTipper}` + : '__anonymous__' + : `wallet:${trimmedTipper}`; + + const stroops = BigInt(Math.round(Number(row.amount) * 1e7)); + const existing = buckets.get(bucketKey); + if (existing) { + existing.totalStroops += stroops; + existing.tipCount += 1; + } else { + buckets.set(bucketKey, { + rawWallet: anon ? null : trimmedTipper, + anonymous: anon, + totalStroops: stroops, + tipCount: 1, + }); + } + } + + const sorted = Array.from(buckets.entries()) + .sort(([keyA, a], [keyB, b]) => { + if (b.totalStroops !== a.totalStroops) { + return b.totalStroops > a.totalStroops ? 1 : -1; + } + if (b.tipCount !== a.tipCount) { + return b.tipCount - a.tipCount; + } + return keyA.localeCompare(keyB); + }) + .slice(0, clampedLimit); + + const result: TipLeaderboardEntry[] = sorted.map(([, entry], index) => ({ + rank: index + 1, + wallet: entry.anonymous ? null : truncateWallet(entry.rawWallet ?? ''), + total_tipped: (Number(entry.totalStroops) / 1e7).toFixed(7), + tip_count: entry.tipCount, + anonymous: entry.anonymous, + })); + + try { + await cacheSetJson( + cacheKey, + result, + KEY_TIPS_LEADERBOARD_CACHE_TTL_SECONDS + ); + } catch { + // Cache write failures are non-critical. + } + + return result; +} + /** Content fetched from IPFS for a tip message. */ export interface TipMessageContent { message_hash: string;