From 17a6ce3a40d82536283b45a4a515f58007d608fb Mon Sep 17 00:00:00 2001 From: Jenn Date: Tue, 20 Oct 2026 20:57:37 +0100 Subject: [PATCH] feat: staking epoch indexer and rewards accumulation API (#1086) Index StakeCreated, EpochRewardDistributed, and RewardsClaimed contract events into stakes and epoch_rewards tables. Expose three REST endpoints: GET /staking/:wallet/positions active stakes with live rewards GET /staking/:wallet/positions/:id single stake + epoch breakdown GET /keys/:keyId/epochs epoch reward history for a key Accrued rewards are fetched live per request via simulateTransaction on the staking contract's get_stake_rewards view function; failures degrade to '0' so a temporarily unavailable contract doesn't break list responses. - prisma/schema/staking-epoch.prisma: Stake and EpochReward models; both idempotent on @@unique([txHash, eventIndex]) - migration 20261020100000_add_staking_epochs: stakes and epoch_rewards DDL with composite indexes for active-position and epoch-history queries - staking-epoch.contract.ts: BigInt-safe ScVal decoder, mirrors lp-rewards contract pattern - staking-epoch.service.ts: recordStake/removeStake/recordEpochReward writes; getActivePositions/getPosition/getKeyEpochs reads with parallel live rewards - staking-epoch-indexer.service.ts: validates required fields + numeric/date sanity per event; REWARDS_CLAIMED closes position only when closePosition is true; checkpoints highest ledger after each batch - staking-epoch.routes.ts: wallet-scoped positions routes behind requireWalletParamMatch; epochs route behind requireJwtAuth - config.schema.ts: adds optional STAKING_CONTRACT_ID env var - indexer-pipeline.service.ts: routes STAKING_EPOCH_EVENT_TYPES via lazy import before cursor advance --- .env.example | 4 + .../migration.sql | 52 +++ prisma/schema/staking-epoch.prisma | 66 +++ src/config.schema.ts | 5 + src/modules/index.ts | 3 + .../indexer/indexer-pipeline.service.ts | 9 + .../indexer/staking-epoch-indexer.service.ts | 377 ++++++++++++++++++ .../staking/staking-epoch.constants.ts | 35 ++ src/modules/staking/staking-epoch.contract.ts | 225 +++++++++++ .../staking/staking-epoch.controller.ts | 191 +++++++++ src/modules/staking/staking-epoch.routes.ts | 52 +++ src/modules/staking/staking-epoch.schemas.ts | 89 +++++ src/modules/staking/staking-epoch.service.ts | 321 +++++++++++++++ 13 files changed, 1429 insertions(+) create mode 100644 prisma/schema/migrations/20261020100000_add_staking_epochs/migration.sql create mode 100644 prisma/schema/staking-epoch.prisma create mode 100644 src/modules/indexer/staking-epoch-indexer.service.ts create mode 100644 src/modules/staking/staking-epoch.constants.ts create mode 100644 src/modules/staking/staking-epoch.contract.ts create mode 100644 src/modules/staking/staking-epoch.controller.ts create mode 100644 src/modules/staking/staking-epoch.routes.ts create mode 100644 src/modules/staking/staking-epoch.schemas.ts create mode 100644 src/modules/staking/staking-epoch.service.ts diff --git a/.env.example b/.env.example index b733f5ef..51fb689a 100644 --- a/.env.example +++ b/.env.example @@ -96,6 +96,10 @@ NONCE_CACHE_TTL_SECONDS=30 # LP position registry contract LP_CONTRACT_ID= +# Staking contract (#1086): deployed staking contract for live get_stake_rewards +# simulation calls. When unset, accrued rewards degrade gracefully to '0'. +STAKING_CONTRACT_ID= + # Webhook settings WEBHOOK_MAX_PER_CREATOR=5 WEBHOOK_RETRY_MAX_ATTEMPTS=3 diff --git a/prisma/schema/migrations/20261020100000_add_staking_epochs/migration.sql b/prisma/schema/migrations/20261020100000_add_staking_epochs/migration.sql new file mode 100644 index 00000000..a3396162 --- /dev/null +++ b/prisma/schema/migrations/20261020100000_add_staking_epochs/migration.sql @@ -0,0 +1,52 @@ +-- Staking epoch indexer: stakes and epoch_rewards tables (#1086) + +CREATE TABLE "stakes" ( + "id" TEXT NOT NULL, + "wallet" TEXT NOT NULL, + "keyId" TEXT NOT NULL, + "amount" DECIMAL(20,7) NOT NULL, + "startEpoch" INTEGER NOT NULL, + "epochsTotal" INTEGER, + "removedAt" TIMESTAMP(3), + "ledger" INTEGER NOT NULL, + "txHash" TEXT NOT NULL, + "eventIndex" INTEGER NOT NULL, + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updatedAt" TIMESTAMP(3) NOT NULL, + + CONSTRAINT "stakes_pkey" PRIMARY KEY ("id") +); + +CREATE UNIQUE INDEX "stakes_txHash_eventIndex_key" + ON "stakes"("txHash", "eventIndex"); + +CREATE INDEX "stakes_wallet_removedAt_idx" + ON "stakes"("wallet", "removedAt"); + +CREATE INDEX "stakes_keyId_idx" + ON "stakes"("keyId"); + +-- ────────────────────────────────────────────────────────────────────────────── + +CREATE TABLE "epoch_rewards" ( + "id" TEXT NOT NULL, + "epoch" INTEGER NOT NULL, + "keyId" TEXT NOT NULL, + "totalDistributed" DECIMAL(20,7) NOT NULL, + "ledger" INTEGER NOT NULL, + "txHash" TEXT NOT NULL, + "eventIndex" INTEGER NOT NULL, + "distributedAt" TIMESTAMP(3) NOT NULL, + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + + CONSTRAINT "epoch_rewards_pkey" PRIMARY KEY ("id") +); + +CREATE UNIQUE INDEX "epoch_rewards_txHash_eventIndex_key" + ON "epoch_rewards"("txHash", "eventIndex"); + +CREATE INDEX "epoch_rewards_keyId_epoch_idx" + ON "epoch_rewards"("keyId", "epoch"); + +CREATE INDEX "epoch_rewards_epoch_idx" + ON "epoch_rewards"("epoch"); diff --git a/prisma/schema/staking-epoch.prisma b/prisma/schema/staking-epoch.prisma new file mode 100644 index 00000000..94b679bf --- /dev/null +++ b/prisma/schema/staking-epoch.prisma @@ -0,0 +1,66 @@ +// prisma/schema/staking-epoch.prisma +// Staking epoch indexer models (#1086): +// - Stake — indexed from StakeCreated / RewardsClaimed contract events +// - EpochReward — indexed from EpochRewardDistributed contract events + +/// A single staking position indexed from on-chain StakeCreated events. +/// Unstaking is recorded by setting `removedAt` when a RewardsClaimed event +/// signals the position is closed. +/// +/// Idempotent via the (txHash, eventIndex) unique constraint so replays are safe. +model Stake { + id String @id @default(cuid()) + /// Wallet that owns the stake. + wallet String + /// Creator key the staked tokens belong to. + keyId String + /// Amount of tokens staked (Decimal(20,7) = stroop-level precision). + amount Decimal @db.Decimal(20, 7) + /// Epoch in which this stake was created. + startEpoch Int + /// Total epochs this position is locked for; null means no fixed duration. + epochsTotal Int? + /// Set when the stake is removed (RewardsClaimed with closePosition=true). + removedAt DateTime? + + // Chain correlation for the creating event + ledger Int + txHash String + eventIndex Int + + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + + @@unique([txHash, eventIndex]) + @@index([wallet, removedAt]) + @@index([keyId]) + @@map("stakes") +} + +/// Per-epoch reward distribution record indexed from EpochRewardDistributed events. +/// One row per on-chain distribution event; the (txHash, eventIndex) pair +/// makes replays idempotent. +model EpochReward { + id String @id @default(cuid()) + /// The epoch sequence number this distribution covers. + epoch Int + /// Creator key the rewards are distributed for. + keyId String + /// Total tokens distributed in this epoch for this key. + totalDistributed Decimal @db.Decimal(20, 7) + + // Chain correlation + ledger Int + txHash String + eventIndex Int + + /// On-chain event timestamp. + distributedAt DateTime + + createdAt DateTime @default(now()) + + @@unique([txHash, eventIndex]) + @@index([keyId, epoch]) + @@index([epoch]) + @@map("epoch_rewards") +} diff --git a/src/config.schema.ts b/src/config.schema.ts index 80b316b8..026f5c49 100644 --- a/src/config.schema.ts +++ b/src/config.schema.ts @@ -254,6 +254,11 @@ export const envSchema = z // LP position registry contract LP_CONTRACT_ID: optionalNonEmptyString, + // Staking contract (#1086): deployed staking contract used for live + // get_stake_rewards simulation calls. When unset, accrued rewards + // degrade gracefully to '0' rather than failing. + STAKING_CONTRACT_ID: optionalNonEmptyString, + // Invoice protocol adapters. Unconfigured reads/execution fail closed. ISSUER_REPUTATION_CONTRACT_ID: optionalNonEmptyString, INVOICE_TRANSFER_CONTRACT_ID: optionalNonEmptyString, diff --git a/src/modules/index.ts b/src/modules/index.ts index 23825c10..bb7bc497 100644 --- a/src/modules/index.ts +++ b/src/modules/index.ts @@ -44,6 +44,7 @@ import referralRouter from './referrals/referrals.routes'; import referralEarningsRouter from './referrals/referral-earnings.routes'; import contractsRouter from './contracts/contract.routes'; import stakingRouter from './staking/staking.routes'; +import { stakingEpochRouter, keyEpochRouter } from './staking/staking-epoch.routes'; import sellersRouter from './sellers/sellers.routes'; import governanceRouter from './governance/governance.routes'; import factoryRouter from './factory/factory.routes'; @@ -129,6 +130,8 @@ router.use( watchlistRouter ); router.use('/staking', routeBodySizeLimit('default'), stakingRouter); +router.use('/staking', routeBodySizeLimit('default'), stakingEpochRouter); +router.use('/keys', routeBodySizeLimit('default'), keyEpochRouter); router.use('/sellers', routeBodySizeLimit('default'), sellersRouter); router.use('/governance', routeBodySizeLimit('default'), governanceRouter); router.use('/factory', routeBodySizeLimit('default'), factoryRouter); diff --git a/src/modules/indexer/indexer-pipeline.service.ts b/src/modules/indexer/indexer-pipeline.service.ts index 0d883990..240073ab 100644 --- a/src/modules/indexer/indexer-pipeline.service.ts +++ b/src/modules/indexer/indexer-pipeline.service.ts @@ -489,6 +489,15 @@ export async function processIndexerEvents( await processLpEvents(lpEvents); } + const { STAKING_EPOCH_EVENT_TYPES, processStakingEpochEvents } = + await import('./staking-epoch-indexer.service'); + const stakingEpochEvents = events.filter(event => + STAKING_EPOCH_EVENT_TYPES.has(event.eventType) + ); + if (stakingEpochEvents.length > 0) { + await processStakingEpochEvents(stakingEpochEvents); + } + if (options.advanceCursor !== false) await advanceIndexerCursorForBatch(events); } diff --git a/src/modules/indexer/staking-epoch-indexer.service.ts b/src/modules/indexer/staking-epoch-indexer.service.ts new file mode 100644 index 00000000..e99f9e07 --- /dev/null +++ b/src/modules/indexer/staking-epoch-indexer.service.ts @@ -0,0 +1,377 @@ +// src/modules/indexer/staking-epoch-indexer.service.ts +// Indexes StakeCreated, EpochRewardDistributed, and RewardsClaimed contract +// events into the stakes and epoch_rewards read models (#1086). +// +// Handler contract: +// - Each batch is deduped by (txHash, eventIndex) before processing. +// - Malformed / incomplete events are skipped with a warn log; they do not +// abort the batch so healthy events in the same delivery still land. +// - Every write is idempotent: duplicate delivery of the same event is a +// safe no-op at the DB layer (upsert with empty update: {}). +// - The highest processed ledger is checkpointed after each batch so the +// gap detector can flag missed ranges. + +import { createHash } from 'crypto'; +import { logger } from '../../utils/logger.utils'; +import { + IndexerChainEvent, + processIndexerChainEvents, +} from '../../utils/indexer-event-processor.utils'; +import { dedupeChainEvents } from '../../utils/indexer-dedupe.utils'; +import { updateIndexedLedger } from './ledger-gap-detection.service'; +import { + recordStake, + recordEpochReward, + removeStake, +} from '../staking/staking-epoch.service'; + +// ── Event type constants (exported for pipeline registration) ─ + +export const STAKING_EPOCH_EVENT_TYPES = new Set([ + 'STAKE_CREATED', + 'StakeCreated', + 'EPOCH_REWARD_DISTRIBUTED', + 'EpochRewardDistributed', + 'REWARDS_CLAIMED', + 'RewardsClaimed', +]); + +// ── Typed event interfaces ──────────────────────────────────── + +/** + * Emitted by the staking contract when a wallet stakes tokens. + * + * Required fields: + * wallet — staking wallet address + * keyId — creator key the tokens are staked against + * amount — staked token amount (decimal string) + * startEpoch — epoch sequence in which the stake begins + * ledger — ledger sequence of the event + * txHash — transaction hash (dedup key together with eventIndex) + * eventIndex — position within the transaction + * + * Optional: + * epochsTotal — lock duration in epochs; omit for flexible (unlocked) stakes + */ +export interface StakeCreatedEvent extends IndexerChainEvent { + eventType: 'STAKE_CREATED' | 'StakeCreated'; + wallet: string; + keyId: string; + amount: string; + startEpoch: number; + epochsTotal?: number; +} + +/** + * Emitted once per epoch per key when the staking contract distributes rewards. + * + * Required fields: + * epoch — the epoch sequence that was distributed + * keyId — creator key the rewards belong to + * totalDistributed — total reward tokens distributed (decimal string) + * distributedAt — ISO-8601 timestamp of the distribution + * ledger / txHash / eventIndex — chain correlation + */ +export interface EpochRewardDistributedEvent extends IndexerChainEvent { + eventType: 'EPOCH_REWARD_DISTRIBUTED' | 'EpochRewardDistributed'; + epoch: number; + keyId: string; + totalDistributed: string; + distributedAt: string; +} + +/** + * Emitted when a wallet claims accumulated rewards and optionally closes the + * staking position. + * + * Required fields: + * stakeId — DB id of the Stake row to close (set by the chain indexer + * pipeline, which maps the on-chain position handle to our + * internal id before dispatching the event) + * wallet — claiming wallet + * closePosition — true signals the position should be marked removed + * claimedAt — ISO-8601 timestamp of the claim + * ledger / txHash / eventIndex — chain correlation + */ +export interface RewardsClaimedEvent extends IndexerChainEvent { + eventType: 'REWARDS_CLAIMED' | 'RewardsClaimed'; + stakeId: string; + wallet: string; + closePosition: boolean; + claimedAt: string; +} + +// ── Required-field validators ───────────────────────────────── + +const STAKE_CREATED_REQUIRED: (keyof StakeCreatedEvent)[] = [ + 'wallet', + 'keyId', + 'amount', + 'startEpoch', + 'ledger', + 'txHash', + 'eventIndex', +]; + +const EPOCH_REWARD_REQUIRED: (keyof EpochRewardDistributedEvent)[] = [ + 'epoch', + 'keyId', + 'totalDistributed', + 'distributedAt', + 'ledger', + 'txHash', + 'eventIndex', +]; + +const REWARDS_CLAIMED_REQUIRED: (keyof RewardsClaimedEvent)[] = [ + 'stakeId', + 'wallet', + 'claimedAt', + 'ledger', + 'txHash', + 'eventIndex', +]; + +function missingField( + event: IndexerChainEvent, + fields: string[] +): string | null { + for (const field of fields) { + const value = (event as Record)[field]; + if (value === undefined || value === null || value === '') return field; + } + return null; +} + +// ── Batch processor ─────────────────────────────────────────── + +/** + * Processes a mixed batch of staking epoch events, routing each to the + * appropriate handler. Malformed events are skipped; the ledger cursor is + * checkpointed after all events are processed. + */ +export async function processStakingEpochEvents( + events: IndexerChainEvent[] +): Promise { + await processIndexerChainEvents(events, async event => { + const type = event.eventType; + + if (type === 'STAKE_CREATED' || type === 'StakeCreated') { + await handleStakeCreated(event as StakeCreatedEvent); + return; + } + + if ( + type === 'EPOCH_REWARD_DISTRIBUTED' || + type === 'EpochRewardDistributed' + ) { + await handleEpochRewardDistributed( + event as EpochRewardDistributedEvent + ); + return; + } + + if (type === 'REWARDS_CLAIMED' || type === 'RewardsClaimed') { + await handleRewardsClaimed(event as RewardsClaimedEvent); + return; + } + }); + + await checkpointHighestLedger(events); +} + +// ── Individual handlers ─────────────────────────────────────── + +async function handleStakeCreated(e: StakeCreatedEvent): Promise { + const missing = missingField(e, STAKE_CREATED_REQUIRED); + if (missing) { + logger.warn( + { eventId: `${e.txHash}:${e.eventIndex}`, missingField: missing }, + 'Skipping STAKE_CREATED event due to missing required field' + ); + return; + } + + const amount = Number(e.amount); + if (!Number.isFinite(amount) || amount <= 0) { + logger.warn( + { eventId: `${e.txHash}:${e.eventIndex}`, amount: e.amount }, + 'Skipping STAKE_CREATED event with invalid amount' + ); + return; + } + + const startEpoch = Number(e.startEpoch); + if (!Number.isInteger(startEpoch) || startEpoch < 0) { + logger.warn( + { eventId: `${e.txHash}:${e.eventIndex}`, startEpoch: e.startEpoch }, + 'Skipping STAKE_CREATED event with invalid startEpoch' + ); + return; + } + + await recordStake({ + wallet: e.wallet, + keyId: e.keyId, + amount: e.amount, + startEpoch, + epochsTotal: e.epochsTotal != null ? Number(e.epochsTotal) : undefined, + ledger: Number(e.ledger), + txHash: String(e.txHash), + eventIndex: Number(e.eventIndex), + }); + + logger.info( + { + wallet: e.wallet, + keyId: e.keyId, + amount: e.amount, + startEpoch, + ledger: Number(e.ledger), + txHash: String(e.txHash), + }, + 'STAKE_CREATED event processed' + ); +} + +async function handleEpochRewardDistributed( + e: EpochRewardDistributedEvent +): Promise { + const missing = missingField(e, EPOCH_REWARD_REQUIRED); + if (missing) { + logger.warn( + { eventId: `${e.txHash}:${e.eventIndex}`, missingField: missing }, + 'Skipping EPOCH_REWARD_DISTRIBUTED event due to missing required field' + ); + return; + } + + const epoch = Number(e.epoch); + if (!Number.isInteger(epoch) || epoch < 0) { + logger.warn( + { eventId: `${e.txHash}:${e.eventIndex}`, epoch: e.epoch }, + 'Skipping EPOCH_REWARD_DISTRIBUTED event with invalid epoch' + ); + return; + } + + const totalDistributed = Number(e.totalDistributed); + if (!Number.isFinite(totalDistributed) || totalDistributed < 0) { + logger.warn( + { + eventId: `${e.txHash}:${e.eventIndex}`, + totalDistributed: e.totalDistributed, + }, + 'Skipping EPOCH_REWARD_DISTRIBUTED event with invalid totalDistributed' + ); + return; + } + + const distributedAt = new Date(e.distributedAt); + if (Number.isNaN(distributedAt.getTime())) { + logger.warn( + { + eventId: `${e.txHash}:${e.eventIndex}`, + distributedAt: e.distributedAt, + }, + 'Skipping EPOCH_REWARD_DISTRIBUTED event with invalid distributedAt' + ); + return; + } + + await recordEpochReward({ + epoch, + keyId: e.keyId, + totalDistributed: e.totalDistributed, + ledger: Number(e.ledger), + txHash: String(e.txHash), + eventIndex: Number(e.eventIndex), + distributedAt, + }); + + logger.info( + { + epoch, + keyId: e.keyId, + totalDistributed: e.totalDistributed, + ledger: Number(e.ledger), + txHash: String(e.txHash), + }, + 'EPOCH_REWARD_DISTRIBUTED event processed' + ); +} + +async function handleRewardsClaimed(e: RewardsClaimedEvent): Promise { + const missing = missingField(e, REWARDS_CLAIMED_REQUIRED); + if (missing) { + logger.warn( + { eventId: `${e.txHash}:${e.eventIndex}`, missingField: missing }, + 'Skipping REWARDS_CLAIMED event due to missing required field' + ); + return; + } + + const claimedAt = new Date(e.claimedAt); + if (Number.isNaN(claimedAt.getTime())) { + logger.warn( + { eventId: `${e.txHash}:${e.eventIndex}`, claimedAt: e.claimedAt }, + 'Skipping REWARDS_CLAIMED event with invalid claimedAt' + ); + return; + } + + // Only close the position when the event explicitly signals it; a plain + // reward claim without closePosition=true leaves the stake active. + if (e.closePosition === true) { + await removeStake(e.stakeId, { + txHash: String(e.txHash), + eventIndex: Number(e.eventIndex), + removedAt: claimedAt, + }); + + logger.info( + { + stakeId: e.stakeId, + wallet: e.wallet, + ledger: Number(e.ledger), + txHash: String(e.txHash), + }, + 'REWARDS_CLAIMED (closePosition=true): stake removed' + ); + } else { + logger.info( + { + stakeId: e.stakeId, + wallet: e.wallet, + ledger: Number(e.ledger), + txHash: String(e.txHash), + }, + 'REWARDS_CLAIMED event processed (position kept open)' + ); + } +} + +// ── Ledger checkpoint ───────────────────────────────────────── + +async function checkpointHighestLedger( + events: IndexerChainEvent[] +): Promise { + const unique = dedupeChainEvents(events); + const ledgers = unique + .map(e => e.ledger) + .filter((l): l is number => typeof l === 'number'); + + if (ledgers.length === 0) return; + + const maxLedger = Math.max(...ledgers); + const identifiers = unique + .map(e => `${e.txHash}:${e.eventIndex}`) + .sort() + .join('|'); + const batchHash = createHash('sha256') + .update(identifiers, 'utf8') + .digest('hex') + .slice(0, 16); + + await updateIndexedLedger(maxLedger, `${maxLedger}-000`, batchHash); +} diff --git a/src/modules/staking/staking-epoch.constants.ts b/src/modules/staking/staking-epoch.constants.ts new file mode 100644 index 00000000..597ce380 --- /dev/null +++ b/src/modules/staking/staking-epoch.constants.ts @@ -0,0 +1,35 @@ +// src/modules/staking/staking-epoch.constants.ts +// Cache key builders and TTLs for the staking epoch endpoints (#1086). + +/** Redis TTL for the active-positions list (30 seconds — kept short because + * accrued rewards are fetched live per request, not cached here). */ +export const STAKING_POSITIONS_CACHE_TTL_SECONDS = 30; + +/** Redis TTL for the key epoch history list (60 seconds). */ +export const STAKING_KEY_EPOCHS_CACHE_TTL_SECONDS = 60; + +const PREFIX = 'staking:epoch'; + +export function buildPositionsCacheKey(wallet: string): string { + return `${PREFIX}:positions:${wallet.toLowerCase()}`; +} + +export function buildPositionDetailCacheKey( + wallet: string, + stakeId: string +): string { + return `${PREFIX}:position:${wallet.toLowerCase()}:${stakeId}`; +} + +export function buildKeyEpochsCacheKey( + keyId: string, + offset: number, + limit: number +): string { + return `${PREFIX}:key-epochs:${keyId}:${offset}:${limit}`; +} + +/** Glob pattern for all cached artifacts for a wallet. */ +export function stakingEpochWalletCachePattern(wallet: string): string { + return `${PREFIX}:position*:${wallet.toLowerCase()}*`; +} diff --git a/src/modules/staking/staking-epoch.contract.ts b/src/modules/staking/staking-epoch.contract.ts new file mode 100644 index 00000000..d39c5ae7 --- /dev/null +++ b/src/modules/staking/staking-epoch.contract.ts @@ -0,0 +1,225 @@ +// src/modules/staking/staking-epoch.contract.ts +// Live contract reader for get_stake_rewards (#1086). +// +// Builds a Soroban simulateTransaction envelope for the staking contract's +// read-only `get_stake_rewards(wallet, stake_id)` view function and decodes +// the returned i128/u128 ScVal into a stroop-precision decimal string. +// +// Mirrors the pattern established in lp-rewards.contract.ts. + +import { + Account, + Address, + Contract, + Networks, + TransactionBuilder, + xdr, +} from '@stellar/stellar-base'; +import { envConfig } from '../../config'; +import { + defaultSorobanTransport, + SorobanPostJson, +} from '../../clients/soroban.client'; +import { logger } from '../../utils/logger.utils'; + +// ── Errors ──────────────────────────────────────────────────── + +export class StakingContractNotConfiguredError extends Error { + constructor() { + super('STAKING_CONTRACT_ID is not configured'); + this.name = 'StakingContractNotConfiguredError'; + } +} + +export class StakeRewardsFetchError extends Error { + public readonly cause?: unknown; + constructor(message: string, options?: { cause?: unknown }) { + super(message); + this.name = 'StakeRewardsFetchError'; + this.cause = options?.cause; + } +} + +// ── Helpers ─────────────────────────────────────────────────── + +function networkPassphrase(): string { + return envConfig.STELLAR_NETWORK === 'mainnet' + ? Networks.PUBLIC + : Networks.TESTNET; +} + +/** + * Builds the unsigned XDR envelope for `get_stake_rewards(wallet, stakeId)`. + * The `sequence` defaults to '0' since this is a read-only simulation — no + * sequence increment is applied on-chain. + */ +export function buildGetStakeRewardsXdr( + contractId: string, + wallet: string, + stakeId: string, + sequence = '0' +): string { + const operation = new Contract(contractId).call( + 'get_stake_rewards', + new Address(wallet).toScVal(), + xdr.ScVal.scvString(stakeId) + ); + + return new TransactionBuilder(new Account(wallet, sequence), { + fee: '100', + networkPassphrase: networkPassphrase(), + }) + .addOperation(operation) + .setTimeout(300) + .build() + .toXDR(); +} + +/** + * Decodes a base64 XDR ScVal (u32 / i32 / u64 / i64 / u128 / i128) into a + * BigInt-safe decimal string. Precision loss is prevented by never casting + * through Number. + */ +export function decodeStakeRewardsScVal(scValXdr: string): string { + const val = xdr.ScVal.fromXDR(scValXdr, 'base64'); + const t = val.switch(); + + if (t === xdr.ScValType.scvU32()) return BigInt(val.u32()).toString(); + if (t === xdr.ScValType.scvI32()) return BigInt(val.i32()).toString(); + + if (t === xdr.ScValType.scvU64()) + return BigInt(val.u64().toString()).toString(); + if (t === xdr.ScValType.scvI64()) + return BigInt(val.i64().toString()).toString(); + + if (t === xdr.ScValType.scvU128()) { + const p = val.u128() as unknown as { + hi(): { toString(): string }; + lo(): { toString(): string }; + }; + return ( + (BigInt(p.hi().toString()) << 64n) | + BigInt.asUintN(64, BigInt(p.lo().toString())) + ).toString(); + } + + if (t === xdr.ScValType.scvI128()) { + const p = val.i128() as unknown as { + hi(): { toString(): string }; + lo(): { toString(): string }; + }; + return ( + (BigInt(p.hi().toString()) << 64n) | + BigInt.asUintN(64, BigInt(p.lo().toString())) + ).toString(); + } + + throw new StakeRewardsFetchError( + `Unsupported ScValType for stake rewards: ${t.name}` + ); +} + +// ── Public API ──────────────────────────────────────────────── + +export interface GetStakeRewardsDeps { + postJson?: SorobanPostJson; + /** Override contract ID (useful in tests). */ + contractIdOverride?: string; +} + +/** + * Calls `get_stake_rewards(wallet, stakeId)` against the staking contract via + * Soroban RPC simulation and returns the accrued rewards as a decimal string + * in stroops. + * + * Throws {@link StakingContractNotConfiguredError} when the contract ID env + * var is missing, or {@link StakeRewardsFetchError} on any RPC/decode failure. + */ +export async function getStakeRewardsLive( + wallet: string, + stakeId: string, + deps?: GetStakeRewardsDeps +): Promise { + const contractId = + deps?.contractIdOverride ?? envConfig.STAKING_CONTRACT_ID; + if (!contractId) { + throw new StakingContractNotConfiguredError(); + } + + const postJson = deps?.postJson ?? defaultSorobanTransport; + const rpcUrl = envConfig.STELLAR_SOROBAN_RPC_URL; + if (!rpcUrl) { + throw new StakeRewardsFetchError('STELLAR_SOROBAN_RPC_URL is not configured'); + } + + let txXdr: string; + try { + txXdr = buildGetStakeRewardsXdr(contractId, wallet, stakeId); + } catch (error) { + throw new StakeRewardsFetchError( + 'Failed to build transaction for get_stake_rewards', + { cause: error } + ); + } + + let envelope: unknown; + try { + envelope = await postJson(rpcUrl, { + jsonrpc: '2.0', + id: 'get-stake-rewards', + method: 'simulateTransaction', + params: { transaction: txXdr }, + }); + } catch (error) { + throw new StakeRewardsFetchError( + 'RPC request failed for get_stake_rewards', + { cause: error } + ); + } + + const parsed = (envelope ?? {}) as { + error?: { code?: number; message?: string }; + result?: { + error?: string; + results?: Array<{ xdr?: string; retval?: string }>; + }; + }; + + if (parsed.error) { + throw new StakeRewardsFetchError( + `Soroban RPC error: ${parsed.error.message ?? 'unknown'}`, + { cause: parsed.error } + ); + } + + if (parsed.result?.error) { + throw new StakeRewardsFetchError( + `Simulation returned error: ${parsed.result.error}`, + { cause: parsed.result } + ); + } + + const returnedXdr = + parsed.result?.results?.[0]?.retval ?? + parsed.result?.results?.[0]?.xdr; + + if (!returnedXdr) { + throw new StakeRewardsFetchError( + 'Simulation result missing in RPC response' + ); + } + + try { + const rewards = decodeStakeRewardsScVal(returnedXdr); + logger.debug( + { wallet, stakeId, rewards }, + 'get_stake_rewards simulation decoded' + ); + return rewards; + } catch (error) { + throw new StakeRewardsFetchError( + 'Failed to decode get_stake_rewards return value', + { cause: error } + ); + } +} diff --git a/src/modules/staking/staking-epoch.controller.ts b/src/modules/staking/staking-epoch.controller.ts new file mode 100644 index 00000000..5571dd28 --- /dev/null +++ b/src/modules/staking/staking-epoch.controller.ts @@ -0,0 +1,191 @@ +// src/modules/staking/staking-epoch.controller.ts +// HTTP handlers for the staking epoch REST endpoints (#1086). +// +// GET /staking/:wallet/positions +// GET /staking/:wallet/positions/:stakeId +// GET /keys/:keyId/epochs + +import { Response } from 'express'; +import { AuthenticatedRequest } from '../../middlewares/jwt-auth.middleware'; +import { + ErrorCode, + sendError, + sendSuccess, + sendValidationError, + zodIssuesToDetails, +} from '../../utils/api-response.utils'; +import { logger } from '../../utils/logger.utils'; +import { + getActivePositions, + getPosition, + getKeyEpochs, + StakeNotFoundError, +} from './staking-epoch.service'; +import { + PositionsQuerySchema, + PositionDetailQuerySchema, + KeyEpochsQuerySchema, +} from './staking-epoch.schemas'; + +// ── GET /staking/:wallet/positions ──────────────────────────── + +/** + * Returns all active staking positions for the authenticated wallet, each + * hydrated with live accrued rewards from the contract. + * + * `currentEpoch` query param (optional, default 0) is used to compute + * `epochsRemaining` for locked positions. + */ +export async function httpGetStakingPositions( + req: AuthenticatedRequest, + res: Response +): Promise { + const wallet = (req.params.wallet ?? '').trim(); + + const parsed = PositionsQuerySchema.safeParse(req.query); + if (!parsed.success) { + sendValidationError( + res, + 'Invalid query parameters', + zodIssuesToDetails(parsed.error.issues) + ); + return; + } + + try { + const positions = await getActivePositions( + wallet, + parsed.data.currentEpoch + ); + sendSuccess(res, positions); + } catch (error) { + logger.error( + { + type: 'staking_epoch_handler_error', + handler: 'httpGetStakingPositions', + wallet, + ...(req.requestId ? { requestId: req.requestId } : {}), + error, + }, + 'Failed to retrieve staking positions' + ); + sendError( + res, + 500, + ErrorCode.INTERNAL_ERROR, + 'Failed to retrieve staking positions' + ); + } +} + +// ── GET /staking/:wallet/positions/:stakeId ─────────────────── + +/** + * Returns a single active staking position with live accrued rewards and the + * full epoch reward history for its key. + * + * Returns 404 when the stake does not exist or belongs to a different wallet. + */ +export async function httpGetStakingPosition( + req: AuthenticatedRequest, + res: Response +): Promise { + const wallet = (req.params.wallet ?? '').trim(); + const stakeId = (req.params.stakeId ?? '').trim(); + + const parsed = PositionDetailQuerySchema.safeParse(req.query); + if (!parsed.success) { + sendValidationError( + res, + 'Invalid query parameters', + zodIssuesToDetails(parsed.error.issues) + ); + return; + } + + try { + const detail = await getPosition( + stakeId, + wallet, + parsed.data.currentEpoch + ); + sendSuccess(res, detail); + } catch (error) { + if (error instanceof StakeNotFoundError) { + sendError(res, 404, ErrorCode.NOT_FOUND, 'Stake not found'); + return; + } + logger.error( + { + type: 'staking_epoch_handler_error', + handler: 'httpGetStakingPosition', + wallet, + stakeId, + ...(req.requestId ? { requestId: req.requestId } : {}), + error, + }, + 'Failed to retrieve staking position' + ); + sendError( + res, + 500, + ErrorCode.INTERNAL_ERROR, + 'Failed to retrieve staking position' + ); + } +} + +// ── GET /keys/:keyId/epochs ─────────────────────────────────── + +/** + * Returns the epoch reward history for a creator key, ordered oldest-first. + * Offset-paginated via `limit` and `offset` query params. + */ +export async function httpGetKeyEpochs( + req: AuthenticatedRequest, + res: Response +): Promise { + const keyId = (req.params.keyId ?? '').trim(); + + const parsed = KeyEpochsQuerySchema.safeParse(req.query); + if (!parsed.success) { + sendValidationError( + res, + 'Invalid query parameters', + zodIssuesToDetails(parsed.error.issues) + ); + return; + } + + const { limit, offset } = parsed.data; + + try { + const { items, total } = await getKeyEpochs(keyId, limit, offset); + sendSuccess(res, { + items, + meta: { + total, + limit, + offset, + hasMore: offset + items.length < total, + }, + }); + } catch (error) { + logger.error( + { + type: 'staking_epoch_handler_error', + handler: 'httpGetKeyEpochs', + keyId, + ...(req.requestId ? { requestId: req.requestId } : {}), + error, + }, + 'Failed to retrieve key epoch history' + ); + sendError( + res, + 500, + ErrorCode.INTERNAL_ERROR, + 'Failed to retrieve key epoch history' + ); + } +} diff --git a/src/modules/staking/staking-epoch.routes.ts b/src/modules/staking/staking-epoch.routes.ts new file mode 100644 index 00000000..b7e68d01 --- /dev/null +++ b/src/modules/staking/staking-epoch.routes.ts @@ -0,0 +1,52 @@ +// src/modules/staking/staking-epoch.routes.ts +// Staking epoch REST endpoints (#1086). +// +// GET /staking/:wallet/positions — active stakes with live rewards +// GET /staking/:wallet/positions/:stakeId — single stake + epoch breakdown +// GET /keys/:keyId/epochs — epoch reward history for a key + +import { Router } from 'express'; +import { + requireJwtAuth, + requireWalletParamMatch, +} from '../../middlewares/jwt-auth.middleware'; +import { + httpGetStakingPositions, + httpGetStakingPosition, + httpGetKeyEpochs, +} from './staking-epoch.controller'; + +const stakingEpochRouter = Router(); + +// ── Staking position endpoints (wallet-scoped) ──────────────── + +stakingEpochRouter.get( + '/:wallet/positions', + requireWalletParamMatch('wallet'), + httpGetStakingPositions +); +stakingEpochRouter.all('/:wallet/positions', (_req, res) => { + res.set('Allow', 'GET').sendStatus(405); +}); + +stakingEpochRouter.get( + '/:wallet/positions/:stakeId', + requireWalletParamMatch('wallet'), + httpGetStakingPosition +); +stakingEpochRouter.all('/:wallet/positions/:stakeId', (_req, res) => { + res.set('Allow', 'GET').sendStatus(405); +}); + +export { stakingEpochRouter }; + +// ── Key epoch history (public, JWT still required for consistency) ─ + +const keyEpochRouter = Router(); + +keyEpochRouter.get('/:keyId/epochs', requireJwtAuth, httpGetKeyEpochs); +keyEpochRouter.all('/:keyId/epochs', (_req, res) => { + res.set('Allow', 'GET').sendStatus(405); +}); + +export { keyEpochRouter }; diff --git a/src/modules/staking/staking-epoch.schemas.ts b/src/modules/staking/staking-epoch.schemas.ts new file mode 100644 index 00000000..ccdd35b2 --- /dev/null +++ b/src/modules/staking/staking-epoch.schemas.ts @@ -0,0 +1,89 @@ +// src/modules/staking/staking-epoch.schemas.ts +// Zod request/response schemas for the staking epoch REST endpoints (#1086). + +import { z } from 'zod'; +import { + MAX_PAGE_SIZE, + MIN_PAGE_SIZE, +} from '../../constants/pagination.constants'; + +export const DEFAULT_KEY_EPOCHS_PAGE_SIZE = 20; + +// ── Query schemas ───────────────────────────────────────────── + +/** + * Query params for GET /staking/:wallet/positions + * + * `currentEpoch` is the caller-supplied epoch sequence used to compute + * `epochsRemaining`. When omitted it defaults to 0 and `epochsRemaining` + * will reflect the full lock duration remaining from epoch 0. + */ +export const PositionsQuerySchema = z + .object({ + currentEpoch: z.coerce.number().int().min(0).optional().default(0), + }) + .strict(); + +export type PositionsQuery = z.infer; + +/** + * Query params for GET /staking/:wallet/positions/:stakeId + * Same `currentEpoch` semantics as the list endpoint. + */ +export const PositionDetailQuerySchema = z + .object({ + currentEpoch: z.coerce.number().int().min(0).optional().default(0), + }) + .strict(); + +export type PositionDetailQuery = z.infer; + +/** + * Query params for GET /keys/:keyId/epochs + * Offset-based pagination matching the service layer. + */ +export const KeyEpochsQuerySchema = z + .object({ + limit: z.coerce + .number() + .int() + .min(MIN_PAGE_SIZE) + .max(MAX_PAGE_SIZE) + .optional() + .default(DEFAULT_KEY_EPOCHS_PAGE_SIZE), + offset: z.coerce.number().int().min(0).optional().default(0), + }) + .strict(); + +export type KeyEpochsQuery = z.infer; + +// ── Response item shapes ────────────────────────────────────── + +export const StakePositionItemSchema = z.object({ + id: z.string(), + wallet: z.string(), + keyId: z.string(), + amount: z.string(), + startEpoch: z.number().int(), + epochsTotal: z.number().int().nullable(), + epochsRemaining: z.number().int().nullable(), + accruedRewards: z.string(), + createdAt: z.string(), +}); + +export type StakePositionItem = z.infer; + +export const EpochRewardItemSchema = z.object({ + epoch: z.number().int(), + keyId: z.string(), + totalDistributed: z.string(), + distributedAt: z.string(), +}); + +export type EpochRewardItem = z.infer; + +export const StakePositionDetailSchema = StakePositionItemSchema.extend({ + epochHistory: z.array(EpochRewardItemSchema), +}); + +export type StakePositionDetail = z.infer; diff --git a/src/modules/staking/staking-epoch.service.ts b/src/modules/staking/staking-epoch.service.ts new file mode 100644 index 00000000..4a27f2d8 --- /dev/null +++ b/src/modules/staking/staking-epoch.service.ts @@ -0,0 +1,321 @@ +// src/modules/staking/staking-epoch.service.ts +// Persistence and query logic for staking epoch data (#1086). +// +// Three concerns live here: +// 1. recordStake / removeStake — written by the indexer from +// StakeCreated / RewardsClaimed events. +// 2. recordEpochReward — written by the indexer from +// EpochRewardDistributed events. +// 3. getActivePositions / getPosition / getKeyEpochs — read by the HTTP +// handlers; getPosition hydrates live accrued rewards from the contract. + +import { prisma } from '../../utils/prisma.utils'; +import { logger } from '../../utils/logger.utils'; +import { + getStakeRewardsLive, + StakeRewardsFetchError, + StakingContractNotConfiguredError, +} from './staking-epoch.contract'; +import type { GetStakeRewardsDeps } from './staking-epoch.contract'; + +// ── Error classes ───────────────────────────────────────────── + +export class StakeNotFoundError extends Error { + constructor(stakeId: string) { + super(`Stake not found: ${stakeId}`); + this.name = 'StakeNotFoundError'; + } +} + +// ── Indexer write helpers ───────────────────────────────────── + +export interface RecordStakeInput { + wallet: string; + keyId: string; + amount: string; + startEpoch: number; + epochsTotal?: number; + ledger: number; + txHash: string; + eventIndex: number; +} + +/** + * Persists a StakeCreated event as a Stake row. The (txHash, eventIndex) + * unique constraint makes replays a no-op. + */ +export async function recordStake(input: RecordStakeInput): Promise { + await prisma.stake.upsert({ + where: { + txHash_eventIndex: { + txHash: input.txHash, + eventIndex: input.eventIndex, + }, + }, + create: { + wallet: input.wallet, + keyId: input.keyId, + amount: input.amount, + startEpoch: input.startEpoch, + epochsTotal: input.epochsTotal ?? null, + ledger: input.ledger, + txHash: input.txHash, + eventIndex: input.eventIndex, + }, + update: {}, + }); +} + +export interface RemoveStakeInput { + txHash: string; + eventIndex: number; + removedAt: Date; +} + +/** + * Marks a stake as removed when a RewardsClaimed event signals the position + * is closed. Matched by (txHash, eventIndex) of the *creating* event — the + * caller is responsible for passing the stakeId from the claim event payload. + * Falls back to a no-op when the stake row is not found (out-of-order replay). + */ +export async function removeStake( + stakeId: string, + input: RemoveStakeInput +): Promise { + const updated = await prisma.stake.updateMany({ + where: { id: stakeId, removedAt: null }, + data: { removedAt: input.removedAt }, + }); + + if (updated.count === 0) { + logger.warn( + { stakeId, txHash: input.txHash, eventIndex: input.eventIndex }, + 'removeStake: no active stake found to remove (already removed or unknown id)' + ); + } +} + +export interface RecordEpochRewardInput { + epoch: number; + keyId: string; + totalDistributed: string; + ledger: number; + txHash: string; + eventIndex: number; + distributedAt: Date; +} + +/** + * Persists an EpochRewardDistributed event. Idempotent via (txHash, eventIndex). + */ +export async function recordEpochReward( + input: RecordEpochRewardInput +): Promise { + await prisma.epochReward.upsert({ + where: { + txHash_eventIndex: { + txHash: input.txHash, + eventIndex: input.eventIndex, + }, + }, + create: { + epoch: input.epoch, + keyId: input.keyId, + totalDistributed: input.totalDistributed, + ledger: input.ledger, + txHash: input.txHash, + eventIndex: input.eventIndex, + distributedAt: input.distributedAt, + }, + update: {}, + }); +} + +// ── Response shapes ─────────────────────────────────────────── + +export interface StakePositionItem { + id: string; + wallet: string; + keyId: string; + amount: string; + startEpoch: number; + epochsTotal: number | null; + /** Epochs remaining until unlock; null when no fixed duration. */ + epochsRemaining: number | null; + /** Live accrued rewards from contract, in stroops (decimal string). */ + accruedRewards: string; + createdAt: string; +} + +export interface EpochRewardItem { + epoch: number; + keyId: string; + totalDistributed: string; + distributedAt: string; +} + +// ── Accrued rewards helper ──────────────────────────────────── + +/** + * Fetches live accrued rewards for a stake from the contract. + * Returns '0' on any contract error so a single unavailable position doesn't + * break an entire list response. + */ +async function fetchAccruedRewards( + wallet: string, + stakeId: string, + deps?: GetStakeRewardsDeps +): Promise { + try { + return await getStakeRewardsLive(wallet, stakeId, deps); + } catch (error) { + if ( + error instanceof StakingContractNotConfiguredError || + error instanceof StakeRewardsFetchError + ) { + logger.warn( + { wallet, stakeId, error: (error as Error).message }, + 'get_stake_rewards unavailable; returning 0 for accrued rewards' + ); + return '0'; + } + throw error; + } +} + +// ── Read queries ────────────────────────────────────────────── + +/** + * Returns all active (not removed) stakes for a wallet, with live accrued + * rewards fetched in parallel from the contract. + * + * `currentEpoch` is supplied by the caller (from the request context or a + * contract query); it is used only to compute `epochsRemaining` and defaults + * to 0 when omitted. + */ +export async function getActivePositions( + wallet: string, + currentEpoch = 0, + deps?: GetStakeRewardsDeps +): Promise { + const rows = await prisma.stake.findMany({ + where: { wallet, removedAt: null }, + orderBy: { createdAt: 'asc' }, + }); + + if (rows.length === 0) return []; + + // Fetch live rewards for all positions in parallel. + const rewardsResults = await Promise.allSettled( + rows.map(r => fetchAccruedRewards(wallet, r.id, deps)) + ); + + return rows.map((row, i) => { + const rewardResult = rewardsResults[i]; + const accruedRewards = + rewardResult.status === 'fulfilled' ? rewardResult.value : '0'; + + const epochsRemaining = + row.epochsTotal != null + ? Math.max(0, row.startEpoch + row.epochsTotal - currentEpoch) + : null; + + return { + id: row.id, + wallet: row.wallet, + keyId: row.keyId, + amount: row.amount.toString(), + startEpoch: row.startEpoch, + epochsTotal: row.epochsTotal, + epochsRemaining, + accruedRewards, + createdAt: row.createdAt.toISOString(), + }; + }); +} + +export interface StakePositionDetail extends StakePositionItem { + /** Per-epoch reward history for the key this stake belongs to. */ + epochHistory: EpochRewardItem[]; +} + +/** + * Returns a single active stake with live accrued rewards and the epoch + * reward history for its key. + * + * Throws {@link StakeNotFoundError} when the stake doesn't exist or belongs + * to a different wallet. + */ +export async function getPosition( + stakeId: string, + wallet: string, + currentEpoch = 0, + deps?: GetStakeRewardsDeps +): Promise { + const row = await prisma.stake.findFirst({ + where: { id: stakeId, wallet, removedAt: null }, + }); + + if (!row) throw new StakeNotFoundError(stakeId); + + const [accruedRewards, epochHistory] = await Promise.all([ + fetchAccruedRewards(wallet, stakeId, deps), + prisma.epochReward.findMany({ + where: { keyId: row.keyId }, + orderBy: { epoch: 'asc' }, + }), + ]); + + const epochsRemaining = + row.epochsTotal != null + ? Math.max(0, row.startEpoch + row.epochsTotal - currentEpoch) + : null; + + return { + id: row.id, + wallet: row.wallet, + keyId: row.keyId, + amount: row.amount.toString(), + startEpoch: row.startEpoch, + epochsTotal: row.epochsTotal, + epochsRemaining, + accruedRewards, + createdAt: row.createdAt.toISOString(), + epochHistory: epochHistory.map(e => ({ + epoch: e.epoch, + keyId: e.keyId, + totalDistributed: e.totalDistributed.toString(), + distributedAt: e.distributedAt.toISOString(), + })), + }; +} + +/** + * Returns the epoch reward history for a key, ordered oldest-first. + * Simple offset pagination via `limit` / `offset`. + */ +export async function getKeyEpochs( + keyId: string, + limit: number, + offset: number +): Promise<{ items: EpochRewardItem[]; total: number }> { + const [rows, total] = await Promise.all([ + prisma.epochReward.findMany({ + where: { keyId }, + orderBy: { epoch: 'asc' }, + take: limit, + skip: offset, + }), + prisma.epochReward.count({ where: { keyId } }), + ]); + + return { + items: rows.map(r => ({ + epoch: r.epoch, + keyId: r.keyId, + totalDistributed: r.totalDistributed.toString(), + distributedAt: r.distributedAt.toISOString(), + })), + total, + }; +}