diff --git a/.env.example b/.env.example index b733f5ef..600bcc4f 100644 --- a/.env.example +++ b/.env.example @@ -151,6 +151,10 @@ INVOICE_TIMELOCK_CONTRACT_ID= INVOICE_MARKET_CONTRACT_ID= INVOICE_MARKET_ABI= +# Portfolio health score contract (get_health_score). Unset returns 503 from +# GET /investors/:wallet/health-score. +PORTFOLIO_HEALTH_CONTRACT_ID= + # Multi-tier circuit breaker (#726): per-dependency health tracking and automatic recovery probes # Global defaults CIRCUIT_BREAKER_FAILURE_THRESHOLD=5 diff --git a/docs/api-endpoints.md b/docs/api-endpoints.md index 1422ba50..d6704f50 100644 --- a/docs/api-endpoints.md +++ b/docs/api-endpoints.md @@ -316,6 +316,99 @@ GET /invoices/compare?ids=cm123,cm456 --- +### GET /invoices/:invoice_id/early-incentive + +Early funding incentive slot status for an invoice. `slots_filled` counts indexed +`EarlySlotFilled` events and is capped at `slots_total`. Configuration comes from +the most recent event. Responses are cached for 30 seconds. + +- **Auth:** None +- **Response:** `200 OK`; `404` when the invoice does not exist. An invoice with + no incentive configured returns `is_active: false`. + +```json +{ + "success": true, + "data": { + "invoice_id": "cm123", + "slots_total": 10, + "slots_filled": 4, + "slots_remaining": 6, + "bonus_bps": 250, + "is_active": true + } +} +``` + +--- + +## Issuer Endpoints + +### GET /issuers/:wallet/verification + +Verification badge status for an issuer. `is_verified` is `false` for revoked, +expired, or never-verified issuers, and `tier` is `null` in those cases. +`days_remaining` is `null` for a grant with no expiry. Responses are cached for 60 +seconds. + +- **Auth:** None +- **Response:** `200 OK`; `400` for an invalid Stellar wallet + +```json +{ + "success": true, + "data": { + "wallet": "GISSUER...", + "is_verified": true, + "tier": "GOLD", + "expiry_timestamp": "2027-01-01T00:00:00.000Z", + "days_remaining": 71 + } +} +``` + +### GET /issuers/verified + +Paginated list of currently verified issuers. + +- **Auth:** None +- **Query:** `page` (default `1`), `limit` (default `20`, max `100`) +- **Response:** `200 OK` with `meta` pagination + +--- + +## Investor Endpoints + +### GET /investors/:wallet/health-score + +Portfolio health score for a wallet, read from the health-score contract's +`get_health_score` and cached for 5 minutes. `score` and `factors` are `null` for +a wallet with no portfolio. Requires `PORTFOLIO_HEALTH_CONTRACT_ID`; without it the +endpoint returns `503`. + +- **Auth:** None +- **Response:** `200 OK`; `400` for an invalid wallet; `503` when the contract is + not configured or the read fails + +```json +{ + "success": true, + "data": { + "wallet": "GINVESTOR...", + "score": 742, + "factors": { + "concentration_score": 80, + "maturity_distribution_score": 70, + "issuer_diversity_score": 65, + "default_exposure_score": 90 + }, + "computed_at": "2026-10-21T10:00:00.000Z" + } +} +``` + +--- + ## Admin Endpoints ### POST /admin/proposals @@ -611,6 +704,45 @@ window, with alert and auto-suspension state. --- +### GET /admin/treasury + +Platform treasury balance for the admin dashboard. `total_balance` is gross fee +inflow minus allocations. `per_source_breakdown` is gross inflow per source; +allocations are not attributed to a source. Amounts are strings in stroops. +Cached for 60 seconds. + +- **Auth:** Admin JWT (`adminGuard`) +- **Response:** `200 OK` + +```json +{ + "success": true, + "data": { + "total_balance": "1200000000", + "per_source_breakdown": [{ "source": "trading", "balance": "1500000000" }], + "last_updated": "2026-10-21T10:00:00.000Z" + } +} +``` + +### GET /admin/treasury/allocations + +Paginated `FundsAllocated` history (`destination`, `amount`, `purpose`, `tx_hash`, +`ledger`, `occurred_at`). + +- **Auth:** Admin JWT +- **Query:** `page` (default `1`), `limit` (default `20`, max `100`) + +### GET /admin/treasury/inflows + +Paginated `FeeReceived` history (`source`, `amount`, `tx_hash`, `ledger`, +`occurred_at`). + +- **Auth:** Admin JWT +- **Query:** `page` (default `1`), `limit` (default `20`, max `100`) + +--- + ## Keys Endpoints ### GET /keys/:keyId/price diff --git a/prisma/schema/early-incentive.prisma b/prisma/schema/early-incentive.prisma new file mode 100644 index 00000000..6271ebee --- /dev/null +++ b/prisma/schema/early-incentive.prisma @@ -0,0 +1,23 @@ +// prisma/schema/early-incentive.prisma +// Invoice early funding incentive slot tracking (#1180). + +/// Append-only log of EarlySlotFilled events. Slots filled is a count of these +/// rows; the incentive configuration (total slots, bonus) is taken from the +/// most recent event for the invoice. The unique chain position makes replays +/// no-ops. +model InvoiceEarlySlotEvent { + id String @id @default(cuid()) + invoiceId String + slotsTotal Int + bonusBps Int + investor String? + txHash String + eventIndex Int + ledger Int + occurredAt DateTime + createdAt DateTime @default(now()) + + @@unique([txHash, eventIndex]) + @@index([invoiceId, ledger(sort: Desc), eventIndex(sort: Desc)]) + @@map("invoice_early_slot_events") +} diff --git a/prisma/schema/issuer-verification.prisma b/prisma/schema/issuer-verification.prisma new file mode 100644 index 00000000..753063ad --- /dev/null +++ b/prisma/schema/issuer-verification.prisma @@ -0,0 +1,34 @@ +// prisma/schema/issuer-verification.prisma +// Issuer verification badge state fed by VerificationGranted / VerificationRevoked (#1179). + +/// Append-only log of verification events. The unique chain position makes +/// replays no-ops, so IssuerVerification only changes for newly seen events. +model IssuerVerificationEvent { + id String @id @default(cuid()) + wallet String + eventType String + tier String? + expiresAt DateTime? + txHash String + eventIndex Int + ledger Int + occurredAt DateTime + createdAt DateTime @default(now()) + + @@unique([txHash, eventIndex]) + @@index([wallet, occurredAt(sort: Desc)]) + @@map("issuer_verification_events") +} + +/// Current verification status per issuer wallet, derived from the event log. +model IssuerVerification { + wallet String @id + isVerified Boolean @default(false) + tier String? + expiresAt DateTime? + lastLedger Int + updatedAt DateTime @updatedAt + + @@index([isVerified, expiresAt]) + @@map("issuer_verification") +} diff --git a/prisma/schema/migrations/20261021120000_add_treasury_early_incentive_verification/migration.sql b/prisma/schema/migrations/20261021120000_add_treasury_early_incentive_verification/migration.sql new file mode 100644 index 00000000..c2e19853 --- /dev/null +++ b/prisma/schema/migrations/20261021120000_add_treasury_early_incentive_verification/migration.sql @@ -0,0 +1,111 @@ +-- CreateTable +CREATE TABLE "treasury_balance" ( + "source" TEXT NOT NULL, + "balance" DECIMAL(40,0) NOT NULL, + "updatedAt" TIMESTAMP(3) NOT NULL, + + CONSTRAINT "treasury_balance_pkey" PRIMARY KEY ("source") +); + +-- CreateTable +CREATE TABLE "treasury_inflows" ( + "id" TEXT NOT NULL, + "source" TEXT NOT NULL, + "amount" DECIMAL(40,0) NOT NULL, + "txHash" TEXT NOT NULL, + "eventIndex" INTEGER NOT NULL, + "ledger" INTEGER NOT NULL, + "occurredAt" TIMESTAMP(3) NOT NULL, + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + + CONSTRAINT "treasury_inflows_pkey" PRIMARY KEY ("id") +); + +-- CreateTable +CREATE TABLE "treasury_allocations" ( + "id" TEXT NOT NULL, + "destination" TEXT NOT NULL, + "amount" DECIMAL(40,0) NOT NULL, + "purpose" TEXT NOT NULL, + "txHash" TEXT NOT NULL, + "eventIndex" INTEGER NOT NULL, + "ledger" INTEGER NOT NULL, + "occurredAt" TIMESTAMP(3) NOT NULL, + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + + CONSTRAINT "treasury_allocations_pkey" PRIMARY KEY ("id") +); + +-- CreateTable +CREATE TABLE "invoice_early_slot_events" ( + "id" TEXT NOT NULL, + "invoiceId" TEXT NOT NULL, + "slotsTotal" INTEGER NOT NULL, + "bonusBps" INTEGER NOT NULL, + "investor" TEXT, + "txHash" TEXT NOT NULL, + "eventIndex" INTEGER NOT NULL, + "ledger" INTEGER NOT NULL, + "occurredAt" TIMESTAMP(3) NOT NULL, + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + + CONSTRAINT "invoice_early_slot_events_pkey" PRIMARY KEY ("id") +); + +-- CreateTable +CREATE TABLE "issuer_verification_events" ( + "id" TEXT NOT NULL, + "wallet" TEXT NOT NULL, + "eventType" TEXT NOT NULL, + "tier" TEXT, + "expiresAt" TIMESTAMP(3), + "txHash" TEXT NOT NULL, + "eventIndex" INTEGER NOT NULL, + "ledger" INTEGER NOT NULL, + "occurredAt" TIMESTAMP(3) NOT NULL, + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + + CONSTRAINT "issuer_verification_events_pkey" PRIMARY KEY ("id") +); + +-- CreateTable +CREATE TABLE "issuer_verification" ( + "wallet" TEXT NOT NULL, + "isVerified" BOOLEAN NOT NULL DEFAULT false, + "tier" TEXT, + "expiresAt" TIMESTAMP(3), + "lastLedger" INTEGER NOT NULL, + "updatedAt" TIMESTAMP(3) NOT NULL, + + CONSTRAINT "issuer_verification_pkey" PRIMARY KEY ("wallet") +); + +-- CreateIndex +CREATE UNIQUE INDEX "treasury_inflows_txHash_eventIndex_key" ON "treasury_inflows"("txHash", "eventIndex"); + +-- CreateIndex +CREATE INDEX "treasury_inflows_source_occurredAt_idx" ON "treasury_inflows"("source", "occurredAt" DESC); + +-- CreateIndex +CREATE INDEX "treasury_inflows_occurredAt_idx" ON "treasury_inflows"("occurredAt" DESC); + +-- CreateIndex +CREATE UNIQUE INDEX "treasury_allocations_txHash_eventIndex_key" ON "treasury_allocations"("txHash", "eventIndex"); + +-- CreateIndex +CREATE INDEX "treasury_allocations_occurredAt_idx" ON "treasury_allocations"("occurredAt" DESC); + +-- CreateIndex +CREATE UNIQUE INDEX "invoice_early_slot_events_txHash_eventIndex_key" ON "invoice_early_slot_events"("txHash", "eventIndex"); + +-- CreateIndex +CREATE INDEX "invoice_early_slot_events_invoiceId_ledger_eventIndex_idx" ON "invoice_early_slot_events"("invoiceId", "ledger" DESC, "eventIndex" DESC); + +-- CreateIndex +CREATE UNIQUE INDEX "issuer_verification_events_txHash_eventIndex_key" ON "issuer_verification_events"("txHash", "eventIndex"); + +-- CreateIndex +CREATE INDEX "issuer_verification_events_wallet_occurredAt_idx" ON "issuer_verification_events"("wallet", "occurredAt" DESC); + +-- CreateIndex +CREATE INDEX "issuer_verification_isVerified_expiresAt_idx" ON "issuer_verification"("isVerified", "expiresAt"); diff --git a/prisma/schema/treasury.prisma b/prisma/schema/treasury.prisma new file mode 100644 index 00000000..3f2a8a0e --- /dev/null +++ b/prisma/schema/treasury.prisma @@ -0,0 +1,48 @@ +// prisma/schema/treasury.prisma +// Platform treasury read models fed by FeeReceived / FundsAllocated (#1181). +// Amounts are stored in the contract's smallest unit (stroops). + +/// Gross fee inflow per source. A balance only moves when its matching +/// TreasuryInflow row is newly inserted, so replayed events cannot double-count. +model TreasuryBalance { + source String @id + balance Decimal @db.Decimal(40, 0) + updatedAt DateTime @updatedAt + + @@map("treasury_balance") +} + +/// Append-only log of FeeReceived events. The unique chain position is the +/// idempotency guard for replays. +model TreasuryInflow { + id String @id @default(cuid()) + source String + amount Decimal @db.Decimal(40, 0) + txHash String + eventIndex Int + ledger Int + occurredAt DateTime + createdAt DateTime @default(now()) + + @@unique([txHash, eventIndex]) + @@index([source, occurredAt(sort: Desc)]) + @@index([occurredAt(sort: Desc)]) + @@map("treasury_inflows") +} + +/// Append-only log of FundsAllocated events (treasury outflows). +model TreasuryAllocation { + id String @id @default(cuid()) + destination String + amount Decimal @db.Decimal(40, 0) + purpose String + txHash String + eventIndex Int + ledger Int + occurredAt DateTime + createdAt DateTime @default(now()) + + @@unique([txHash, eventIndex]) + @@index([occurredAt(sort: Desc)]) + @@map("treasury_allocations") +} diff --git a/src/config.schema.ts b/src/config.schema.ts index 80b316b8..bfc1ac98 100644 --- a/src/config.schema.ts +++ b/src/config.schema.ts @@ -262,6 +262,10 @@ export const envSchema = z INVOICE_FRAUD_CONTRACT_ID: optionalNonEmptyString, INVOICE_TIMELOCK_CONTRACT_ID: optionalNonEmptyString, + // Portfolio health score contract (get_health_score). Unset fails the + // /investors/:wallet/health-score read closed with 503. + PORTFOLIO_HEALTH_CONTRACT_ID: optionalNonEmptyString, + // Social metadata sync (Pinata gateway used to resolve avatar CIDs) PINATA_GATEWAY_URL: z .string() diff --git a/src/modules/admin/admin.routes.ts b/src/modules/admin/admin.routes.ts index 389e1221..dbd3a999 100644 --- a/src/modules/admin/admin.routes.ts +++ b/src/modules/admin/admin.routes.ts @@ -50,6 +50,7 @@ import { Prisma } from '@prisma/client'; import { prisma } from '../../utils/prisma.utils'; import { logger } from '../../utils/logger.utils'; import { invalidateKeyVestingCache } from '../vesting/vesting.service'; +import { treasuryRouter } from '../treasury/treasury.routes'; function isValidStellarAddress(address: string): boolean { return typeof address === 'string' && /^G[A-Z2-7]{55}$/.test(address); @@ -1220,4 +1221,7 @@ adminRouter.post( } ); +// Platform treasury dashboard reads (#1181). Every treasury route is admin-only. +adminRouter.use('/treasury', adminGuard, treasuryRouter); + export default adminRouter; diff --git a/src/modules/index.ts b/src/modules/index.ts index 23825c10..431701ad 100644 --- a/src/modules/index.ts +++ b/src/modules/index.ts @@ -22,6 +22,8 @@ import { issuerRouter } from './invoice/invoice-protocol.routes'; import freezeRouter from './freeze/freeze.routes'; import platformRouter from './platform/platform.routes'; import searchRouter from './search/search.routes'; +import { issuerVerificationRouter } from './issuers/issuer-verification.routes'; +import { investorHealthRouter } from './portfolio/portfolio-health-score.routes'; import tradingRouter from './trading/multi-buy.routes'; import sequencerRouter from './admin/sequencer.routes'; @@ -86,6 +88,8 @@ router.use('/activity', routeBodySizeLimit('default'), activityRouter); router.use('/ownership', routeBodySizeLimit('default'), ownershipRouter); router.use('/invoices', routeBodySizeLimit('default'), invoiceRouter); router.use('/issuers', routeBodySizeLimit('default'), issuerRouter); +router.use('/issuers', routeBodySizeLimit('default'), issuerVerificationRouter); +router.use('/investors', routeBodySizeLimit('default'), investorHealthRouter); router.use('/subscriptions', routeBodySizeLimit('default'), subscriptionRouter); router.use(CREATORS_BASE, routeBodySizeLimit('creators'), webhookRouter); router.use('/wallets', routeBodySizeLimit('default'), walletsRouter); diff --git a/src/modules/indexer/early-incentive-indexer.service.test.ts b/src/modules/indexer/early-incentive-indexer.service.test.ts new file mode 100644 index 00000000..8c65cc1f --- /dev/null +++ b/src/modules/indexer/early-incentive-indexer.service.test.ts @@ -0,0 +1,78 @@ +import { processEarlySlotFilledEvents } from './early-incentive-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: { + invoiceEarlySlotEvent: { + createMany: jest.fn().mockResolvedValue({ count: 1 }), + }, + }, +})); + +const base = { txHash: 'tx-slot', eventIndex: 2, ledger: 700, timestamp: 1_800_000_000 }; + +beforeEach(() => { + jest.clearAllMocks(); +}); + +describe('processEarlySlotFilledEvents', () => { + it('appends one slot-fill row keyed by chain position', async () => { + await processEarlySlotFilledEvents([ + { + ...base, + eventType: 'EarlySlotFilled', + invoice_id: 'inv-1', + slots_total: 10, + bonus_bps: 250, + investor: 'GINVESTOR', + }, + ]); + + expect(prisma.invoiceEarlySlotEvent.createMany).toHaveBeenCalledWith({ + data: [ + expect.objectContaining({ + invoiceId: 'inv-1', + slotsTotal: 10, + bonusBps: 250, + investor: 'GINVESTOR', + txHash: 'tx-slot', + eventIndex: 2, + ledger: 700, + }), + ], + skipDuplicates: true, + }); + expect(cacheInvalidate).toHaveBeenCalledWith('invoice:early-incentive:inv-1'); + }); + + it('skips a fill whose bonus is outside basis-point range', async () => { + await processEarlySlotFilledEvents([ + { + ...base, + eventType: 'EarlySlotFilled', + invoice_id: 'inv-1', + slots_total: 10, + bonus_bps: 20_000, + }, + ]); + + expect(prisma.invoiceEarlySlotEvent.createMany).not.toHaveBeenCalled(); + }); + + it('ignores unrelated event types', async () => { + await processEarlySlotFilledEvents([ + { ...base, eventType: 'KEY_BOUGHT', invoice_id: 'inv-1' }, + ]); + + expect(prisma.invoiceEarlySlotEvent.createMany).not.toHaveBeenCalled(); + }); +}); diff --git a/src/modules/indexer/early-incentive-indexer.service.ts b/src/modules/indexer/early-incentive-indexer.service.ts new file mode 100644 index 00000000..7906a4c9 --- /dev/null +++ b/src/modules/indexer/early-incentive-indexer.service.ts @@ -0,0 +1,74 @@ +// src/modules/indexer/early-incentive-indexer.service.ts +// Indexes EarlySlotFilled contract events (#1180). Each event is appended to +// invoice_early_slot_events once per chain position; replays are no-ops. The +// read model derives slots filled from that log. + +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 { invalidateEarlyIncentiveCache } from '../invoice/invoice-early-incentive.service'; + +export const EARLY_SLOT_FILLED_EVENT_TYPES = new Set([ + 'EarlySlotFilled', + 'EARLY_SLOT_FILLED', +]); + +const earlySlotFilledPayload = z.object({ + invoiceId: eventIdentifierSchema, + slotsTotal: z.number().int().min(1).max(1_000_000), + bonusBps: z.number().int().min(0).max(10_000), + investor: eventIdentifierSchema.nullable().optional(), +}); + +export async function processEarlySlotFilledEvents( + events: IndexerChainEvent[] +): Promise { + await processIndexerChainEvents(events, async event => { + if (!EARLY_SLOT_FILLED_EVENT_TYPES.has(event.eventType)) { + return; + } + + const position = chainPositionSchema.safeParse(event); + const payload = earlySlotFilledPayload.safeParse({ + invoiceId: event.invoice_id ?? event.invoiceId, + slotsTotal: event.slots_total ?? event.slotsTotal, + bonusBps: event.bonus_bps ?? event.bonusBps, + investor: event.investor ?? event.actor, + }); + if (!position.success || !payload.success) { + logger.warn( + { + eventId: `${event.txHash}:${event.eventIndex}`, + eventType: event.eventType, + }, + 'Skipping EarlySlotFilled event with missing or invalid fields' + ); + return; + } + + await prisma.invoiceEarlySlotEvent.createMany({ + data: [ + { + invoiceId: payload.data.invoiceId, + slotsTotal: payload.data.slotsTotal, + bonusBps: payload.data.bonusBps, + investor: payload.data.investor ?? null, + occurredAt: eventOccurredAt(event), + ...position.data, + }, + ], + skipDuplicates: true, + }); + + await invalidateEarlyIncentiveCache(payload.data.invoiceId); + }); +} diff --git a/src/modules/indexer/health-score-indexer.service.ts b/src/modules/indexer/health-score-indexer.service.ts new file mode 100644 index 00000000..e166d839 --- /dev/null +++ b/src/modules/indexer/health-score-indexer.service.ts @@ -0,0 +1,48 @@ +// src/modules/indexer/health-score-indexer.service.ts +// HealthScoreUpdated contract events (#1178). The score itself is read live +// from the contract, so the indexer's only job is to drop the cached score for +// the affected wallet so the next read fetches the new value. + +import { z } from 'zod'; +import { logger } from '../../utils/logger.utils'; +import { + IndexerChainEvent, + processIndexerChainEvents, +} from '../../utils/indexer-event-processor.utils'; +import { eventIdentifierSchema } from '../../utils/indexer-event-fields.utils'; +import { invalidateHealthScoreCache } from '../portfolio/portfolio-health-score.service'; + +export const HEALTH_SCORE_UPDATED_EVENT_TYPES = new Set([ + 'HealthScoreUpdated', + 'HEALTH_SCORE_UPDATED', +]); + +const healthScoreUpdatedPayload = z.object({ + wallet: eventIdentifierSchema, +}); + +export async function processHealthScoreEvents( + events: IndexerChainEvent[] +): Promise { + await processIndexerChainEvents(events, async event => { + if (!HEALTH_SCORE_UPDATED_EVENT_TYPES.has(event.eventType)) { + return; + } + + const payload = healthScoreUpdatedPayload.safeParse({ + wallet: event.wallet ?? event.investor ?? event.actor, + }); + if (!payload.success) { + logger.warn( + { + eventId: `${event.txHash}:${event.eventIndex}`, + eventType: event.eventType, + }, + 'Skipping HealthScoreUpdated event without a wallet' + ); + return; + } + + await invalidateHealthScoreCache(payload.data.wallet); + }); +} diff --git a/src/modules/indexer/indexer-pipeline.service.ts b/src/modules/indexer/indexer-pipeline.service.ts index 45f0ea80..2aa3d488 100644 --- a/src/modules/indexer/indexer-pipeline.service.ts +++ b/src/modules/indexer/indexer-pipeline.service.ts @@ -544,6 +544,49 @@ export async function processIndexerEvents( await processTipEvents(tipEvents); } + // Treasury, early incentive slots, issuer verification and health score + // (#1178, #1179, #1180, #1181). Awaited before the cursor moves so a failure + // redelivers the batch instead of leaving a gap in the read models. + const { TREASURY_EVENT_TYPES, processTreasuryEvents } = + await import('./treasury-indexer.service'); + const treasuryEvents = events.filter(event => + TREASURY_EVENT_TYPES.has(event.eventType) + ); + if (treasuryEvents.length > 0) { + await processTreasuryEvents(treasuryEvents); + } + + const { EARLY_SLOT_FILLED_EVENT_TYPES, processEarlySlotFilledEvents } = + await import('./early-incentive-indexer.service'); + const earlySlotEvents = events.filter(event => + EARLY_SLOT_FILLED_EVENT_TYPES.has(event.eventType) + ); + if (earlySlotEvents.length > 0) { + await processEarlySlotFilledEvents(earlySlotEvents); + } + + const { + ISSUER_VERIFICATION_EVENT_TYPES, + processIssuerVerificationEvents, + } = await import('./issuer-verification-indexer.service'); + const verificationEvents = events.filter(event => + ISSUER_VERIFICATION_EVENT_TYPES.has(event.eventType) + ); + if (verificationEvents.length > 0) { + await processIssuerVerificationEvents(verificationEvents); + } + + const { + HEALTH_SCORE_UPDATED_EVENT_TYPES, + processHealthScoreEvents, + } = await import('./health-score-indexer.service'); + const healthScoreEvents = events.filter(event => + HEALTH_SCORE_UPDATED_EVENT_TYPES.has(event.eventType) + ); + if (healthScoreEvents.length > 0) { + await processHealthScoreEvents(healthScoreEvents); + } + if (options.advanceCursor !== false) await advanceIndexerCursorForBatch(events); } diff --git a/src/modules/indexer/issuer-verification-indexer.service.test.ts b/src/modules/indexer/issuer-verification-indexer.service.test.ts new file mode 100644 index 00000000..0fa575b6 --- /dev/null +++ b/src/modules/indexer/issuer-verification-indexer.service.test.ts @@ -0,0 +1,106 @@ +import { processIssuerVerificationEvents } from './issuer-verification-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', () => { + const tx = { + issuerVerificationEvent: { createMany: jest.fn() }, + issuerVerification: { upsert: jest.fn() }, + }; + return { + prisma: { + $transaction: jest.fn(async (callback: (client: unknown) => unknown) => + callback(tx) + ), + __tx: tx, + }, + }; +}); + +const tx = (prisma as unknown as { __tx: any }).__tx; + +const WALLET = 'GBRPYHIL2CI3FNQ4BXLFMNDLFJUNPU2HY3ZMFSHONUCEOASW7QC7OX2H'; +const base = { txHash: 'tx-v', eventIndex: 0, ledger: 900, timestamp: 1_800_000_000 }; + +beforeEach(() => { + jest.clearAllMocks(); + tx.issuerVerificationEvent.createMany.mockResolvedValue({ count: 1 }); + tx.issuerVerification.upsert.mockResolvedValue({}); +}); + +describe('processIssuerVerificationEvents', () => { + it('marks the issuer verified with tier and expiry on VerificationGranted', async () => { + await processIssuerVerificationEvents([ + { + ...base, + eventType: 'VerificationGranted', + wallet: WALLET, + tier: 'GOLD', + expiry_timestamp: 1_900_000_000, + }, + ]); + + expect(tx.issuerVerification.upsert).toHaveBeenCalledWith({ + where: { wallet: WALLET }, + create: { + wallet: WALLET, + isVerified: true, + tier: 'GOLD', + expiresAt: new Date(1_900_000_000 * 1000), + lastLedger: 900, + }, + update: { + isVerified: true, + tier: 'GOLD', + expiresAt: new Date(1_900_000_000 * 1000), + lastLedger: 900, + }, + }); + expect(cacheInvalidate).toHaveBeenCalledWith( + `issuers:verification:${WALLET}`, + 'issuers:verified:*' + ); + }); + + it('clears verification and tier on VerificationRevoked', async () => { + await processIssuerVerificationEvents([ + { ...base, eventType: 'VerificationRevoked', wallet: WALLET, tier: 'GOLD' }, + ]); + + expect(tx.issuerVerification.upsert).toHaveBeenCalledWith( + expect.objectContaining({ + update: expect.objectContaining({ + isVerified: false, + tier: null, + expiresAt: null, + }), + }) + ); + }); + + it('leaves state untouched when the event was already recorded', async () => { + tx.issuerVerificationEvent.createMany.mockResolvedValue({ count: 0 }); + + await processIssuerVerificationEvents([ + { ...base, eventType: 'VerificationGranted', wallet: WALLET, tier: 'GOLD' }, + ]); + + expect(tx.issuerVerification.upsert).not.toHaveBeenCalled(); + }); + + it('skips a grant without a wallet', async () => { + await processIssuerVerificationEvents([ + { ...base, eventType: 'VerificationGranted', tier: 'GOLD' }, + ]); + + expect(prisma.$transaction).not.toHaveBeenCalled(); + }); +}); diff --git a/src/modules/indexer/issuer-verification-indexer.service.ts b/src/modules/indexer/issuer-verification-indexer.service.ts new file mode 100644 index 00000000..a8304717 --- /dev/null +++ b/src/modules/indexer/issuer-verification-indexer.service.ts @@ -0,0 +1,120 @@ +// src/modules/indexer/issuer-verification-indexer.service.ts +// Indexes VerificationGranted and VerificationRevoked contract events (#1179). +// +// Each event is appended to issuer_verification_events once per chain +// position. Only a newly inserted event updates the issuer_verification state +// row, so replays cannot flip a status. Events are applied in chain order, +// which the indexer pipeline guarantees. + +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 { invalidateIssuerVerificationCache } from '../issuers/issuer-verification.service'; +import { chainTimestamp } from './invoice-protocol.events'; + +export const VERIFICATION_GRANTED_EVENT_TYPES = new Set([ + 'VerificationGranted', + 'VERIFICATION_GRANTED', +]); +export const VERIFICATION_REVOKED_EVENT_TYPES = new Set([ + 'VerificationRevoked', + 'VERIFICATION_REVOKED', +]); +export const ISSUER_VERIFICATION_EVENT_TYPES = new Set([ + ...VERIFICATION_GRANTED_EVENT_TYPES, + ...VERIFICATION_REVOKED_EVENT_TYPES, +]); + +const verificationPayload = z.object({ + wallet: eventIdentifierSchema, + /** Only meaningful on a grant; a revoke clears the tier. */ + tier: eventIdentifierSchema.nullable().optional(), +}); + +export async function processIssuerVerificationEvents( + events: IndexerChainEvent[] +): Promise { + await processIndexerChainEvents(events, async event => { + const isGrant = VERIFICATION_GRANTED_EVENT_TYPES.has(event.eventType); + const isRevoke = VERIFICATION_REVOKED_EVENT_TYPES.has(event.eventType); + if (!isGrant && !isRevoke) { + return; + } + + const position = chainPositionSchema.safeParse(event); + const payload = verificationPayload.safeParse({ + wallet: event.wallet ?? event.issuer, + tier: event.tier, + }); + if (!position.success || !payload.success) { + logger.warn( + { + eventId: `${event.txHash}:${event.eventIndex}`, + eventType: event.eventType, + }, + 'Skipping verification event with missing or invalid fields' + ); + return; + } + + const wallet = payload.data.wallet; + const tier = isGrant ? (payload.data.tier ?? null) : null; + const expiresAt = isGrant + ? parseExpiry(event.expiry_timestamp ?? event.expiresAt) + : null; + + await prisma.$transaction(async tx => { + const { count } = await tx.issuerVerificationEvent.createMany({ + data: [ + { + wallet, + eventType: event.eventType, + tier, + expiresAt, + occurredAt: eventOccurredAt(event), + ...position.data, + }, + ], + skipDuplicates: true, + }); + if (count === 0) { + return; + } + await tx.issuerVerification.upsert({ + where: { wallet }, + create: { + wallet, + isVerified: isGrant, + tier, + expiresAt, + lastLedger: position.data.ledger, + }, + update: { + isVerified: isGrant, + tier, + expiresAt, + lastLedger: position.data.ledger, + }, + }); + }); + + await invalidateIssuerVerificationCache(wallet); + }); +} + +/** Expiry may arrive as unix seconds or ISO-8601; absent means no expiry. */ +function parseExpiry(value: unknown): Date | null { + if (value === undefined || value === null || value === '') { + return null; + } + return chainTimestamp(value); +} diff --git a/src/modules/indexer/treasury-indexer.service.test.ts b/src/modules/indexer/treasury-indexer.service.test.ts new file mode 100644 index 00000000..98d1cccc --- /dev/null +++ b/src/modules/indexer/treasury-indexer.service.test.ts @@ -0,0 +1,123 @@ +import { processTreasuryEvents } from './treasury-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', () => { + const tx = { + treasuryInflow: { createMany: jest.fn() }, + treasuryBalance: { upsert: jest.fn() }, + }; + return { + prisma: { + $transaction: jest.fn(async (callback: (client: unknown) => unknown) => + callback(tx) + ), + treasuryAllocation: { createMany: jest.fn() }, + __tx: tx, + }, + }; +}); + +const tx = (prisma as unknown as { __tx: any }).__tx; + +const feeEvent = { + txHash: 'tx-fee', + eventIndex: 0, + ledger: 500, + eventType: 'FeeReceived', + source: 'trading', + amount: '250', + timestamp: 1_800_000_000, +}; + +beforeEach(() => { + jest.clearAllMocks(); + tx.treasuryInflow.createMany.mockResolvedValue({ count: 1 }); + tx.treasuryBalance.upsert.mockResolvedValue({}); + (prisma.treasuryAllocation.createMany as jest.Mock).mockResolvedValue({ + count: 1, + }); +}); + +describe('processTreasuryEvents', () => { + it('appends the inflow and increments the source balance for a new FeeReceived', async () => { + await processTreasuryEvents([feeEvent]); + + expect(tx.treasuryInflow.createMany).toHaveBeenCalledWith({ + data: [ + expect.objectContaining({ + source: 'trading', + amount: '250', + txHash: 'tx-fee', + eventIndex: 0, + ledger: 500, + }), + ], + skipDuplicates: true, + }); + expect(tx.treasuryBalance.upsert).toHaveBeenCalledWith({ + where: { source: 'trading' }, + create: { source: 'trading', balance: '250' }, + update: { balance: { increment: '250' } }, + }); + expect(cacheInvalidate).toHaveBeenCalledWith('treasury:*'); + }); + + it('does not change any balance when the inflow is a replay', async () => { + tx.treasuryInflow.createMany.mockResolvedValue({ count: 0 }); + + await processTreasuryEvents([feeEvent]); + + expect(tx.treasuryBalance.upsert).not.toHaveBeenCalled(); + }); + + it('records FundsAllocated as an outflow without touching balances', async () => { + await processTreasuryEvents([ + { + txHash: 'tx-alloc', + eventIndex: 1, + ledger: 501, + eventType: 'FundsAllocated', + destination: 'GDESTINATION', + amount: 100, + purpose: 'grants', + timestamp: 1_800_000_100, + }, + ]); + + expect(prisma.treasuryAllocation.createMany).toHaveBeenCalledWith({ + data: [ + expect.objectContaining({ + destination: 'GDESTINATION', + amount: '100', + purpose: 'grants', + }), + ], + skipDuplicates: true, + }); + expect(tx.treasuryBalance.upsert).not.toHaveBeenCalled(); + }); + + it('skips a FeeReceived with a non-integer amount', async () => { + await processTreasuryEvents([{ ...feeEvent, amount: '2.5' }]); + + expect(tx.treasuryInflow.createMany).not.toHaveBeenCalled(); + expect(prisma.$transaction).not.toHaveBeenCalled(); + }); + + it('ignores unrelated event types', async () => { + await processTreasuryEvents([ + { ...feeEvent, eventType: 'KEY_BOUGHT' }, + ]); + + expect(prisma.$transaction).not.toHaveBeenCalled(); + }); +}); diff --git a/src/modules/indexer/treasury-indexer.service.ts b/src/modules/indexer/treasury-indexer.service.ts new file mode 100644 index 00000000..5f34d35d --- /dev/null +++ b/src/modules/indexer/treasury-indexer.service.ts @@ -0,0 +1,143 @@ +// src/modules/indexer/treasury-indexer.service.ts +// Indexes FeeReceived and FundsAllocated treasury contract events (#1181). +// +// - FeeReceived appends a TreasuryInflow row and, only when that row is new, +// increments the per-source TreasuryBalance in the same transaction. A +// replayed event inserts nothing and therefore changes no balance. +// - FundsAllocated appends a TreasuryAllocation row. Allocations reduce the +// total balance at read time. +// - Every write invalidates the treasury read 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, + unitAmountSchema, +} from '../../utils/indexer-event-fields.utils'; +import { invalidateTreasuryCache } from '../treasury/treasury.service'; + +export const FEE_RECEIVED_EVENT_TYPES = new Set([ + 'FeeReceived', + 'FEE_RECEIVED', +]); +export const FUNDS_ALLOCATED_EVENT_TYPES = new Set([ + 'FundsAllocated', + 'FUNDS_ALLOCATED', +]); +export const TREASURY_EVENT_TYPES = new Set([ + ...FEE_RECEIVED_EVENT_TYPES, + ...FUNDS_ALLOCATED_EVENT_TYPES, +]); + +const feeReceivedPayload = z.object({ + source: eventIdentifierSchema, + amount: unitAmountSchema, +}); + +const fundsAllocatedPayload = z.object({ + destination: eventIdentifierSchema, + amount: unitAmountSchema, + purpose: z.string().max(1024).default(''), +}); + +/** + * Applies a batch of treasury events in chain order. Each event is handled + * exactly once per unique chain position; see the module comment above. + */ +export async function processTreasuryEvents( + events: IndexerChainEvent[] +): Promise { + await processIndexerChainEvents(events, async event => { + if (FEE_RECEIVED_EVENT_TYPES.has(event.eventType)) { + await recordFeeReceived(event); + return; + } + if (FUNDS_ALLOCATED_EVENT_TYPES.has(event.eventType)) { + await recordFundsAllocated(event); + } + }); +} + +async function recordFeeReceived(event: IndexerChainEvent): Promise { + const position = chainPositionSchema.safeParse(event); + const payload = feeReceivedPayload.safeParse({ + source: event.source ?? event.payer, + amount: event.amount, + }); + if (!position.success || !payload.success) { + logger.warn( + { + eventId: `${event.txHash}:${event.eventIndex}`, + eventType: event.eventType, + }, + 'Skipping FeeReceived event with missing or invalid fields' + ); + return; + } + + const { source, amount } = payload.data; + const row = { + source, + amount: amount.toString(), + occurredAt: eventOccurredAt(event), + ...position.data, + }; + + await prisma.$transaction(async tx => { + const { count } = await tx.treasuryInflow.createMany({ + data: [row], + skipDuplicates: true, + }); + if (count === 0) { + return; + } + await tx.treasuryBalance.upsert({ + where: { source }, + create: { source, balance: row.amount }, + update: { balance: { increment: row.amount } }, + }); + }); + + await invalidateTreasuryCache(); +} + +async function recordFundsAllocated(event: IndexerChainEvent): Promise { + const position = chainPositionSchema.safeParse(event); + const payload = fundsAllocatedPayload.safeParse({ + destination: event.destination ?? event.to, + amount: event.amount, + purpose: event.purpose, + }); + if (!position.success || !payload.success) { + logger.warn( + { + eventId: `${event.txHash}:${event.eventIndex}`, + eventType: event.eventType, + }, + 'Skipping FundsAllocated event with missing or invalid fields' + ); + return; + } + + await prisma.treasuryAllocation.createMany({ + data: [ + { + destination: payload.data.destination, + amount: payload.data.amount.toString(), + purpose: payload.data.purpose, + occurredAt: eventOccurredAt(event), + ...position.data, + }, + ], + skipDuplicates: true, + }); + + await invalidateTreasuryCache(); +} diff --git a/src/modules/invoice/invoice-early-incentive.routes.ts b/src/modules/invoice/invoice-early-incentive.routes.ts new file mode 100644 index 00000000..e73d64fd --- /dev/null +++ b/src/modules/invoice/invoice-early-incentive.routes.ts @@ -0,0 +1,44 @@ +// src/modules/invoice/invoice-early-incentive.routes.ts +// GET /invoices/:invoice_id/early-incentive (#1180). Public read; mounted on +// invoiceRouter. + +import { Router } from 'express'; +import { z } from 'zod'; +import { + sendNotFound, + sendSuccess, + sendValidationError, + zodIssuesToDetails, +} from '../../utils/api-response.utils'; +import { getEarlyIncentiveStatus } from './invoice-early-incentive.service'; + +export const invoiceEarlyIncentiveRouter = Router(); + +const paramsSchema = z.object({ + invoice_id: z.string().min(1).max(128), +}); + +invoiceEarlyIncentiveRouter.get( + '/:invoice_id/early-incentive', + 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 status = await getEarlyIncentiveStatus(params.data.invoice_id); + if (!status) { + sendNotFound(res, 'Invoice'); + return; + } + sendSuccess(res, status); + } catch (error) { + next(error); + } + } +); diff --git a/src/modules/invoice/invoice-early-incentive.service.test.ts b/src/modules/invoice/invoice-early-incentive.service.test.ts new file mode 100644 index 00000000..9ffd21dd --- /dev/null +++ b/src/modules/invoice/invoice-early-incentive.service.test.ts @@ -0,0 +1,71 @@ +import { buildEarlyIncentiveStatus } from './invoice-early-incentive.service'; + +jest.mock('../../utils/redis.utils', () => ({ + cacheGetJson: jest.fn().mockResolvedValue(null), + cacheSetJson: jest.fn().mockResolvedValue(undefined), + cacheInvalidate: jest.fn().mockResolvedValue(undefined), +})); + +jest.mock('../../utils/prisma.utils', () => ({ prisma: {} })); + +describe('buildEarlyIncentiveStatus', () => { + it('reports inactive for an invoice with no incentive configured', () => { + expect( + buildEarlyIncentiveStatus({ + invoiceId: 'inv-1', + slotsTotal: 0, + bonusBps: 0, + slotsFilledEvents: 0, + }) + ).toEqual({ + invoice_id: 'inv-1', + slots_total: 0, + slots_filled: 0, + slots_remaining: 0, + bonus_bps: 0, + is_active: false, + }); + }); + + it('derives remaining slots from the fill count', () => { + expect( + buildEarlyIncentiveStatus({ + invoiceId: 'inv-1', + slotsTotal: 10, + bonusBps: 250, + slotsFilledEvents: 4, + }) + ).toEqual({ + invoice_id: 'inv-1', + slots_total: 10, + slots_filled: 4, + slots_remaining: 6, + bonus_bps: 250, + is_active: true, + }); + }); + + it('is inactive once every slot is filled', () => { + const status = buildEarlyIncentiveStatus({ + invoiceId: 'inv-1', + slotsTotal: 3, + bonusBps: 100, + slotsFilledEvents: 3, + }); + + expect(status.slots_remaining).toBe(0); + expect(status.is_active).toBe(false); + }); + + it('never reports more slots filled than the total', () => { + const status = buildEarlyIncentiveStatus({ + invoiceId: 'inv-1', + slotsTotal: 2, + bonusBps: 100, + slotsFilledEvents: 5, + }); + + expect(status.slots_filled).toBe(2); + expect(status.slots_remaining).toBe(0); + }); +}); diff --git a/src/modules/invoice/invoice-early-incentive.service.ts b/src/modules/invoice/invoice-early-incentive.service.ts new file mode 100644 index 00000000..825e3d65 --- /dev/null +++ b/src/modules/invoice/invoice-early-incentive.service.ts @@ -0,0 +1,85 @@ +// src/modules/invoice/invoice-early-incentive.service.ts +// Early funding incentive slot status for an invoice (#1180). +// +// slots_filled is the number of indexed EarlySlotFilled events for the +// invoice, capped at slots_total. The incentive configuration (slots_total, +// bonus_bps) is taken from the most recent event, so the status reflects the +// latest on-chain configuration. + +import { prisma } from '../../utils/prisma.utils'; +import { cacheGetJson, cacheInvalidate, cacheSetJson } from '../../utils/redis.utils'; + +export const EARLY_INCENTIVE_CACHE_TTL_SECONDS = 30; + +export interface EarlyIncentiveStatus { + invoice_id: string; + slots_total: number; + slots_filled: number; + slots_remaining: number; + bonus_bps: number; + is_active: boolean; +} + +export function earlyIncentiveCacheKey(invoiceId: string): string { + return `invoice:early-incentive:${invoiceId}`; +} + +export async function invalidateEarlyIncentiveCache( + invoiceId: string +): Promise { + await cacheInvalidate(earlyIncentiveCacheKey(invoiceId)); +} + +/** Derives the public status from the filled count and latest configuration. */ +export function buildEarlyIncentiveStatus(input: { + invoiceId: string; + slotsTotal: number; + bonusBps: number; + slotsFilledEvents: number; +}): EarlyIncentiveStatus { + const slotsFilled = Math.min(input.slotsFilledEvents, input.slotsTotal); + const slotsRemaining = Math.max(input.slotsTotal - slotsFilled, 0); + return { + invoice_id: input.invoiceId, + slots_total: input.slotsTotal, + slots_filled: slotsFilled, + slots_remaining: slotsRemaining, + bonus_bps: input.bonusBps, + is_active: input.slotsTotal > 0 && slotsRemaining > 0, + }; +} + +/** + * Returns the incentive status, or null when the invoice does not exist. An + * existing invoice with no configured incentive reports is_active: false. + */ +export async function getEarlyIncentiveStatus( + invoiceId: string +): Promise { + const cacheKey = earlyIncentiveCacheKey(invoiceId); + const cached = await cacheGetJson(cacheKey); + if (cached) return cached; + + const invoice = await prisma.invoice.findUnique({ + where: { id: invoiceId }, + select: { id: true }, + }); + if (!invoice) return null; + + const [latest, slotsFilledEvents] = await Promise.all([ + prisma.invoiceEarlySlotEvent.findFirst({ + where: { invoiceId }, + orderBy: [{ ledger: 'desc' }, { eventIndex: 'desc' }], + }), + prisma.invoiceEarlySlotEvent.count({ where: { invoiceId } }), + ]); + + const status = buildEarlyIncentiveStatus({ + invoiceId, + slotsTotal: latest?.slotsTotal ?? 0, + bonusBps: latest?.bonusBps ?? 0, + slotsFilledEvents, + }); + await cacheSetJson(cacheKey, status, EARLY_INCENTIVE_CACHE_TTL_SECONDS); + return status; +} diff --git a/src/modules/invoice/invoice.routes.ts b/src/modules/invoice/invoice.routes.ts index 4198ec60..9a9bf840 100644 --- a/src/modules/invoice/invoice.routes.ts +++ b/src/modules/invoice/invoice.routes.ts @@ -5,6 +5,7 @@ import { INVOICE_COMPARISON_CACHE_TTL_SECONDS } from './invoice.schemas'; import { httpGetInvoiceComparison } from './invoice.controllers'; import { invoiceProtocolRouter } from './invoice-protocol.routes'; import { invoiceMarketRouter } from './invoice-market.routes'; +import { invoiceEarlyIncentiveRouter } from './invoice-early-incentive.routes'; const invoiceRouter = Router(); @@ -25,5 +26,6 @@ invoiceRouter.get( invoiceRouter.use(invoiceProtocolRouter); invoiceRouter.use(invoiceMarketRouter); +invoiceRouter.use(invoiceEarlyIncentiveRouter); export default invoiceRouter; diff --git a/src/modules/issuers/issuer-verification.routes.ts b/src/modules/issuers/issuer-verification.routes.ts new file mode 100644 index 00000000..cf290c6e --- /dev/null +++ b/src/modules/issuers/issuer-verification.routes.ts @@ -0,0 +1,71 @@ +// src/modules/issuers/issuer-verification.routes.ts +// Public verification badge reads for issuers (#1179). Mounted at /issuers +// alongside the existing reputation routes. + +import { Router } from 'express'; +import { StrKey } from '@stellar/stellar-base'; +import { z } from 'zod'; +import { + ErrorCode, + sendError, + sendPaginatedSuccess, + sendSuccess, + sendValidationError, + zodIssuesToDetails, +} from '../../utils/api-response.utils'; +import { + getIssuerVerificationStatus, + listVerifiedIssuers, +} from './issuer-verification.service'; + +export const issuerVerificationRouter = Router(); + +const pageQuerySchema = z.object({ + page: z.coerce.number().int().min(1).max(100000).default(1), + limit: z.coerce.number().int().min(1).max(100).default(20), +}); + +const walletSchema = z + .string() + .refine(StrKey.isValidEd25519PublicKey, 'Invalid Stellar wallet'); + +/** GET /issuers/verified — paginated list of currently verified issuers. */ +issuerVerificationRouter.get('/verified', async (req, res, next) => { + const query = pageQuerySchema.safeParse(req.query); + if (!query.success) { + sendValidationError( + res, + 'Invalid query', + zodIssuesToDetails(query.error.issues) + ); + return; + } + try { + const { data, meta } = await listVerifiedIssuers(query.data); + sendPaginatedSuccess(res, data, meta); + } catch (error) { + next(error); + } +}); + +/** GET /issuers/:wallet/verification — verification status for one issuer. */ +issuerVerificationRouter.get( + '/:wallet/verification', + async (req, res, next) => { + const wallet = walletSchema.safeParse(req.params.wallet); + if (!wallet.success) { + sendError( + res, + 400, + ErrorCode.VALIDATION_ERROR, + 'Invalid Stellar wallet' + ); + return; + } + try { + sendSuccess(res, await getIssuerVerificationStatus(wallet.data)); + } catch (error) { + next(error); + } + } +); diff --git a/src/modules/issuers/issuer-verification.service.test.ts b/src/modules/issuers/issuer-verification.service.test.ts new file mode 100644 index 00000000..13f10535 --- /dev/null +++ b/src/modules/issuers/issuer-verification.service.test.ts @@ -0,0 +1,74 @@ +import { toVerificationStatus } from './issuer-verification.service'; + +jest.mock('../../utils/redis.utils', () => ({ + cacheGetJson: jest.fn().mockResolvedValue(null), + cacheSetJson: jest.fn().mockResolvedValue(undefined), + cacheInvalidate: jest.fn().mockResolvedValue(undefined), +})); + +jest.mock('../../utils/prisma.utils', () => ({ prisma: {} })); + +const WALLET = 'GWALLET'; +const NOW = new Date('2026-10-21T00:00:00.000Z'); +const DAY_MS = 86_400_000; + +describe('toVerificationStatus', () => { + it('reports unverified for a wallet with no record', () => { + expect(toVerificationStatus(WALLET, null, NOW)).toEqual({ + wallet: WALLET, + is_verified: false, + tier: null, + expiry_timestamp: null, + days_remaining: null, + }); + }); + + it('reports a non-expiring grant as verified with no day count', () => { + const status = toVerificationStatus( + WALLET, + { isVerified: true, tier: 'GOLD', expiresAt: null }, + NOW + ); + + expect(status.is_verified).toBe(true); + expect(status.tier).toBe('GOLD'); + expect(status.days_remaining).toBeNull(); + }); + + it('rounds days remaining up for an active expiring grant', () => { + const expiresAt = new Date(NOW.getTime() + 2.5 * DAY_MS); + const status = toVerificationStatus( + WALLET, + { isVerified: true, tier: 'SILVER', expiresAt }, + NOW + ); + + expect(status.is_verified).toBe(true); + expect(status.days_remaining).toBe(3); + expect(status.expiry_timestamp).toBe(expiresAt.toISOString()); + }); + + it('reports an expired grant as unverified with zero days remaining', () => { + const expiresAt = new Date(NOW.getTime() - DAY_MS); + const status = toVerificationStatus( + WALLET, + { isVerified: true, tier: 'GOLD', expiresAt }, + NOW + ); + + expect(status.is_verified).toBe(false); + expect(status.tier).toBeNull(); + expect(status.days_remaining).toBe(0); + }); + + it('reports a revoked issuer as unverified', () => { + const status = toVerificationStatus( + WALLET, + { isVerified: false, tier: null, expiresAt: null }, + NOW + ); + + expect(status.is_verified).toBe(false); + expect(status.days_remaining).toBeNull(); + }); +}); diff --git a/src/modules/issuers/issuer-verification.service.ts b/src/modules/issuers/issuer-verification.service.ts new file mode 100644 index 00000000..d4d7be94 --- /dev/null +++ b/src/modules/issuers/issuer-verification.service.ts @@ -0,0 +1,145 @@ +// src/modules/issuers/issuer-verification.service.ts +// Verification badge status for issuers (#1179). +// +// An issuer is verified when the latest event is a grant and the grant has not +// expired. A grant with no expiry never expires. A revoked or expired issuer +// reports is_verified: false with tier: null. + +import { prisma } from '../../utils/prisma.utils'; +import { + cacheGetJson, + cacheInvalidate, + cacheSetJson, +} from '../../utils/redis.utils'; +import { buildPaginationMetadata } from '../../utils/indexer-event-fields.utils'; +import { PaginationMetadata } from '../../utils/api-response.utils'; + +export const VERIFICATION_CACHE_TTL_SECONDS = 60; +const VERIFIED_LIST_CACHE_PATTERN = 'issuers:verified:*'; +const DAY_MS = 86_400_000; + +export interface IssuerVerificationStatus { + wallet: string; + is_verified: boolean; + tier: string | null; + expiry_timestamp: string | null; + days_remaining: number | null; +} + +export interface VerifiedIssuerView { + wallet: string; + tier: string | null; + expiry_timestamp: string | null; + days_remaining: number | null; +} + +export interface VerificationRecord { + isVerified: boolean; + tier: string | null; + expiresAt: Date | null; +} + +export function verificationCacheKey(wallet: string): string { + return `issuers:verification:${wallet}`; +} + +/** Clears the cached status for one wallet and every cached verified-list page. */ +export async function invalidateIssuerVerificationCache( + wallet: string +): Promise { + await cacheInvalidate(verificationCacheKey(wallet), VERIFIED_LIST_CACHE_PATTERN); +} + +/** Pure status derivation from the stored record, relative to `now`. */ +export function toVerificationStatus( + wallet: string, + record: VerificationRecord | null, + now: Date = new Date() +): IssuerVerificationStatus { + const expiresAt = record?.expiresAt ?? null; + const expired = expiresAt !== null && expiresAt.getTime() <= now.getTime(); + const active = Boolean(record?.isVerified) && !expired; + + if (!active) { + return { + wallet, + is_verified: false, + tier: null, + expiry_timestamp: expiresAt?.toISOString() ?? null, + days_remaining: expired ? 0 : null, + }; + } + + return { + wallet, + is_verified: true, + tier: record?.tier ?? null, + expiry_timestamp: expiresAt?.toISOString() ?? null, + days_remaining: expiresAt + ? Math.ceil((expiresAt.getTime() - now.getTime()) / DAY_MS) + : null, + }; +} + +export async function getIssuerVerificationStatus( + wallet: string +): Promise { + const cacheKey = verificationCacheKey(wallet); + const cached = await cacheGetJson(cacheKey); + if (cached) return cached; + + const record = await prisma.issuerVerification.findUnique({ + where: { wallet }, + select: { isVerified: true, tier: true, expiresAt: true }, + }); + const status = toVerificationStatus(wallet, record); + await cacheSetJson(cacheKey, status, VERIFICATION_CACHE_TTL_SECONDS); + return status; +} + +export async function listVerifiedIssuers(params: { + page: number; + limit: number; +}): Promise<{ data: VerifiedIssuerView[]; meta: PaginationMetadata }> { + const cacheKey = `issuers:verified:${params.page}:${params.limit}`; + const cached = await cacheGetJson<{ + data: VerifiedIssuerView[]; + meta: PaginationMetadata; + }>(cacheKey); + if (cached) return cached; + + const now = new Date(); + const where = { + isVerified: true, + OR: [{ expiresAt: null }, { expiresAt: { gt: now } }], + }; + const [rows, totalCount] = await prisma.$transaction([ + prisma.issuerVerification.findMany({ + where, + orderBy: { wallet: 'asc' }, + skip: (params.page - 1) * params.limit, + take: params.limit, + select: { wallet: true, tier: true, expiresAt: true }, + }), + prisma.issuerVerification.count({ where }), + ]); + + const result = { + data: rows.map(row => { + const status = toVerificationStatus( + row.wallet, + { isVerified: true, tier: row.tier, expiresAt: row.expiresAt }, + now + ); + return { + wallet: row.wallet, + tier: status.tier, + expiry_timestamp: status.expiry_timestamp, + days_remaining: status.days_remaining, + }; + }), + meta: buildPaginationMetadata({ ...params, totalCount }), + }; + await cacheSetJson(cacheKey, result, VERIFICATION_CACHE_TTL_SECONDS); + return result; +} diff --git a/src/modules/portfolio/portfolio-health-score.routes.ts b/src/modules/portfolio/portfolio-health-score.routes.ts new file mode 100644 index 00000000..4cd54ad0 --- /dev/null +++ b/src/modules/portfolio/portfolio-health-score.routes.ts @@ -0,0 +1,52 @@ +// src/modules/portfolio/portfolio-health-score.routes.ts +// GET /investors/:wallet/health-score (#1178). Public read of public on-chain +// data; mounted at /investors. + +import { Router } from 'express'; +import { StrKey } from '@stellar/stellar-base'; +import { z } from 'zod'; +import { + ErrorCode, + sendError, + sendSuccess, + sendValidationError, + zodIssuesToDetails, +} from '../../utils/api-response.utils'; +import { + getInvestorHealthScore, + HealthScoreUnavailableError, +} from './portfolio-health-score.service'; + +export const investorHealthRouter = Router(); + +const paramsSchema = z.object({ + wallet: z + .string() + .refine(StrKey.isValidEd25519PublicKey, 'Invalid Stellar wallet'), +}); + +investorHealthRouter.get('/:wallet/health-score', async (req, res, next) => { + const params = paramsSchema.safeParse(req.params); + if (!params.success) { + sendValidationError( + res, + 'Invalid wallet', + zodIssuesToDetails(params.error.issues) + ); + return; + } + try { + sendSuccess(res, await getInvestorHealthScore(params.data.wallet)); + } catch (error) { + if (error instanceof HealthScoreUnavailableError) { + sendError( + res, + 503, + ErrorCode.SERVICE_UNAVAILABLE, + error.message + ); + return; + } + next(error); + } +}); diff --git a/src/modules/portfolio/portfolio-health-score.service.test.ts b/src/modules/portfolio/portfolio-health-score.service.test.ts new file mode 100644 index 00000000..7c79d079 --- /dev/null +++ b/src/modules/portfolio/portfolio-health-score.service.test.ts @@ -0,0 +1,61 @@ +import { + HealthScoreUnavailableError, + parseHealthScoreResult, +} from './portfolio-health-score.service'; + +jest.mock('../../utils/redis.utils', () => ({ + cacheGetJson: jest.fn().mockResolvedValue(null), + cacheSetJson: jest.fn().mockResolvedValue(undefined), + cacheInvalidate: jest.fn().mockResolvedValue(undefined), +})); + +const validRow = { + score: 742, + concentration_score: 80, + maturity_distribution_score: 70, + issuer_diversity_score: 65, + default_exposure_score: 90, +}; + +describe('parseHealthScoreResult', () => { + it('returns null score and factors for a wallet with no portfolio', () => { + expect(parseHealthScoreResult(null)).toEqual({ score: null, factors: null }); + expect(parseHealthScoreResult(undefined)).toEqual({ + score: null, + factors: null, + }); + }); + + it('maps the composite score and all four factor scores', () => { + expect(parseHealthScoreResult(validRow)).toEqual({ + score: 742, + factors: { + concentration_score: 80, + maturity_distribution_score: 70, + issuer_diversity_score: 65, + default_exposure_score: 90, + }, + }); + }); + + it('accepts bigint values from the XDR decoder', () => { + const parsed = parseHealthScoreResult({ + ...validRow, + score: BigInt(742), + }); + + expect(parsed.score).toBe(742); + }); + + it('treats a response missing a factor as unavailable', () => { + expect(() => + parseHealthScoreResult({ ...validRow, default_exposure_score: undefined }) + ).toThrow(HealthScoreUnavailableError); + }); + + it('treats a negative score as unavailable', () => { + expect(() => + parseHealthScoreResult({ ...validRow, score: -1 }) + ).toThrow(HealthScoreUnavailableError); + }); +}); diff --git a/src/modules/portfolio/portfolio-health-score.service.ts b/src/modules/portfolio/portfolio-health-score.service.ts new file mode 100644 index 00000000..44591af6 --- /dev/null +++ b/src/modules/portfolio/portfolio-health-score.service.ts @@ -0,0 +1,173 @@ +// src/modules/portfolio/portfolio-health-score.service.ts +// Portfolio health score for a wallet (#1178). The score is computed on chain +// by the health-score contract; this service reads it with a read-only +// simulation and caches the result for five minutes. HealthScoreUpdated events +// clear the cache for the affected wallet. +// +// A wallet with no portfolio returns score: null and factors: null. + +import { + Account, + Address, + Contract, + Networks, + TransactionBuilder, + scValToNative, + xdr, +} from '@stellar/stellar-base'; +import { envConfig } from '../../config'; +import { defaultSorobanTransport } from '../../clients/soroban.client'; +import { + cacheGetJson, + cacheInvalidate, + cacheSetJson, +} from '../../utils/redis.utils'; + +export const HEALTH_SCORE_CACHE_TTL_SECONDS = 300; +const U32_MAX = 4_294_967_295; + +export class HealthScoreUnavailableError extends Error { + constructor(message = 'Portfolio health score is unavailable') { + super(message); + this.name = 'HealthScoreUnavailableError'; + } +} + +export interface HealthScoreFactors { + concentration_score: number; + maturity_distribution_score: number; + issuer_diversity_score: number; + default_exposure_score: number; +} + +export interface InvestorHealthScore { + wallet: string; + score: number | null; + factors: HealthScoreFactors | null; + computed_at: string; +} + +export function healthScoreCacheKey(wallet: string): string { + return `investors:health-score:${wallet}`; +} + +export async function invalidateHealthScoreCache(wallet: string): Promise { + await cacheInvalidate(healthScoreCacheKey(wallet)); +} + +function toScore(value: unknown): number | null { + const numeric = typeof value === 'bigint' ? Number(value) : value; + return typeof numeric === 'number' && + Number.isSafeInteger(numeric) && + numeric >= 0 && + numeric <= U32_MAX + ? numeric + : null; +} + +/** + * Decodes the contract's return value. null or undefined means the wallet has + * no portfolio. Any other shape must carry the composite score and all four + * factor scores, or the read is treated as unavailable. + */ +export function parseHealthScoreResult( + value: unknown +): Pick { + if (value === null || value === undefined) { + return { score: null, factors: null }; + } + + const row = value as Record; + const score = toScore(row.score); + const factors = { + concentration_score: toScore(row.concentration_score), + maturity_distribution_score: toScore(row.maturity_distribution_score), + issuer_diversity_score: toScore(row.issuer_diversity_score), + default_exposure_score: toScore(row.default_exposure_score), + }; + if ( + score === null || + factors.concentration_score === null || + factors.maturity_distribution_score === null || + factors.issuer_diversity_score === null || + factors.default_exposure_score === null + ) { + throw new HealthScoreUnavailableError('Invalid health score response'); + } + return { + score, + factors: factors as HealthScoreFactors, + }; +} + +/** Read-only simulation of get_health_score. Never signs or submits. */ +async function simulateHealthScore(wallet: string): Promise { + const contractId = envConfig.PORTFOLIO_HEALTH_CONTRACT_ID; + if (!contractId) { + throw new HealthScoreUnavailableError( + 'Health score contract is not configured' + ); + } + + const networkPassphrase = + envConfig.STELLAR_NETWORK === 'mainnet' ? Networks.PUBLIC : Networks.TESTNET; + const transaction = new TransactionBuilder(new Account(wallet, '0'), { + fee: '100', + networkPassphrase, + }) + .addOperation( + new Contract(contractId).call( + 'get_health_score', + new Address(wallet).toScVal() + ) + ) + .setTimeout(300) + .build() + .toXDR(); + + try { + const envelope = (await defaultSorobanTransport( + envConfig.STELLAR_SOROBAN_RPC_URL, + { + jsonrpc: '2.0', + id: 'health-score', + method: 'simulateTransaction', + params: { transaction }, + } + )) as { + error?: unknown; + result?: { + error?: unknown; + results?: Array<{ xdr?: string; retval?: string }>; + }; + }; + const first = envelope.result?.results?.[0]; + const returnValue = first?.xdr ?? first?.retval; + if (envelope.error || envelope.result?.error || !returnValue) { + throw new HealthScoreUnavailableError('Simulation returned no result'); + } + return scValToNative(xdr.ScVal.fromXDR(returnValue, 'base64')); + } catch (error) { + if (error instanceof HealthScoreUnavailableError) { + throw error; + } + throw new HealthScoreUnavailableError('Unable to read get_health_score'); + } +} + +export async function getInvestorHealthScore( + wallet: string +): Promise { + const cacheKey = healthScoreCacheKey(wallet); + const cached = await cacheGetJson(cacheKey); + if (cached) return cached; + + const parsed = parseHealthScoreResult(await simulateHealthScore(wallet)); + const result: InvestorHealthScore = { + wallet, + ...parsed, + computed_at: new Date().toISOString(), + }; + await cacheSetJson(cacheKey, result, HEALTH_SCORE_CACHE_TTL_SECONDS); + return result; +} diff --git a/src/modules/treasury/treasury.routes.ts b/src/modules/treasury/treasury.routes.ts new file mode 100644 index 00000000..cad1313d --- /dev/null +++ b/src/modules/treasury/treasury.routes.ts @@ -0,0 +1,71 @@ +// src/modules/treasury/treasury.routes.ts +// Admin dashboard read routes for the platform treasury (#1181). +// Mounted under /admin/treasury behind adminGuard in admin.routes.ts. + +import { Router } from 'express'; +import { z } from 'zod'; +import { + sendPaginatedSuccess, + sendSuccess, + sendValidationError, + zodIssuesToDetails, +} from '../../utils/api-response.utils'; +import { + getTreasurySummary, + listTreasuryAllocations, + listTreasuryInflows, +} from './treasury.service'; + +export const treasuryRouter = Router(); + +const pageQuerySchema = z.object({ + page: z.coerce.number().int().min(1).max(100000).default(1), + limit: z.coerce.number().int().min(1).max(100).default(20), +}); + +/** GET /admin/treasury — total balance and per-source breakdown. */ +treasuryRouter.get('/', async (_req, res, next) => { + try { + sendSuccess(res, await getTreasurySummary()); + } catch (error) { + next(error); + } +}); + +/** GET /admin/treasury/allocations — paginated FundsAllocated history. */ +treasuryRouter.get('/allocations', async (req, res, next) => { + const query = pageQuerySchema.safeParse(req.query); + if (!query.success) { + sendValidationError( + res, + 'Invalid query', + zodIssuesToDetails(query.error.issues) + ); + return; + } + try { + const { data, meta } = await listTreasuryAllocations(query.data); + sendPaginatedSuccess(res, data, meta); + } catch (error) { + next(error); + } +}); + +/** GET /admin/treasury/inflows — paginated FeeReceived history. */ +treasuryRouter.get('/inflows', async (req, res, next) => { + const query = pageQuerySchema.safeParse(req.query); + if (!query.success) { + sendValidationError( + res, + 'Invalid query', + zodIssuesToDetails(query.error.issues) + ); + return; + } + try { + const { data, meta } = await listTreasuryInflows(query.data); + sendPaginatedSuccess(res, data, meta); + } catch (error) { + next(error); + } +}); diff --git a/src/modules/treasury/treasury.service.test.ts b/src/modules/treasury/treasury.service.test.ts new file mode 100644 index 00000000..c0ddde69 --- /dev/null +++ b/src/modules/treasury/treasury.service.test.ts @@ -0,0 +1,63 @@ +import { getTreasurySummary } from './treasury.service'; +import { prisma } from '../../utils/prisma.utils'; + +jest.mock('../../utils/redis.utils', () => ({ + cacheGetJson: jest.fn().mockResolvedValue(null), + cacheSetJson: jest.fn().mockResolvedValue(undefined), + cacheInvalidate: jest.fn().mockResolvedValue(undefined), +})); + +jest.mock('../../utils/prisma.utils', () => ({ + prisma: { + treasuryBalance: { findMany: jest.fn() }, + treasuryAllocation: { aggregate: jest.fn() }, + }, +})); + +/** Minimal stand-in for a Prisma Decimal: only toFixed is used by the service. */ +const decimal = (value: string) => ({ toFixed: () => value }); + +describe('getTreasurySummary', () => { + beforeEach(() => { + jest.clearAllMocks(); + }); + + it('nets allocations against gross inflows for total_balance', async () => { + const updatedAt = new Date('2026-10-20T10:00:00.000Z'); + (prisma.treasuryBalance.findMany as jest.Mock).mockResolvedValue([ + { source: 'fees', balance: decimal('1000'), updatedAt }, + { + source: 'trading', + balance: decimal('500'), + updatedAt: new Date('2026-10-21T10:00:00.000Z'), + }, + ]); + (prisma.treasuryAllocation.aggregate as jest.Mock).mockResolvedValue({ + _sum: { amount: decimal('300') }, + }); + + const summary = await getTreasurySummary(); + + expect(summary.total_balance).toBe('1200'); + expect(summary.per_source_breakdown).toEqual([ + { source: 'fees', balance: '1000' }, + { source: 'trading', balance: '500' }, + ]); + expect(summary.last_updated).toBe('2026-10-21T10:00:00.000Z'); + }); + + it('reports zero and null last_updated when nothing has been indexed', async () => { + (prisma.treasuryBalance.findMany as jest.Mock).mockResolvedValue([]); + (prisma.treasuryAllocation.aggregate as jest.Mock).mockResolvedValue({ + _sum: { amount: null }, + }); + + const summary = await getTreasurySummary(); + + expect(summary).toEqual({ + total_balance: '0', + per_source_breakdown: [], + last_updated: null, + }); + }); +}); diff --git a/src/modules/treasury/treasury.service.ts b/src/modules/treasury/treasury.service.ts new file mode 100644 index 00000000..9a4a8fb0 --- /dev/null +++ b/src/modules/treasury/treasury.service.ts @@ -0,0 +1,151 @@ +// src/modules/treasury/treasury.service.ts +// Read side of the platform treasury (#1181). Amounts are returned as strings +// in the contract's smallest unit (stroops) to avoid precision loss. +// +// total_balance = sum of gross fee inflows - sum of allocations +// per_source_breakdown = gross fee inflow per source (allocations are not +// attributed to a source on-chain) + +import { prisma } from '../../utils/prisma.utils'; +import { + cacheGetJson, + cacheInvalidate, + cacheSetJson, +} from '../../utils/redis.utils'; +import { buildPaginationMetadata } from '../../utils/indexer-event-fields.utils'; +import { PaginationMetadata } from '../../utils/api-response.utils'; + +export const TREASURY_CACHE_TTL_SECONDS = 60; +/** Every treasury cache key lives under this prefix so one pattern clears them. */ +export const TREASURY_CACHE_PATTERN = 'treasury:*'; +const SUMMARY_CACHE_KEY = 'treasury:summary'; + +export interface TreasurySummary { + total_balance: string; + per_source_breakdown: Array<{ source: string; balance: string }>; + last_updated: string | null; +} + +export interface TreasuryAllocationView { + destination: string; + amount: string; + purpose: string; + tx_hash: string; + ledger: number; + occurred_at: string; +} + +export interface TreasuryInflowView { + source: string; + amount: string; + tx_hash: string; + ledger: number; + occurred_at: string; +} + +export interface TreasuryPage { + data: T[]; + meta: PaginationMetadata; +} + +export interface TreasuryPageParams { + page: number; + limit: number; +} + +/** Clears every cached treasury read. Called after any treasury event. */ +export async function invalidateTreasuryCache(): Promise { + await cacheInvalidate(TREASURY_CACHE_PATTERN); +} + +export async function getTreasurySummary(): Promise { + const cached = await cacheGetJson(SUMMARY_CACHE_KEY); + if (cached) return cached; + + const [balances, allocated] = await Promise.all([ + prisma.treasuryBalance.findMany({ orderBy: { source: 'asc' } }), + prisma.treasuryAllocation.aggregate({ _sum: { amount: true } }), + ]); + + const collected = balances.reduce( + (sum, row) => sum + BigInt(row.balance.toFixed(0)), + 0n + ); + const spent = BigInt(allocated._sum.amount?.toFixed(0) ?? '0'); + const lastUpdated = balances.reduce( + (latest, row) => (!latest || row.updatedAt > latest ? row.updatedAt : latest), + null + ); + + const summary: TreasurySummary = { + total_balance: (collected - spent).toString(), + per_source_breakdown: balances.map(row => ({ + source: row.source, + balance: row.balance.toFixed(0), + })), + last_updated: lastUpdated?.toISOString() ?? null, + }; + await cacheSetJson(SUMMARY_CACHE_KEY, summary, TREASURY_CACHE_TTL_SECONDS); + return summary; +} + +export async function listTreasuryAllocations( + params: TreasuryPageParams +): Promise> { + const cacheKey = `treasury:allocations:${params.page}:${params.limit}`; + const cached = await cacheGetJson>(cacheKey); + if (cached) return cached; + + const [rows, totalCount] = await prisma.$transaction([ + prisma.treasuryAllocation.findMany({ + orderBy: [{ occurredAt: 'desc' }, { id: 'desc' }], + skip: (params.page - 1) * params.limit, + take: params.limit, + }), + prisma.treasuryAllocation.count(), + ]); + + const page: TreasuryPage = { + data: rows.map(row => ({ + destination: row.destination, + amount: row.amount.toFixed(0), + purpose: row.purpose, + tx_hash: row.txHash, + ledger: row.ledger, + occurred_at: row.occurredAt.toISOString(), + })), + meta: buildPaginationMetadata({ ...params, totalCount }), + }; + await cacheSetJson(cacheKey, page, TREASURY_CACHE_TTL_SECONDS); + return page; +} + +export async function listTreasuryInflows( + params: TreasuryPageParams +): Promise> { + const cacheKey = `treasury:inflows:${params.page}:${params.limit}`; + const cached = await cacheGetJson>(cacheKey); + if (cached) return cached; + + const [rows, totalCount] = await prisma.$transaction([ + prisma.treasuryInflow.findMany({ + orderBy: [{ occurredAt: 'desc' }, { id: 'desc' }], + skip: (params.page - 1) * params.limit, + take: params.limit, + }), + prisma.treasuryInflow.count(), + ]); + + const page: TreasuryPage = { + data: rows.map(row => ({ + source: row.source, + amount: row.amount.toFixed(0), + tx_hash: row.txHash, + ledger: row.ledger, + occurred_at: row.occurredAt.toISOString(), + })), + meta: buildPaginationMetadata({ ...params, totalCount }), + }; + await cacheSetJson(cacheKey, page, TREASURY_CACHE_TTL_SECONDS); + return page; +} diff --git a/src/utils/indexer-event-fields.utils.ts b/src/utils/indexer-event-fields.utils.ts new file mode 100644 index 00000000..ea3f2879 --- /dev/null +++ b/src/utils/indexer-event-fields.utils.ts @@ -0,0 +1,69 @@ +// src/utils/indexer-event-fields.utils.ts +// Shared field readers for indexer handlers that write append-only event logs +// keyed by chain position (txHash + eventIndex). + +import { z } from 'zod'; +import { IndexerChainEvent } from './indexer-event-processor.utils'; +import { PaginationMetadata } from './api-response.utils'; + +/** Chain position that makes an event log row unique. */ +export const chainPositionSchema = z.object({ + txHash: z.string().min(1), + eventIndex: z.number().int().nonnegative(), + ledger: z.number().int().nonnegative(), +}); + +/** Non-negative integer amount in the contract's smallest unit. */ +export const unitAmountSchema = z + .union([ + z.string().regex(/^(0|[1-9]\d*)$/), + z.number().int().nonnegative().safe(), + z.bigint().nonnegative(), + ]) + .transform(value => BigInt(value)); + +/** Non-empty string identifier (wallet, invoice id, source, ...). */ +export const eventIdentifierSchema = z.string().trim().min(1).max(256); + +/** + * Event time from the chain. Numeric values are unix seconds; strings may be + * ISO-8601. Throws on an unparseable value rather than guessing a time. + */ +export function eventOccurredAt(event: IndexerChainEvent): Date { + const raw = event.timestamp ?? event.ledgerClosedAt; + if (raw === undefined || raw === null) { + return new Date(); + } + 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)); + if (Number.isNaN(date.getTime())) { + throw new Error('Invalid chain timestamp'); + } + return date; +} + +/** Builds the standard pagination envelope used by paginated responses. */ +export function buildPaginationMetadata({ + page, + limit, + totalCount, +}: { + page: number; + limit: number; + totalCount: number; +}): PaginationMetadata { + const totalPages = Math.ceil(totalCount / limit); + return { + page, + limit, + totalCount, + totalPages, + hasNextPage: page < totalPages, + hasPrevPage: page > 1, + }; +}