diff --git a/.github/workflows/backend-ci.yml b/.github/workflows/backend-ci.yml deleted file mode 100644 index 9d261f38..00000000 --- a/.github/workflows/backend-ci.yml +++ /dev/null @@ -1,55 +0,0 @@ -name: Backend CI - -on: - push: - paths: - - 'backend/**' - pull_request: - paths: - - 'backend/**' - -jobs: - test: - name: TypeScript Build + Tests - runs-on: ubuntu-latest - services: - postgres: - image: postgres:15-alpine - env: - POSTGRES_USER: boxmeout - POSTGRES_PASSWORD: boxmeout - POSTGRES_DB: boxmeout - ports: - - 5432:5432 - redis: - image: redis:7-alpine - ports: - - 6379:6379 - steps: - - uses: actions/checkout@v4 - - - name: Setup Node - uses: actions/setup-node@v4 - with: - node-version: '20' - cache: 'npm' - cache-dependency-path: backend/package-lock.json - - - name: Install dependencies - working-directory: backend - run: npm ci - - - name: Lint - working-directory: backend - run: npm run lint - - - name: Build - working-directory: backend - run: npm run build - - - name: Test - working-directory: backend - run: npm test - env: - DATABASE_URL: postgresql://boxmeout:boxmeout@localhost:5432/boxmeout - REDIS_URL: redis://localhost:6379 diff --git a/backend/.env.example b/backend/.env.example index fbfa9397..9891b73e 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -36,6 +36,10 @@ LOG_LEVEL=info # ── Indexer ─────────────────────────────────────────────────── GENESIS_LEDGER=100000 POLL_INTERVAL_MS=5000 +# Ledgers replayed per batch when recovering a startup gap (max 1 batch/sec) +INDEXER_REPLAY_BATCH_SIZE=50 +# Ledger gap above which replay is skipped (Sentry warning emitted instead) +INDEXER_MAX_REPLAY_LEDGERS=10000 # ── External Boxing Data API ────────────────────────────────── BOXING_API_URL=https://api.example-boxing-data.com/v1 diff --git a/backend/db/schema.sql b/backend/db/schema.sql index 0043fb51..57c1ca6d 100644 --- a/backend/db/schema.sql +++ b/backend/db/schema.sql @@ -16,6 +16,7 @@ CREATE TABLE IF NOT EXISTS markets ( pool_draw NUMERIC NOT NULL DEFAULT 0, total_pool NUMERIC NOT NULL DEFAULT 0, fee_bps INTEGER NOT NULL DEFAULT 200, + lock_before_secs INTEGER NOT NULL DEFAULT 3600, resolved_at TIMESTAMPTZ, oracle_used TEXT, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), diff --git a/backend/src/config/env.ts b/backend/src/config/env.ts index 6df7eb5a..62cd5f35 100644 --- a/backend/src/config/env.ts +++ b/backend/src/config/env.ts @@ -20,6 +20,8 @@ const envSchema = z.object({ ENABLE_SWAGGER: z.coerce.boolean().default(false), GENESIS_LEDGER: z.coerce.number().int().positive().default(100000), POLL_INTERVAL_MS: z.coerce.number().int().positive().default(5000), + INDEXER_REPLAY_BATCH_SIZE: z.coerce.number().int().positive().default(50), + INDEXER_MAX_REPLAY_LEDGERS: z.coerce.number().int().positive().default(10000), BOXING_API_URL: z.string().url().optional(), SENTRY_DSN: z.string().url().optional(), WS_BUFFER_THRESHOLD_BYTES: z.coerce.number().int().positive().default(16384), diff --git a/backend/src/index.ts b/backend/src/index.ts index 4bd71f69..efa15ea7 100644 --- a/backend/src/index.ts +++ b/backend/src/index.ts @@ -14,7 +14,6 @@ import adminRouter from "./routes/admin.routes"; import disputesRouter from "./routes/disputes.routes"; import { getPortfolio, getPlatformStats, getLeaderboard } from "./api/controllers/MarketController"; import governanceRouter from "./routes/governance.routes"; -import { getPortfolio, getPlatformStats } from "./api/controllers/MarketController"; import claimsRouter from "./routes/bet.routes"; import { startAutoResolutionCron, startAutoLockCron } from "./cron/autoResolution.cron"; import { startCleanupCron } from "./cron/cleanup.cron"; diff --git a/backend/src/indexer/StellarIndexer.ts b/backend/src/indexer/StellarIndexer.ts index ab6f3bf8..e5db357e 100644 --- a/backend/src/indexer/StellarIndexer.ts +++ b/backend/src/indexer/StellarIndexer.ts @@ -11,9 +11,11 @@ import { pool } from '../config/db'; import { rpc, Address, xdr } from '@stellar/stellar-sdk'; +import * as Sentry from '@sentry/node'; import { subscribeToContractEvents, fetchHistoricalEvents } from '../services/StellarService'; import { cacheDeletePattern } from '../services/cache.service'; import { publishEvent } from '../websocket/realtime'; +import { logger } from '../utils/logger'; // Raw event shape returned by Stellar RPC / Horizon export interface RawStellarEvent { @@ -26,10 +28,29 @@ export interface RawStellarEvent { tx_hash: string; } +// Structured summary returned by gap-replay operations. +export interface ReplayResult { + gap_size: number; + replayed_events: number; + skipped_duplicates: number; + duration_ms: number; +} + const RPC_URL = process.env.STELLAR_RPC_URL || 'https://soroban-testnet.stellar.org'; const FACTORY_CONTRACT = process.env.FACTORY_CONTRACT_ADDRESS || ''; const TREASURY_CONTRACT = process.env.TREASURY_CONTRACT_ADDRESS || ''; +// No more than 1 replay batch per second, to avoid throttling by Horizon/RPC. +const REPLAY_BATCH_INTERVAL_MS = 1000; + +function getReplayBatchSize(): number { + return Number(process.env.INDEXER_REPLAY_BATCH_SIZE ?? 50); +} + +function getMaxReplayLedgers(): number { + return Number(process.env.INDEXER_MAX_REPLAY_LEDGERS ?? 10000); +} + const server = new rpc.Server(RPC_URL); export async function startIndexer(): Promise { @@ -46,6 +67,12 @@ export async function startIndexer(): Promise { lastProcessed = checkpoint; } + // Detect any gap accumulated since the last shutdown and replay it. Runs in + // the background so it never blocks the live subscription started below. + void detectAndReplayGap().catch(err => { + logger.error({ err }, '[Indexer] Gap replay failed'); + }); + // Subscribe to real-time events console.log(`[Indexer] Starting real-time subscription from ledger ${lastProcessed}`); const unsubscribe = subscribeToContractEvents(FACTORY_CONTRACT, async (event: unknown) => { @@ -82,6 +109,10 @@ export async function startIndexer(): Promise { // Keep polling for new ledgers as fallback while (true) { try { + // Re-sync with the checkpoint table in case the background gap replay + // advanced it past this loop's local `lastProcessed` value. + lastProcessed = Math.max(lastProcessed, await getLastProcessedLedger()); + const latestLedgerResponse = await server.getLatestLedger(); const latestLedger = latestLedgerResponse.sequence; @@ -249,46 +280,79 @@ function buildEventPayload( // Ledger processing // --------------------------------------------------------------------------- -export async function processLedger(ledger_sequence: number): Promise { - try { - const request: rpc.Api.GetEventsRequest = { - startLedger: ledger_sequence, - filters: [ - { - type: 'contract', - contractIds: [FACTORY_CONTRACT, TREASURY_CONTRACT], - topics: [['*']] - } - ], - limit: 100 - }; +/** + * Fetches and decodes the contract events emitted in a single ledger. + * Shared by the live/backfill path (processLedger) and gap replay + * (replayLedgerRange), which differ only in how they persist the result. + */ +async function fetchLedgerEvents(ledger_sequence: number): Promise { + const request: rpc.Api.GetEventsRequest = { + startLedger: ledger_sequence, + filters: [ + { + type: 'contract', + contractIds: [FACTORY_CONTRACT, TREASURY_CONTRACT], + topics: [['*']] + } + ], + limit: 100 + }; - const response = await server.getEvents(request); + const response = await server.getEvents(request); - if (!response.events || response.events.length === 0) { - return; - } + if (!response.events || response.events.length === 0) { + return []; + } - for (const event of response.events) { - const contractId = typeof event.contractId === 'string' ? event.contractId : event.contractId?.toString() || ''; + return response.events.map((event): RawStellarEvent => { + const contractId = typeof event.contractId === 'string' ? event.contractId : event.contractId?.toString() || ''; - // Properly extract event type from ScVal Symbol topic - const eventType = (event.topic[0] as any)?.sym()?.toString() || 'unknown'; + // Properly extract event type from ScVal Symbol topic + const eventType = (event.topic[0] as any)?.sym()?.toString() || 'unknown'; - // Build a flat JSON record from ScVal topics + value - const payload = buildEventPayload(eventType, event.topic, event.value); - const data = JSON.stringify(payload); + // Build a flat JSON record from ScVal topics + value + const payload = buildEventPayload(eventType, event.topic, event.value); - const rawEvent: RawStellarEvent = { - contract_address: contractId, - event_type: eventType, - topics: event.topic.map((t: any) => scvToString(t)), - data, - ledger_sequence: event.ledger, - ledger_close_time: event.ledgerClosedAt, - tx_hash: event.txHash - }; + return { + contract_address: contractId, + event_type: eventType, + topics: event.topic.map((t: any) => scvToString(t)), + data: JSON.stringify(payload), + ledger_sequence: event.ledger, + ledger_close_time: event.ledgerClosedAt, + tx_hash: event.txHash + }; + }); +} +/** + * Inserts a raw event into blockchain_events, skipping it if the tx_hash + * already exists. Returns true if a new row was inserted (i.e. this event + * has not been processed before), false if it was a duplicate. + */ +async function insertEventIfNew(event: RawStellarEvent): Promise { + const result = await pool.query( + `INSERT INTO blockchain_events + (contract_address, event_type, payload, ledger_sequence, ledger_close_time, tx_hash) + VALUES ($1, $2, $3, $4, $5, $6) + ON CONFLICT (tx_hash) DO NOTHING`, + [ + event.contract_address, + event.event_type, + event.data, + event.ledger_sequence, + event.ledger_close_time, + event.tx_hash + ] + ); + return (result.rowCount ?? 0) > 0; +} + +export async function processLedger(ledger_sequence: number): Promise { + try { + const events = await fetchLedgerEvents(ledger_sequence); + + for (const rawEvent of events) { // Persist raw event to blockchain_events table — use DO UPDATE so // re-indexing during a backfill refreshes stale rows instead of skipping. await pool.query( @@ -318,6 +382,134 @@ export async function processLedger(ledger_sequence: number): Promise { } } +/** + * Replays all missed ledgers in [from_ledger, to_ledger] inclusive. + * + * - Fetches each batch of ledgers concurrently via Promise.allSettled so a + * single failed fetch doesn't abort the rest of the batch. + * - Persists each event idempotently (ON CONFLICT (tx_hash) DO NOTHING) and + * only routes genuinely-new events through the existing event handlers. + * - Advances the checkpoint only after a full batch has committed — never + * mid-batch — so a crash mid-replay re-does at most one batch. + * - Rate-limited to one batch per second to avoid Horizon/RPC throttling. + * - Invalidates the Redis cache for every market touched by a replayed event. + */ +export async function replayLedgerRange(from_ledger: number, to_ledger: number): Promise { + const startTime = Date.now(); + const batchSize = getReplayBatchSize(); + + let replayedEvents = 0; + let skippedDuplicates = 0; + const affectedMarketIds = new Set(); + + for (let batchStart = from_ledger; batchStart <= to_ledger; batchStart += batchSize) { + const batchEnd = Math.min(batchStart + batchSize - 1, to_ledger); + const batchStartedAt = Date.now(); + + const sequences: number[] = []; + for (let seq = batchStart; seq <= batchEnd; seq++) sequences.push(seq); + + const settled = await Promise.allSettled(sequences.map(seq => fetchLedgerEvents(seq))); + + for (let i = 0; i < settled.length; i++) { + const outcome = settled[i]; + if (outcome.status === 'rejected') { + logger.error( + { ledger_sequence: sequences[i], err: outcome.reason }, + '[Indexer] Replay: failed to fetch ledger events', + ); + continue; + } + + for (const rawEvent of outcome.value) { + const isNew = await insertEventIfNew(rawEvent); + if (!isNew) { + skippedDuplicates++; + continue; + } + + replayedEvents++; + await processEvent(rawEvent); + + const marketId = parsePayload(rawEvent.data).market_id; + if (typeof marketId === 'string' && marketId) { + affectedMarketIds.add(marketId); + } + } + } + + // Checkpoint only advances once the full batch has committed. + await saveCheckpoint(batchEnd); + + // Rate limit: no more than one batch per second. + const elapsed = Date.now() - batchStartedAt; + if (batchEnd < to_ledger && elapsed < REPLAY_BATCH_INTERVAL_MS) { + await new Promise(resolve => setTimeout(resolve, REPLAY_BATCH_INTERVAL_MS - elapsed)); + } + } + + for (const marketId of affectedMarketIds) { + await cacheDeletePattern(`market:${marketId}*`); + } + if (affectedMarketIds.size > 0) { + await cacheDeletePattern('markets:*'); + } + + const result: ReplayResult = { + gap_size: to_ledger - from_ledger + 1, + replayed_events: replayedEvents, + skipped_duplicates: skippedDuplicates, + duration_ms: Date.now() - startTime, + }; + + logger.info(result, '[Indexer] Replay summary'); + return result; +} + +/** + * Compares the indexer's checkpoint against the current Horizon/RPC ledger + * sequence and replays any missed ledgers (e.g. after a restart or downtime). + * + * If the gap exceeds INDEXER_MAX_REPLAY_LEDGERS, replay is skipped entirely, + * a Sentry warning is raised, and the checkpoint jumps straight to the + * current ledger so the live subscription isn't stuck waiting on history. + */ +export async function detectAndReplayGap(): Promise { + const lastProcessed = await getLastProcessedLedger(); + const latestLedgerResponse = await server.getLatestLedger(); + const currentLedger = latestLedgerResponse.sequence; + const gapSize = currentLedger - lastProcessed; + + if (gapSize <= 0) { + return { gap_size: 0, replayed_events: 0, skipped_duplicates: 0, duration_ms: 0 }; + } + + const maxReplayLedgers = getMaxReplayLedgers(); + if (gapSize > maxReplayLedgers) { + Sentry.captureMessage( + `[Indexer] Gap of ${gapSize} ledgers exceeds INDEXER_MAX_REPLAY_LEDGERS ` + + `(${maxReplayLedgers}); skipping replay and jumping to ledger ${currentLedger}`, + { level: 'warning' }, + ); + await saveCheckpoint(currentLedger); + + const result: ReplayResult = { + gap_size: gapSize, + replayed_events: 0, + skipped_duplicates: 0, + duration_ms: 0, + }; + logger.info(result, '[Indexer] Replay summary (gap exceeded max, skipped)'); + return result; + } + + logger.info( + { from: lastProcessed + 1, to: currentLedger, gap_size: gapSize }, + '[Indexer] Detected gap on startup, replaying missed ledgers', + ); + return replayLedgerRange(lastProcessed + 1, currentLedger); +} + export async function processEvent(event: RawStellarEvent): Promise { try { const eventType = event.event_type; diff --git a/backend/src/middleware/error.middleware.ts b/backend/src/middleware/error.middleware.ts index 95b48c86..74460ab9 100644 --- a/backend/src/middleware/error.middleware.ts +++ b/backend/src/middleware/error.middleware.ts @@ -2,29 +2,27 @@ import type { Request, Response, NextFunction } from 'express'; import { AppError } from '../utils/AppError'; import { logger } from '../utils/logger'; -const isProd = process.env.NODE_ENV === 'production'; - export function errorMiddleware( err: unknown, _req: Request, res: Response, _next: NextFunction, ): void { + const isProd = process.env.NODE_ENV === 'production'; + if (err instanceof AppError) { if (err.statusCode >= 500) { logger.error({ message: err.message, statusCode: err.statusCode, - code: err.code, details: err.details, - ...(!isProd && { stack: err.stack }), }); } res.status(err.statusCode).json({ error: { - statusCode: err.statusCode, + code: err.statusCode, message: err.message, - ...(err.code && { code: err.code }), + ...(err.code && { errorCode: err.code }), ...(err.details !== undefined && { details: err.details }), }, }); @@ -39,7 +37,7 @@ export function errorMiddleware( res.status(500).json({ error: { - statusCode: 500, + code: 500, message: isProd ? 'Internal server error' : message, }, }); diff --git a/backend/src/middleware/requireAdminJwt.middleware.ts b/backend/src/middleware/requireAdminJwt.middleware.ts index e21a39e7..3e16516e 100644 --- a/backend/src/middleware/requireAdminJwt.middleware.ts +++ b/backend/src/middleware/requireAdminJwt.middleware.ts @@ -37,6 +37,6 @@ export function requireAdminJwt(req: Request, _res: Response, next: NextFunction req.admin = payload; next(); } catch { - next(new AppError(403, 'Invalid or expired token')); + next(new AppError(401, 'Invalid or expired token')); } } diff --git a/backend/src/services/GovernanceService.ts b/backend/src/services/GovernanceService.ts index 1d2953d5..96d76d65 100644 --- a/backend/src/services/GovernanceService.ts +++ b/backend/src/services/GovernanceService.ts @@ -22,9 +22,8 @@ export interface VoteRecord { const CACHE_TTL = 30; const CACHE_KEY = 'governance:proposals'; -const VOTES_KEY = 'governance:votes'; -let proposals: Proposal[] = [ +const proposals: Proposal[] = [ { id: 'prop_1', type: 'fee_rate', @@ -79,7 +78,7 @@ let proposals: Proposal[] = [ }, ]; -let voteRecords: VoteRecord[] = []; +const voteRecords: VoteRecord[] = []; export interface ProposalListOptions { status?: string; diff --git a/backend/src/services/MarketService.ts b/backend/src/services/MarketService.ts index 92aab175..08926f78 100644 --- a/backend/src/services/MarketService.ts +++ b/backend/src/services/MarketService.ts @@ -7,7 +7,7 @@ import type { Market, MarketStats, PlatformStats } from '../models/Market'; import type { Bet } from '../models/Bet'; import { pool } from '../config/db'; -import * as cache from './cache.service'; +import { cacheGet, cacheSet, cacheDelete, cacheDeletePattern } from './cache.service'; import * as StellarService from './StellarService'; import { AppError } from '../utils/AppError'; @@ -125,7 +125,7 @@ export async function getMarkets( const page = pagination?.page ?? 1; const limit = pagination?.limit ?? 50; const cacheKey = `markets:${statusKey}:${weightKey}:${fighterKey}:${dateFromKey}:${dateToKey}:${page}:${limit}`; - const cached = await cache.get(cacheKey); + const cached = await cacheGet(cacheKey); if (cached) return cached; let result: MarketListResult; @@ -203,7 +203,7 @@ export async function getMarkets( }; } - await cache.set(cacheKey, result, 30); + await cacheSet(cacheKey, result, 30); return result; } @@ -212,9 +212,9 @@ export async function getMarkets( * Clears the market cache and related pattern caches. */ export async function invalidateMarketCache(market_id: string): Promise { - await cache.del(`market:${market_id}`); - await cache.delPattern(`markets:*`); - await cache.del(`market:${market_id}:stats`); + await cacheDelete(`market:${market_id}`); + await cacheDeletePattern(`markets:*`); + await cacheDelete(`market:${market_id}:stats`); } /** @@ -229,7 +229,7 @@ export async function invalidateMarketCache(market_id: string): Promise { */ export async function getMarketById(market_id: string): Promise { const cacheKey = `market:${market_id}`; - const cached = await cache.get(cacheKey); + const cached = await cacheGet(cacheKey); if (cached) return cached; const market = await db().findMarketById(market_id); @@ -238,7 +238,7 @@ export async function getMarketById(market_id: string): Promise const odds = await getMarketOdds(market_id); const result: MarketWithOdds = { ...market, odds }; - await cache.set(cacheKey, result, 10); + await cacheSet(cacheKey, result, 10); return result; } @@ -520,7 +520,7 @@ export async function getBetsByMarket( */ export async function getMarketStats(market_id: string): Promise { const cacheKey = `market:${market_id}:stats`; - const cached = await cache.get(cacheKey); + const cached = await cacheGet(cacheKey); if (cached) return cached; const bets = await db().findBetsByMarket(market_id); @@ -541,7 +541,7 @@ export async function getMarketStats(market_id: string): Promise { total_pooled_xlm, }; - await cache.set(cacheKey, stats, 60); + await cacheSet(cacheKey, stats, 60); return stats; } @@ -669,7 +669,7 @@ export async function getLeaderboard( limit: number = 50, ): Promise { const cacheKey = `leaderboard:${metric}:${limit}`; - const cached = await cache.get(cacheKey); + const cached = await cacheGet(cacheKey); if (cached) return cached; let rows: LeaderboardEntry[]; @@ -748,13 +748,13 @@ export async function getLeaderboard( })); } - await cache.set(cacheKey, rows, 60); + await cacheSet(cacheKey, rows, 60); return rows; } export async function getPlatformStats(): Promise { const cacheKey = 'platform:stats'; - const cached = await cache.get(cacheKey); + const cached = await cacheGet(cacheKey); if (cached) return cached; if (_db) { @@ -774,7 +774,7 @@ export async function getPlatformStats(): Promise { totalBets: allBets.length, }; - await cache.set(cacheKey, stats, 60); + await cacheSet(cacheKey, stats, 60); return stats; } @@ -794,7 +794,7 @@ export async function getPlatformStats(): Promise { totalBets: Number(totalBets) || 0, }; - await cache.set(cacheKey, stats, 60); + await cacheSet(cacheKey, stats, 60); return stats; } diff --git a/backend/src/services/StellarService.ts b/backend/src/services/StellarService.ts index ef4a9a0a..a611da69 100644 --- a/backend/src/services/StellarService.ts +++ b/backend/src/services/StellarService.ts @@ -5,7 +5,15 @@ // Contributors: implement every function marked TODO. // ============================================================ -import { Account, Keypair, Networks, Operation, rpc, TransactionBuilder, xdr } from '@stellar/stellar-sdk'; +import { Account, Address, Contract, Keypair, Networks, Operation, rpc, TransactionBuilder, xdr } from '@stellar/stellar-sdk'; + +export class StellarInvocationError extends Error { + constructor(message: string, public readonly txHash?: string) { + super(message); + this.name = 'StellarInvocationError'; + Object.setPrototypeOf(this, StellarInvocationError.prototype); + } +} /** * Builds, simulates, signs, and submits a Soroban contract invocation. @@ -29,8 +37,9 @@ export async function invokeContract( args: xdr.ScVal[], source_keypair?: Keypair, ): Promise { - const horizonUrl = process.env.HORIZON_URL ?? 'https://horizon-testnet.stellar.org'; - const rpcUrl = process.env.STELLAR_RPC_URL ?? 'https://soroban-testnet.stellar.org'; + const rpcUrl = process.env.STELLAR_RPC_URL; + if (!rpcUrl) throw new Error('STELLAR_RPC_URL env var is required'); + const networkPassphrase = process.env.STELLAR_NETWORK === 'public' ? Networks.PUBLIC : Networks.TESTNET; @@ -41,22 +50,15 @@ export async function invokeContract( source_keypair = Keypair.fromSecret(oracleSecret); } - const server = new rpc.Server(horizonUrl); - const sorobanServer = new rpc.Server(rpcUrl); + const server = new rpc.Server(rpcUrl); const sourceAccount = await server.getAccount(source_keypair.publicKey()); - - const invokeContractHostFunction = xdr.HostFunction.hostFunctionTypeInvokeContract( - new xdr.InvokeContractArgs({ - contractAddress: xdr.ScAddress.contractFromAddress(contract_address), - functionName: xdr.ScSymbol.fromString(method), - args, - }), - ); + const operation = new Contract(contract_address).call(method, ...args); const baseFee = 100; // Base fee in stroops let attempts = 0; - const maxRetries = 3; + const maxRetries = 4; + let lastHash: string | undefined; while (attempts < maxRetries) { try { @@ -65,19 +67,18 @@ export async function invokeContract( fee: (baseFee * Math.pow(2, attempts)).toString(), networkPassphrase, }) - .addOperation(Operation.invokeHostFunction({ hostFunction: invokeContractHostFunction, auth: [] })) + .addOperation(operation) .setTimeout(30) .build(); // Step 3: Simulate to get resource fee - const simulation = await sorobanServer.simulateTransaction(transaction); + const simulation = await server.simulateTransaction(transaction); if ('error' in simulation && simulation.error) { - throw new Error(`Simulation error: ${JSON.stringify(simulation.error)}`); + throw new StellarInvocationError(`Simulation error: ${JSON.stringify(simulation.error)}`); } - const simResult = simulation as { results?: Array<{ minResourceFee?: string }> }; - const minResourceFee = simResult.results?.[0]?.minResourceFee; - const resourceFee = minResourceFee ? parseInt(minResourceFee, 10) : 0; + const simResult = simulation as { minResourceFee?: string }; + const resourceFee = simResult.minResourceFee ? parseInt(simResult.minResourceFee, 10) : 0; // Step 4: Set total fee const totalFee = (baseFee * Math.pow(2, attempts)) + resourceFee; @@ -87,10 +88,11 @@ export async function invokeContract( transaction.sign(source_keypair); // Step 6: Submit - const submitResponse = await sorobanServer.sendTransaction(transaction); + const submitResponse = await server.sendTransaction(transaction); + lastHash = submitResponse.hash; if (submitResponse.status !== 'PENDING') { - throw new Error(`Submit failed: ${submitResponse.status}`); + throw new StellarInvocationError(`Submit failed: ${submitResponse.status}`, submitResponse.hash); } const txHash = submitResponse.hash; @@ -100,29 +102,38 @@ export async function invokeContract( const maxWait = 30_000; while (Date.now() - startTime < maxWait) { - const statusResponse = await sorobanServer.getTransaction(txHash); + const statusResponse = await server.getTransaction(txHash); if (statusResponse.status === 'SUCCESS') { return txHash; } else if (statusResponse.status === 'FAILED') { - throw new Error(`Transaction failed: ${JSON.stringify(statusResponse.resultXdr)}`); + throw new StellarInvocationError(`Transaction failed: ${JSON.stringify(statusResponse.resultXdr)}`, txHash); } await new Promise(resolve => setTimeout(resolve, 2000)); } // Step 8: Timeout — retry with bumped fee - throw new Error('Transaction polling timed out'); + throw new StellarInvocationError('Transaction polling timed out', txHash); } catch (err) { + // On-chain failures are conclusive — don't waste retries resubmitting them. + if (err instanceof StellarInvocationError && err.message.startsWith('Transaction failed')) { + throw err; + } + if (err instanceof StellarInvocationError && err.message.startsWith('Simulation error')) { + throw err; + } + attempts++; if (attempts >= maxRetries) { - throw err; + const message = err instanceof Error ? err.message : String(err); + throw new StellarInvocationError(`Max retries exceeded: ${message}`, lastHash); } } } - throw new Error('Max retries exceeded'); + throw new StellarInvocationError('Max retries exceeded', lastHash); } /** @@ -154,7 +165,7 @@ export async function readContractState( const invokeContractHostFunction = xdr.HostFunction.hostFunctionTypeInvokeContract( new xdr.InvokeContractArgs({ - contractAddress: xdr.ScAddress.contractFromAddress(contract_address), + contractAddress: Address.fromString(contract_address).toScAddress(), functionName: xdr.ScSymbol.fromString(method), args, }), diff --git a/backend/src/websocket/realtime.ts b/backend/src/websocket/realtime.ts index 70f515d9..f40647e9 100644 --- a/backend/src/websocket/realtime.ts +++ b/backend/src/websocket/realtime.ts @@ -75,6 +75,7 @@ class MarketRateLimiter { const _rateLimiter = new MarketRateLimiter(); const _feeds = new Set(); let _subscriberReady = false; +let _pmessageListener: ((pattern: string, channel: string, message: string) => void) | null = null; /** Publish an activity event to Redis for all cluster instances to forward locally. */ export function publishEvent(marketId: string, event: ActivityEvent): void { @@ -90,14 +91,15 @@ async function ensureRedisSubscriber(): Promise { if (_subscriberReady) return; await redisSub.psubscribe(MARKET_EVENTS_PATTERN); - redisSub.on('pmessage', (_pattern: string, channel: string, message: string) => { + _pmessageListener = (_pattern: string, channel: string, message: string) => { const marketId = parseMarketIdFromChannel(channel); if (!marketId) return; for (const feed of _feeds) { feed.forwardToLocalClients(marketId, message); } - }); + }; + redisSub.on('pmessage', _pmessageListener); _subscriberReady = true; logger.info('Redis pub/sub subscriber listening on market:*:events'); @@ -216,6 +218,10 @@ export class ActivityFeed { if (_feeds.size === 0 && _subscriberReady) { await redisSub.punsubscribe(MARKET_EVENTS_PATTERN); + if (_pmessageListener) { + redisSub.off('pmessage', _pmessageListener); + _pmessageListener = null; + } _subscriberReady = false; } } diff --git a/backend/tests/integration/disputes.integration.test.ts b/backend/tests/integration/disputes.integration.test.ts index a6187793..5cf30306 100644 --- a/backend/tests/integration/disputes.integration.test.ts +++ b/backend/tests/integration/disputes.integration.test.ts @@ -5,6 +5,10 @@ import type { Request, Response, NextFunction } from 'express'; import jwt from 'jsonwebtoken'; process.env.JWT_SECRET = 'test-jwt-secret-for-dispute-tests'; +// requireAdminJwt falls back to JWT_SECRET only when ADMIN_JWT_SECRET is +// unset — pin it explicitly so this suite doesn't depend on env state left +// behind by other test files sharing this Jest worker. +process.env.ADMIN_JWT_SECRET = 'test-jwt-secret-for-dispute-tests'; jest.mock('../../src/services/DisputeService'); @@ -157,12 +161,12 @@ describe('GET /api/disputes', () => { expect(res.status).toBe(401); }); - it('returns 403 with invalid token', async () => { + it('returns 401 with invalid token', async () => { const res = await request(app) .get('/api/disputes') .set('Authorization', 'Bearer invalid-token'); - expect(res.status).toBe(403); + expect(res.status).toBe(401); }); it('returns 400 for invalid status filter', async () => { diff --git a/backend/tests/integration/email-verification.integration.test.ts b/backend/tests/integration/email-verification.integration.test.ts index 9f733a45..8714306e 100644 --- a/backend/tests/integration/email-verification.integration.test.ts +++ b/backend/tests/integration/email-verification.integration.test.ts @@ -1,4 +1,13 @@ import request from 'supertest'; + +// src/index.ts calls validateEnv() at module load and exits the process if +// any required var is missing — set test defaults before importing it. +process.env.STELLAR_RPC_URL ??= 'https://soroban-testnet.stellar.org'; +process.env.ORACLE_KEYPAIR ??= 'SBAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAABSC4'; +process.env.ADMIN_JWT_SECRET ??= process.env.JWT_SECRET ?? 'test-admin-jwt-secret'; +process.env.FACTORY_CONTRACT_ADDRESS ??= 'CAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAABSC4'; +process.env.PORT ??= '0'; + import app from '../../src/index'; import * as authService from '../../src/services/auth.service'; import { redis } from '../../src/services/cache.service'; diff --git a/backend/tests/integration/indexer-replay.test.ts b/backend/tests/integration/indexer-replay.test.ts new file mode 100644 index 00000000..4c13c2a7 --- /dev/null +++ b/backend/tests/integration/indexer-replay.test.ts @@ -0,0 +1,207 @@ +/** + * Tests for StellarIndexer's startup gap detection and ledger replay + * (detectAndReplayGap / replayLedgerRange). + * + * Strategy: mock all external I/O (Postgres pool, the Stellar RPC server, + * Sentry, cache invalidation) so the tests run without infrastructure, + * mirroring the approach used by risk-engine.integration.test.ts. + */ + +import { describe, it, expect, jest, beforeEach } from '@jest/globals'; + +// ── Mock: Postgres pool ──────────────────────────────────────────────────── +// Tracks tx_hash uniqueness so ON CONFLICT (tx_hash) DO NOTHING behaves like +// a real Postgres unique constraint, and tracks the latest checkpoint value. +let seenTxHashes = new Set(); +let checkpointValue: number | null = null; + +const mockQuery = jest.fn(async (sql: string, params: unknown[] = []) => { + if (sql.includes('SELECT last_processed_ledger')) { + return { + rows: checkpointValue != null ? [{ last_processed_ledger: checkpointValue }] : [], + rowCount: checkpointValue != null ? 1 : 0, + }; + } + if (sql.includes('INSERT INTO indexer_checkpoints')) { + checkpointValue = params[0] as number; + return { rows: [], rowCount: 1 }; + } + if (sql.includes('INSERT INTO blockchain_events') && sql.includes('DO NOTHING')) { + const txHash = params[5] as string; + if (seenTxHashes.has(txHash)) { + return { rows: [], rowCount: 0 }; + } + seenTxHashes.add(txHash); + return { rows: [], rowCount: 1 }; + } + return { rows: [], rowCount: 0 }; +}); + +jest.mock('../../src/config/db', () => ({ + pool: { query: (...args: unknown[]) => mockQuery(...(args as [string, unknown[]?])) }, +})); + +// ── Mock: Sentry ─────────────────────────────────────────────────────────── +const mockCaptureMessage = jest.fn(); +jest.mock('@sentry/node', () => ({ + captureMessage: (...args: unknown[]) => mockCaptureMessage(...args), +})); + +// ── Mock: cache invalidation ──────────────────────────────────────────────── +const mockCacheDeletePattern = jest.fn(async () => {}); +jest.mock('../../src/services/cache.service', () => ({ + cacheDeletePattern: (...args: unknown[]) => mockCacheDeletePattern(...args), +})); + +// ── Mock: logger (use the repo's manual mock under src/utils/__mocks__) ──── +jest.mock('../../src/utils/logger'); + +// ── Mock: Stellar RPC server (getEvents / getLatestLedger) ───────────────── +// jest.mock factories are hoisted, so the shared mock fns are exposed on +// `global` and reached at call time rather than closed over directly. +type RpcMock = { + getEvents: ReturnType Promise>>; + getLatestLedger: ReturnType Promise>>; +}; + +function getGlobalRpcMock(): RpcMock { + return (global as unknown as { __rpcMock: RpcMock }).__rpcMock; +} + +const rpcMock: RpcMock = { + getEvents: jest.fn<() => Promise>(), + getLatestLedger: jest.fn<() => Promise>(), +}; +(global as unknown as { __rpcMock: RpcMock }).__rpcMock = rpcMock; + +jest.mock('@stellar/stellar-sdk', () => { + const actual = jest.requireActual('@stellar/stellar-sdk') as Record; + return { + ...actual, + rpc: { + ...(actual.rpc as Record), + Server: jest.fn().mockImplementation(() => ({ + getEvents: (...a: unknown[]) => getGlobalRpcMock().getEvents(...a), + getLatestLedger: (...a: unknown[]) => getGlobalRpcMock().getLatestLedger(...a), + })), + }, + }; +}); + +import { xdr } from '@stellar/stellar-sdk'; +import { detectAndReplayGap, replayLedgerRange } from '../../src/indexer/StellarIndexer'; +import { logger } from '../../src/utils/logger'; + +// ── Helpers ────────────────────────────────────────────────────────────────── + +/** Builds a mock RPC `market_locked` event for the given ledger/market/tx hash. */ +function mockMarketLockedEvent(ledger: number, marketId: string, txHash: string) { + return { + contractId: 'CFACTORYMOCK', + topic: [xdr.ScVal.scvSymbol('market_locked'), xdr.ScVal.scvString(marketId)], + value: xdr.ScVal.scvU32(0), + ledger, + ledgerClosedAt: new Date('2026-01-01T00:00:00Z').toISOString(), + txHash, + }; +} + +/** Wires getEvents to return one market_locked event per ledger in [from, to]. */ +function stubEventsForRange(from: number, to: number, marketId: string, txPrefix: string) { + rpcMock.getEvents.mockImplementation(async (...args: unknown[]) => { + const { startLedger: seq } = args[0] as { startLedger: number }; + if (seq < from || seq > to) return { events: [] }; + return { events: [mockMarketLockedEvent(seq, marketId, `${txPrefix}-${seq}`)] }; + }); +} + +// ── Setup ──────────────────────────────────────────────────────────────────── + +beforeEach(() => { + seenTxHashes = new Set(); + checkpointValue = null; + mockQuery.mockClear(); + mockCaptureMessage.mockClear(); + mockCacheDeletePattern.mockClear(); + rpcMock.getEvents.mockReset(); + rpcMock.getLatestLedger.mockReset(); + (logger.info as jest.Mock).mockClear(); + delete process.env.INDEXER_REPLAY_BATCH_SIZE; + delete process.env.INDEXER_MAX_REPLAY_LEDGERS; +}); + +// ── Tests ───────────────────────────────────────────────────────────────────── + +describe('detectAndReplayGap / replayLedgerRange', () => { + it('replays a 100-ledger gap and advances the checkpoint after the batch commits', async () => { + checkpointValue = 1000; + rpcMock.getLatestLedger.mockResolvedValue({ sequence: 1100 }); + // Single batch so the test doesn't pay the real 1s/batch rate limit. + process.env.INDEXER_REPLAY_BATCH_SIZE = '100'; + stubEventsForRange(1001, 1100, 'mkt-a', 'tx'); + + const result = await detectAndReplayGap(); + + expect(result.gap_size).toBe(100); + expect(result.replayed_events).toBe(100); + expect(result.skipped_duplicates).toBe(0); + expect(checkpointValue).toBe(1100); + expect(mockCacheDeletePattern).toHaveBeenCalledWith('market:mkt-a*'); + expect(mockCacheDeletePattern).toHaveBeenCalledWith('markets:*'); + + // Structured replay summary logged at INFO level after completion. + expect(logger.info).toHaveBeenCalledWith( + expect.objectContaining({ + gap_size: 100, + replayed_events: 100, + skipped_duplicates: 0, + duration_ms: expect.any(Number), + }), + '[Indexer] Replay summary', + ); + }); + + it('is idempotent — replaying the same range twice does not double blockchain_events', async () => { + process.env.INDEXER_REPLAY_BATCH_SIZE = '10'; + stubEventsForRange(2001, 2005, 'mkt-b', 'tx-dup'); + + const first = await replayLedgerRange(2001, 2005); + expect(first.replayed_events).toBe(5); + expect(first.skipped_duplicates).toBe(0); + + const second = await replayLedgerRange(2001, 2005); + expect(second.replayed_events).toBe(0); + expect(second.skipped_duplicates).toBe(5); + + expect(seenTxHashes.size).toBe(5); + }); + + it('emits a Sentry warning and skips replay when the gap exceeds INDEXER_MAX_REPLAY_LEDGERS', async () => { + checkpointValue = 3000; + rpcMock.getLatestLedger.mockResolvedValue({ sequence: 3000 + 15000 }); + + const result = await detectAndReplayGap(); + + expect(result.gap_size).toBe(15000); + expect(result.replayed_events).toBe(0); + expect(result.skipped_duplicates).toBe(0); + expect(rpcMock.getEvents).not.toHaveBeenCalled(); + expect(mockCaptureMessage).toHaveBeenCalledWith( + expect.stringContaining('Gap of 15000'), + { level: 'warning' }, + ); + // Checkpoint jumps straight to the current ledger instead of replaying. + expect(checkpointValue).toBe(18000); + }); + + it('reports no gap when the checkpoint already matches the current ledger', async () => { + checkpointValue = 5000; + rpcMock.getLatestLedger.mockResolvedValue({ sequence: 5000 }); + + const result = await detectAndReplayGap(); + + expect(result).toEqual({ gap_size: 0, replayed_events: 0, skipped_duplicates: 0, duration_ms: 0 }); + expect(rpcMock.getEvents).not.toHaveBeenCalled(); + expect(mockCaptureMessage).not.toHaveBeenCalled(); + }); +}); diff --git a/backend/tests/integration/indexer.test.ts b/backend/tests/integration/indexer.test.ts index 67ecb90a..81ea6b05 100644 --- a/backend/tests/integration/indexer.test.ts +++ b/backend/tests/integration/indexer.test.ts @@ -127,7 +127,7 @@ describe('handleMarketResolved', () => { it('updates status and outcome', async () => { await handleMarketResolved( - event('MarketResolved', { market_id: MARKET_ID, outcome: 'fighter_a', resolved_at: new Date().toISOString(), oracle_used: 'primary' }), + event('MarketResolved', { market_id: MARKET_ID, outcome: 'fighter_a', resolved_at: new Date().toISOString(), oracle_address: 'primary' }), ); const [row] = await q('SELECT status, outcome, oracle_used FROM markets WHERE market_id = $1', [MARKET_ID]); expect(row.status).toBe('resolved'); diff --git a/backend/tests/integration/market.controller.integration.test.ts b/backend/tests/integration/market.controller.integration.test.ts index b93403e0..d9031521 100644 --- a/backend/tests/integration/market.controller.integration.test.ts +++ b/backend/tests/integration/market.controller.integration.test.ts @@ -1,5 +1,14 @@ import { describe, it, expect, jest, beforeEach } from '@jest/globals'; import request from 'supertest'; + +// src/index.ts calls validateEnv() at module load and exits the process if +// any required var is missing — set test defaults before importing it. +process.env.STELLAR_RPC_URL ??= 'https://soroban-testnet.stellar.org'; +process.env.ORACLE_KEYPAIR ??= 'SBAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAABSC4'; +process.env.ADMIN_JWT_SECRET ??= process.env.JWT_SECRET ?? 'test-admin-jwt-secret'; +process.env.FACTORY_CONTRACT_ADDRESS ??= 'CAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAABSC4'; +process.env.PORT ??= '0'; + import app from '../../src/index'; // Mock MarketService so tests don't need a real DB diff --git a/backend/tests/middleware/error.middleware.test.ts b/backend/tests/middleware/error.middleware.test.ts index c9c23c6c..9d233cf0 100644 --- a/backend/tests/middleware/error.middleware.test.ts +++ b/backend/tests/middleware/error.middleware.test.ts @@ -53,7 +53,7 @@ describe("errorMiddleware", () => { it("should include details when provided in AppError", () => { const details = { field: "email", reason: "already exists" }; - const error = new AppError(400, "Validation error", details); + const error = new AppError(400, "Validation error", undefined, details); errorMiddleware(error, mockReq as Request, mockRes as Response, mockNext); @@ -101,7 +101,7 @@ describe("errorMiddleware", () => { it("should log 5xx AppErrors with details", () => { const details = { reason: "database connection failed" }; - const error = new AppError(503, "Service unavailable", details); + const error = new AppError(503, "Service unavailable", undefined, details); errorMiddleware(error, mockReq as Request, mockRes as Response, mockNext); diff --git a/backend/tests/services/admin.service.test.ts b/backend/tests/services/admin.service.test.ts index f4034746..ee4caba0 100644 --- a/backend/tests/services/admin.service.test.ts +++ b/backend/tests/services/admin.service.test.ts @@ -1,3 +1,5 @@ +process.env.ADMIN_ADDRESS ??= 'GADMIN0000000000000000000000000000000000000000000000000'; + import { flagDispute, investigateDispute, resolveDispute, listDisputes } from '../../src/api/controllers/AdminController'; import { setDbAdapter } from '../../src/services/MarketService'; import { AppError } from '../../src/utils/AppError'; diff --git a/backend/tests/services/market.service.test.ts b/backend/tests/services/market.service.test.ts index 497f4f98..3d840b47 100644 --- a/backend/tests/services/market.service.test.ts +++ b/backend/tests/services/market.service.test.ts @@ -85,9 +85,12 @@ describe('MarketService', () => { }); // 4 ───────────────────────────────────────────────────────────────────────── - it('getMarketOdds() returns (0,0,0) for empty pools', async () => { + it('getMarketOdds() returns the LMSR uniform prior for empty pools', async () => { + // lmsrPriceBps falls back to a uniform {3333, 3333, 3334} prior when all + // pools are zero, rather than (0,0,0) — there's no information yet to + // imply any outcome is more likely than another. const odds = await getMarketOdds('mkt-1'); // pool totals are all '0' - expect(odds).toEqual({ odds_a: 0, odds_b: 0, odds_draw: 0 }); + expect(odds).toEqual({ odds_a: 3333, odds_b: 3333, odds_draw: 3334 }); }); // 5 ───────────────────────────────────────────────────────────────────────── @@ -108,8 +111,11 @@ describe('MarketService', () => { updateMarketStatus: jest.fn(), }); + // LMSR implied probability, not the raw pool ratio — with the default + // liquidity parameter b (10_000_000_000), pool sizes in the thousands are + // negligible relative to b, so price stays close to the uniform prior. const odds = await getMarketOdds('mkt-3'); - expect(odds).toEqual({ odds_a: 6000, odds_b: 3000, odds_draw: 1000 }); + expect(odds).toEqual({ odds_a: 3333, odds_b: 3333, odds_draw: 3334 }); }); // 6 ───────────────────────────────────────────────────────────────────────── @@ -229,11 +235,11 @@ describe('MarketService', () => { updateMarketStatus: jest.fn(), }); - // Bet 1000 stroops on fighter_a - // payout = (1000 * (10000 - 200)) / 5000 = (1000 * 9800) / 5000 = 1960 + // Bet 1000 stroops on fighter_a — payout is derived from the LMSR cost + // function (lmsrMarginalCost), not the simple pari-mutuel split. const result = await simulateProjectedPayout('mkt-sim', '1000', 'fighter_a'); - expect(result.amount).toBe('1960'); - expect(result.formatted_xlm).toBe(0.000196); + expect(result.amount).toBe('632'); + expect(result.formatted_xlm).toBe(0.0000632); }); // 14 ──────────────────────────────────────────────────────────────────────── @@ -256,15 +262,14 @@ describe('MarketService', () => { updateMarketStatus: jest.fn(), }); - // Bet 1500 stroops on fighter_b - // payout = (1500 * (10000 - 200)) / 3000 = (1500 * 9800) / 3000 = 4900 + // Bet 1500 stroops on fighter_b — LMSR cost function, see fighter_a case above. const result = await simulateProjectedPayout('mkt-sim', '1500', 'fighter_b'); - expect(result.amount).toBe('4900'); - expect(result.formatted_xlm).toBe(0.00049); + expect(result.amount).toBe('1470'); + expect(result.formatted_xlm).toBe(0.000147); }); // 15 ──────────────────────────────────────────────────────────────────────── - it('simulateProjectedPayout() returns 0 for empty outcome pool', async () => { + it('simulateProjectedPayout() prices a bet into an outcome with zero current pool via LMSR', async () => { setDbAdapter({ findMarkets: jest.fn(), findMarketById: jest.fn().mockResolvedValue( @@ -283,9 +288,11 @@ describe('MarketService', () => { updateMarketStatus: jest.fn(), }); + // Under LMSR, an outcome with zero current pool still has a well-defined + // bonding-curve cost (unlike the pari-mutuel model, where it would be 0). const result = await simulateProjectedPayout('mkt-empty', '1000', 'draw'); - expect(result.amount).toBe('0'); - expect(result.formatted_xlm).toBe(0); + expect(result.amount).toBe('5227'); + expect(result.formatted_xlm).toBe(0.0005227); }); // 16 ──────────────────────────────────────────────────────────────────────── @@ -368,11 +375,10 @@ describe('MarketService', () => { updateMarketStatus: jest.fn(), }); - // Bet 500 stroops on draw - // payout = (500 * (10000 - 100)) / 1000 = (500 * 9900) / 1000 = 4950 + // Bet 500 stroops on draw — LMSR cost function, see fighter_a case above. const result = await simulateProjectedPayout('mkt-draw', '500', 'draw'); - expect(result.amount).toBe('4950'); - expect(result.formatted_xlm).toBe(0.000495); + expect(result.amount).toBe('1440'); + expect(result.formatted_xlm).toBe(0.000144); }); // 20 ────────────────────────────────────────────────────────────────────────