diff --git a/.env.example b/.env.example index 9fd6adf5..77f03486 100644 --- a/.env.example +++ b/.env.example @@ -152,3 +152,9 @@ TRUST_PROXY=false # Smoke-test query timeout in milliseconds (default: 15000) # BACKUP_SMOKE_TIMEOUT_MS=15000 +# ── Soroban indexer: testnet deployment (src/services/indexerService.ts) ────── + +# Both are required by the env schema; set them to run the indexer against testnet. +# Only contract events from PREDICTIFY_CONTRACT_ID are decoded and stored. +# SOROBAN_RPC_URL=https://soroban-testnet.stellar.org +# PREDICTIFY_CONTRACT_ID=CCWRJXIUBSMFQ4AF2RXXH3AFBW6MEZG4GK4IYES7C4Z3X4PR6K7VAFHC diff --git a/docs/indexer-events.md b/docs/indexer-events.md new file mode 100644 index 00000000..0268e56d --- /dev/null +++ b/docs/indexer-events.md @@ -0,0 +1,126 @@ +# Indexer events + +The indexer (`src/services/indexerService.ts`) reads contract events for +`PREDICTIFY_CONTRACT_ID` from the Soroban RPC and stores one row per event in +`indexer_events`. `src/indexer/eventDecoder.ts` turns each raw event (topic +`ScVal`s plus the payload `ScVal`) into a typed, validated record before it is +stored. + +Source of truth for symbols and layouts: `contracts/predictify-hybrid/src/events.rs` +in `Predictify-org/predictify-contracts` (commit `78ce97e`). Line numbers below +refer to that commit. + +## Topics + +| Symbol (`topic[0]`) | Topic layout | Payload fields (`events.rs` struct) | Backend table fed | `events.rs` | +|---|---|---|---|---| +| `mkt_crt` | `(mkt_crt, market_id: Symbol)` | `MarketCreatedEvent { market_id, question: String, outcomes: Vec, admin: Address, end_time: u64, nonce: u64, timestamp: u64 }` | `indexer_events` | 2518 | +| `bet_plc` | `(bet_plc, market_id: Symbol)` | `BetPlacedEvent { market_id, bettor: Address, outcome: String, amount: i128, nonce: u64, timestamp: u64 }` | `indexer_events`; read by `predictionsConfirmer` to confirm `predictions` | 2667 | +| `bet_upd` | `(bet_upd, market_id: Symbol)` | `BetStatusUpdatedEvent { market_id, bettor: Address, old_status: String, new_status: String, payout_amount: Option, nonce: u64, timestamp: u64 }` | `indexer_events` | 2717 | +| `mkt_res` | `(mkt_res, market_id: Symbol, resolution_method: String)` | `MarketResolvedEvent { market_id, final_outcome: String, oracle_result: String, community_consensus: String, resolution_method: String, confidence_score: i128, nonce: u64, timestamp: u64 }` | `indexer_events` | 3022 | +| `win_clm` | `(win_clm, market_id: Symbol)` | `WinningsClaimedEvent { market_id, user: Address, amount: i128, nonce: u64, timestamp: u64 }` | `indexer_events` | 3946 | + +Projecting decoded events into `markets` and `predictions` (beyond the +confirmer join) is out of scope; the five topics above feed `indexer_events` +only. + +### Aliases + +`event_topic_catalog.rs` lists long names that differ from the symbols +`events.rs` publishes. The decoder accepts them as aliases and stores the +emitted symbol: + +| Catalog name | Stored as | +|---|---| +| `market_created` | `mkt_crt` | +| `bet_placed` | `bet_plc` | +| `market_resolved` | `mkt_res` | +| `winnings_claimed` | `win_clm` | + +`bet_upd` has no catalog alias. + +### Anything else + +Every other topic the contract publishes (for example `vote`, `fbk_used`, +`stats_upd`) is stored with `event_type = 'unknown:'`. A new topic added +on-chain therefore needs no code change to be indexed. Batch claim events are +not decoded. + +## Stored row + +| Column | Value | +|---|---| +| `event_type` | the topic symbol (`bet_plc`), `unknown:`, or `invalid:` | +| `market_id` | `topic[1]` for market-scoped events, otherwise `NULL` | +| `data` | JSON object, see below | + +- **Decoded** (`event_type` is a topic symbol): `data` holds the payload fields + in camelCase (`bettor`, `outcome`, `amount`, `nonce`, `timestamp`, ...). + `market_id` is the column, not repeated in `data`. +- **Unknown** (`unknown:`): `data = { "raw": { "topics": [...], "value": } }`. +- **Invalid** (`invalid:`): a known topic whose layout or payload failed + validation. `data = { "raw": {...}, "errors": [...] }`. Validation + covers missing or mistyped fields, an address that is not a `G...`/`C...` + strkey, `topic[1]` or `topic[2]` disagreeing with the payload, and + out-of-range integers. `invalid:topic` is used when `topic[0]` is absent, is + not a `Symbol`, or the XDR cannot be parsed. + +Unknown and invalid events are never dropped and never throw out of a batch. +Events whose RPC `type` is not `contract`, or that come from another contract, +are filtered out before persistence. + +### Numbers + +All 64- and 128-bit integers (`amount`, `payout_amount`, `confidence_score`, +`nonce`, `timestamp`, `end_time`) are decimal strings. No floating point is used, +so `i128::MAX` round-trips as `170141183460469231731687303715884105727`. +`payout_amount` is `null` when the contract emitted `None`. + +## Confirmer join + +`predictionsConfirmer` confirms a pending prediction only when `indexer_events` +has a row with all of: + +- the same `tx_hash` as the prediction, +- `event_type = 'bet_plc'` and `canonical = true`, +- `market_id` equal to the prediction's market, +- `data->>'bettor'` equal to the prediction user's `stellar_address`. + +An `unknown:*`, `invalid:*` or other-topic event in the same transaction does not +confirm anything, and does not stop the prediction from aging out to `failed`. + +## Metrics + +| Metric | Labels | Meaning | +|---|---|---| +| `indexer_events_decoded_total` | `event_type` | Events decoded, by topic symbol. Unknown topics are grouped under `unknown` to keep label cardinality bounded. | +| `indexer_events_invalid_total` | `event_type` | Events that failed validation, by topic symbol (`topic` for an unreadable `topic[0]`). | + +## Testnet configuration + +```bash +export SOROBAN_RPC_URL=https://soroban-testnet.stellar.org +export PREDICTIFY_CONTRACT_ID=CCWRJXIUBSMFQ4AF2RXXH3AFBW6MEZG4GK4IYES7C4Z3X4PR6K7VAFHC +npm run indexer:dev +``` + +Both are commented in `.env.example`. + +## Cross-check against `events.rs` + +```bash +grep -n 'symbol_short!("\(mkt_crt\|bet_plc\|bet_upd\|mkt_res\|win_clm\)")' events.rs | grep -v nonce +``` + +``` +2516: Self::store_event(env, &symbol_short!("mkt_crt"), &event); +2518: .publish((symbol_short!("mkt_crt"), market_id.clone()), event); +2665: Self::store_event(env, &symbol_short!("bet_plc"), &event); +2667: .publish((symbol_short!("bet_plc"), market_id.clone()), event); +2715: Self::store_event(env, &symbol_short!("bet_upd"), &event); +2717: .publish((symbol_short!("bet_upd"), market_id.clone()), event); +3019: env.storage().persistent().set(&symbol_short!("mkt_res"), &event); +3022: symbol_short!("mkt_res"), +3944: Self::store_event(env, &symbol_short!("win_clm"), &event); +3946: .publish((symbol_short!("win_clm"), market_id.clone()), event); +``` diff --git a/src/indexer/eventDecoder.ts b/src/indexer/eventDecoder.ts new file mode 100644 index 00000000..66a8adbd --- /dev/null +++ b/src/indexer/eventDecoder.ts @@ -0,0 +1,531 @@ +/** + * eventDecoder.ts + * + * Pure, deterministic decoder for events published by the Predictify contract + * (`contracts/predictify-hybrid/src/events.rs` in predictify-contracts). + * + * The decoder reads `topic[0]` as the event symbol, `topic[1]` as the + * `market_id` symbol where present, converts the payload with `scValToNative` + * and validates it with one zod schema per topic. It performs no I/O and never + * throws: anything it cannot understand comes back as an `unknown` or + * `invalid` result carrying the raw base64 XDR. + * + * Number handling: every 64/128-bit integer (i128 amounts, u64 nonces and + * timestamps) is emitted as a decimal string. No floating point is involved. + */ + +import { scValToNative, StrKey, xdr } from "@stellar/stellar-sdk"; +import { z } from "zod"; + +// ─── Public types ──────────────────────────────────────────────────────────── + +/** Topic symbols published by the contract that this decoder understands. */ +export const DECODED_TOPICS = [ + "mkt_crt", + "bet_plc", + "bet_upd", + "mkt_res", + "win_clm", +] as const; + +export type DecodedTopic = (typeof DECODED_TOPICS)[number]; + +const DECODED_TOPIC_SET: ReadonlySet = new Set(DECODED_TOPICS); + +/** + * `event_topic_catalog.rs` lists long names that differ from the symbols + * `events.rs` actually emits. They are accepted as aliases of the emitted + * symbol and normalised to it. + */ +export const TOPIC_ALIASES: Readonly> = { + market_created: "mkt_crt", + bet_placed: "bet_plc", + market_resolved: "mkt_res", + winnings_claimed: "win_clm", +}; + +const TOPIC_ALIAS_MAP: ReadonlyMap = new Map(Object.entries(TOPIC_ALIASES)); + +/** A raw event: XDR `ScVal` objects or base64-encoded XDR strings. */ +export interface RawContractEvent { + topic: ReadonlyArray; + value: xdr.ScVal | string; +} + +export interface RawXdr { + topics: string[]; + value: string; +} + +export interface MarketCreatedFields { + question: string; + outcomes: string[]; + admin: string; + endTime: string; + nonce: string; + timestamp: string; +} + +export interface BetPlacedFields { + bettor: string; + outcome: string; + amount: string; + nonce: string; + timestamp: string; +} + +export interface BetUpdatedFields { + bettor: string; + oldStatus: string; + newStatus: string; + payoutAmount: string | null; + nonce: string; + timestamp: string; +} + +export interface MarketResolvedFields { + resolutionMethod: string; + finalOutcome: string; + oracleResult: string; + communityConsensus: string; + confidenceScore: string; + nonce: string; + timestamp: string; +} + +export interface WinningsClaimedFields { + user: string; + amount: string; + nonce: string; + timestamp: string; +} + +interface DecodedBase { + status: "decoded"; + eventType: T; + marketId: string; +} + +export type DecodedEvent = + | (DecodedBase<"mkt_crt"> & MarketCreatedFields) + | (DecodedBase<"bet_plc"> & BetPlacedFields) + | (DecodedBase<"bet_upd"> & BetUpdatedFields) + | (DecodedBase<"mkt_res"> & MarketResolvedFields) + | (DecodedBase<"win_clm"> & WinningsClaimedFields); + +/** A topic the decoder does not know. Forward compatible: no code change needed. */ +export interface UnknownEvent { + status: "unknown"; + /** `unknown:` */ + eventType: string; + marketId: string | null; + raw: RawXdr; +} + +/** A known topic whose topic layout or payload failed validation. */ +export interface InvalidEvent { + status: "invalid"; + /** `invalid:` */ + eventType: string; + marketId: string | null; + errors: string[]; + raw: RawXdr; +} + +export type DecodeResult = DecodedEvent | UnknownEvent | InvalidEvent; + +// ─── Validation primitives ─────────────────────────────────────────────────── + +const I128_MIN = -(BigInt(2) ** BigInt(127)); +const I128_MAX = BigInt(2) ** BigInt(127) - BigInt(1); +const U64_MAX = BigInt(2) ** BigInt(64) - BigInt(1); +const ZERO = BigInt(0); + +/** + * Field checks run once per payload inside a single transform per schema. + * (Per-field refine/transform wrappers made zod the dominant decode cost.) + * Each helper reports through the zod context, so failures surface as normal + * zod issues with the field path. + */ +function i128Of(value: bigint, path: string, ctx: z.RefinementCtx): string { + if (value < I128_MIN || value > I128_MAX) { + ctx.addIssue({ code: z.ZodIssueCode.custom, path: [path], message: "out of i128 range" }); + } + return value.toString(); +} + +function u64Of(value: bigint, path: string, ctx: z.RefinementCtx): string { + if (value < ZERO || value > U64_MAX) { + ctx.addIssue({ code: z.ZodIssueCode.custom, path: [path], message: "out of u64 range" }); + } + return value.toString(); +} + +/** `G…` account or `C…` contract strkey, checksum verified. */ +function addressOf(value: string, path: string, ctx: z.RefinementCtx): string { + if (!isValidAddress(value)) { + ctx.addIssue({ code: z.ZodIssueCode.custom, path: [path], message: "not a valid G… or C… strkey" }); + } + return value; +} + +/** + * Checksum results are memoised (bounded) because the same few addresses + * repeat constantly; the memo only caches a pure function of the string, so + * output is unchanged. + */ +const ADDRESS_MEMO_LIMIT = 4096; +const addressMemo = new Map(); +const strkeyMemo = new Map(); + +/** + * Encode raw address bytes as a strkey. The result is valid by construction, + * so it is also recorded as a validated address; both memos are bounded. + */ +function strkeyOf(kind: "G" | "C", bytes: Buffer): string { + const key = kind + bytes.toString("latin1"); + let encoded = strkeyMemo.get(key); + if (encoded === undefined) { + encoded = kind === "G" ? StrKey.encodeEd25519PublicKey(bytes) : StrKey.encodeContract(bytes); + if (strkeyMemo.size >= ADDRESS_MEMO_LIMIT) strkeyMemo.clear(); + strkeyMemo.set(key, encoded); + if (addressMemo.size >= ADDRESS_MEMO_LIMIT) addressMemo.clear(); + addressMemo.set(encoded, true); + } + return encoded; +} + +function isValidAddress(value: string): boolean { + const cached = addressMemo.get(value); + if (cached !== undefined) return cached; + const valid = StrKey.isValidEd25519PublicKey(value) || StrKey.isValidContract(value); + if (addressMemo.size >= ADDRESS_MEMO_LIMIT) addressMemo.clear(); + addressMemo.set(value, valid); + return valid; +} + +const text = z.string(); +const marketIdField = z.string().min(1); +const int = z.bigint(); + +// ─── Per-topic payload schemas (keys are the on-chain snake_case names) ────── + +const marketCreatedSchema = z + .object({ + market_id: marketIdField, + question: text, + outcomes: z.array(text), + admin: text, + end_time: int, + nonce: int, + timestamp: int, + }) + .transform((p, ctx) => ({ + marketId: p.market_id, + fields: { + question: p.question, + outcomes: p.outcomes, + admin: addressOf(p.admin, "admin", ctx), + endTime: u64Of(p.end_time, "end_time", ctx), + nonce: u64Of(p.nonce, "nonce", ctx), + timestamp: u64Of(p.timestamp, "timestamp", ctx), + } satisfies MarketCreatedFields, + })); + +const betPlacedSchema = z + .object({ + market_id: marketIdField, + bettor: text, + outcome: text, + amount: int, + nonce: int, + timestamp: int, + }) + .transform((p, ctx) => ({ + marketId: p.market_id, + fields: { + bettor: addressOf(p.bettor, "bettor", ctx), + outcome: p.outcome, + amount: i128Of(p.amount, "amount", ctx), + nonce: u64Of(p.nonce, "nonce", ctx), + timestamp: u64Of(p.timestamp, "timestamp", ctx), + } satisfies BetPlacedFields, + })); + +const betUpdatedSchema = z + .object({ + market_id: marketIdField, + bettor: text, + old_status: text, + new_status: text, + // Option: `None` is ScVal void, which scValToNative maps to null. + payout_amount: int.nullish(), + nonce: int, + timestamp: int, + }) + .transform((p, ctx) => ({ + marketId: p.market_id, + fields: { + bettor: addressOf(p.bettor, "bettor", ctx), + oldStatus: p.old_status, + newStatus: p.new_status, + payoutAmount: + p.payout_amount === null || p.payout_amount === undefined + ? null + : i128Of(p.payout_amount, "payout_amount", ctx), + nonce: u64Of(p.nonce, "nonce", ctx), + timestamp: u64Of(p.timestamp, "timestamp", ctx), + } satisfies BetUpdatedFields, + })); + +const marketResolvedSchema = z + .object({ + market_id: marketIdField, + final_outcome: text, + oracle_result: text, + community_consensus: text, + resolution_method: text, + confidence_score: int, + nonce: int, + timestamp: int, + }) + .transform((p, ctx) => ({ + marketId: p.market_id, + fields: { + resolutionMethod: p.resolution_method, + finalOutcome: p.final_outcome, + oracleResult: p.oracle_result, + communityConsensus: p.community_consensus, + confidenceScore: i128Of(p.confidence_score, "confidence_score", ctx), + nonce: u64Of(p.nonce, "nonce", ctx), + timestamp: u64Of(p.timestamp, "timestamp", ctx), + } satisfies MarketResolvedFields, + })); + +const winningsClaimedSchema = z + .object({ + market_id: marketIdField, + user: text, + amount: int, + nonce: int, + timestamp: int, + }) + .transform((p, ctx) => ({ + marketId: p.market_id, + fields: { + user: addressOf(p.user, "user", ctx), + amount: i128Of(p.amount, "amount", ctx), + nonce: u64Of(p.nonce, "nonce", ctx), + timestamp: u64Of(p.timestamp, "timestamp", ctx), + } satisfies WinningsClaimedFields, + })); + +const PAYLOAD_SCHEMAS: Record< + DecodedTopic, + z.ZodType<{ marketId: string; fields: object }, z.ZodTypeDef, unknown> +> = { + mkt_crt: marketCreatedSchema, + bet_plc: betPlacedSchema, + bet_upd: betUpdatedSchema, + mkt_res: marketResolvedSchema, + win_clm: winningsClaimedSchema, +}; + +// ─── Helpers ───────────────────────────────────────────────────────────────── + +function toScVal(input: xdr.ScVal | string): xdr.ScVal { + return typeof input === "string" ? xdr.ScVal.fromXDR(input, "base64") : input; +} + +function safeBase64(input: xdr.ScVal | string): string { + try { + return typeof input === "string" ? input : input.toXDR("base64"); + } catch { + return ""; + } +} + +function rawOf(raw: RawContractEvent): RawXdr { + return { + topics: Array.from(raw.topic ?? [], safeBase64), + value: raw.value === undefined ? "" : safeBase64(raw.value), + }; +} + +function symbolText(value: xdr.ScVal | undefined): string | null { + if (value === undefined || value.switch().name !== "scvSymbol") return null; + return value.sym().toString(); +} + +/** Symbol or String topic element as text; null for any other ScVal type. */ +function textOf(value: xdr.ScVal | undefined): string | null { + if (value === undefined) return null; + const kind = value.switch().name; + if (kind === "scvSymbol") return value.sym().toString(); + if (kind === "scvString") return value.str().toString(); + return null; +} + +const TWO_POW_64 = BigInt(1) << BigInt(64); + +/** + * scValToNative with fast paths for the two conversions that dominate its + * cost (Address and 128-bit integers). Every other ScVal type, and any address + * type other than account/contract, is delegated to scValToNative unchanged. + * Containers are walked here so the fast paths apply to struct fields. + */ +function nativeOf(value: xdr.ScVal): unknown { + switch (value.switch().name) { + case "scvMap": { + const out: Record = {}; + for (const entry of value.map() ?? []) { + out[nativeOf(entry.key()) as string] = nativeOf(entry.val()); + } + return out; + } + case "scvVec": + return (value.vec() ?? []).map(nativeOf); + case "scvSymbol": + return value.sym().toString(); + case "scvString": + return value.str().toString(); + case "scvU64": + return value.u64().toBigInt(); + case "scvI128": { + const parts = value.i128(); + return BigInt(parts.hi().toString()) * TWO_POW_64 + BigInt(parts.lo().toString()); + } + case "scvU128": { + const parts = value.u128(); + return BigInt(parts.hi().toString()) * TWO_POW_64 + BigInt(parts.lo().toString()); + } + case "scvAddress": { + const address = value.address(); + switch (address.switch().name) { + case "scAddressTypeAccount": + return strkeyOf("G", address.accountId().ed25519() as unknown as Buffer); + case "scAddressTypeContract": + return strkeyOf("C", address.contractId() as unknown as Buffer); + default: + return scValToNative(value); + } + } + default: + return scValToNative(value); + } +} + +function issuesOf(error: z.ZodError): string[] { + return error.issues.map((i) => `${i.path.join(".") || "(root)"}: ${i.message}`); +} + +function messageOf(err: unknown): string { + return err instanceof Error ? err.message : String(err); +} + +// ─── Decoder ───────────────────────────────────────────────────────────────── + +/** + * Decode one raw Predictify contract event. Pure: the same input always yields + * the same output and nothing is thrown. + */ +export function decodePredictifyEvent(raw: RawContractEvent): DecodeResult { + let topics: xdr.ScVal[]; + let value: xdr.ScVal; + try { + topics = Array.from(raw.topic, toScVal); + value = toScVal(raw.value); + } catch (err) { + return { + status: "invalid", + eventType: "invalid:topic", + marketId: null, + errors: [`malformed XDR: ${messageOf(err)}`], + raw: rawOf(raw), + }; + } + + const symbol = symbolText(topics[0]); + if (symbol === null) { + return { + status: "invalid", + eventType: "invalid:topic", + marketId: null, + errors: ["topic[0] is missing or is not a Symbol"], + raw: rawOf(raw), + }; + } + + const topicMarketId = symbolText(topics[1]); + const canonical: DecodedTopic | undefined = DECODED_TOPIC_SET.has(symbol) + ? (symbol as DecodedTopic) + : TOPIC_ALIAS_MAP.get(symbol); + + if (canonical === undefined) { + return { + status: "unknown", + eventType: `unknown:${symbol}`, + marketId: topicMarketId, + raw: rawOf(raw), + }; + } + + const invalid = (errors: string[]): InvalidEvent => ({ + status: "invalid", + eventType: `invalid:${canonical}`, + marketId: topicMarketId, + errors, + raw: rawOf(raw), + }); + + if (topicMarketId === null || topicMarketId === "") { + return invalid(["topic[1] (market_id) is missing or is not a Symbol"]); + } + + let native: unknown; + try { + native = nativeOf(value); + } catch (err) { + return invalid([`payload conversion failed: ${messageOf(err)}`]); + } + + const parsed = PAYLOAD_SCHEMAS[canonical].safeParse(native); + if (!parsed.success) { + return invalid(issuesOf(parsed.error)); + } + + const { marketId, fields } = parsed.data; + if (marketId !== topicMarketId) { + return invalid([`payload market_id "${marketId}" does not match topic[1] "${topicMarketId}"`]); + } + + if (canonical === "mkt_res") { + const topicMethod = textOf(topics[2]); + const payloadMethod = (fields as MarketResolvedFields).resolutionMethod; + if (topicMethod === null) { + return invalid(["topic[2] (resolution_method) is missing or is not a String/Symbol"]); + } + if (topicMethod !== payloadMethod) { + return invalid([ + `payload resolution_method "${payloadMethod}" does not match topic[2] "${topicMethod}"`, + ]); + } + } + + return { status: "decoded", eventType: canonical, marketId, ...fields } as DecodedEvent; +} + +/** The JSON-safe object persisted in `indexer_events.data` for a decode result. */ +export function eventDataOf(result: DecodeResult): Record { + if (result.status === "decoded") { + const fields: Record = { ...result }; + delete fields.status; + delete fields.eventType; + delete fields.marketId; + return fields; + } + if (result.status === "invalid") { + return { raw: result.raw, errors: result.errors }; + } + return { raw: result.raw }; +} diff --git a/src/metrics/registry.ts b/src/metrics/registry.ts index 26941667..449006c1 100644 --- a/src/metrics/registry.ts +++ b/src/metrics/registry.ts @@ -56,6 +56,20 @@ export const indexerPollsTotal = new Counter({ registers: [register], }); +export const indexerEventsDecodedTotal = new Counter({ + name: "indexer_events_decoded_total", + help: "Contract events decoded by the indexer, by topic symbol (unknown topics are grouped as 'unknown')", + labelNames: ["event_type"] as const, + registers: [register], +}); + +export const indexerEventsInvalidTotal = new Counter({ + name: "indexer_events_invalid_total", + help: "Contract events whose topic layout or payload failed validation, by topic symbol", + labelNames: ["event_type"] as const, + registers: [register], +}); + export const indexerLagLedgers = new Gauge({ name: "indexer_lag_ledgers", help: "Current lag in ledgers between the indexer cursor and the chain tip", diff --git a/src/services/indexerService.ts b/src/services/indexerService.ts index be1670f0..2acc8738 100644 --- a/src/services/indexerService.ts +++ b/src/services/indexerService.ts @@ -1,4 +1,4 @@ -import { rpc } from "@stellar/stellar-sdk"; +import { rpc, type xdr } from "@stellar/stellar-sdk"; import { env } from "../config/env"; import { logger } from "../config/logger"; import { getPool } from "../db/client"; @@ -9,8 +9,11 @@ import { type DecodedEventIdentity, type RecoveryEvent, } from "./indexerRecovery"; +import { decodePredictifyEvent, eventDataOf } from "../indexer/eventDecoder"; import { + indexerEventsDecodedTotal, indexerEventsInsertedTotal, + indexerEventsInvalidTotal, indexerReorgsDetectedTotal, } from "../metrics/registry"; @@ -31,6 +34,7 @@ export interface IndexerEventInput { opIndex: number; eventIndex?: number; eventType?: string; + /** Market the event is scoped to (set for decoded market-scoped events). */ marketId?: string | null; payload?: unknown; } @@ -63,6 +67,57 @@ export function extractMarketId(event: IndexerEventInput): string | null { return null; } +/** The subset of an RPC `EventResponse` that mapRpcEvent reads. */ +export interface RpcEventLike { + id: string; + type: string; + ledger: number; + txHash: string; + contractId?: { contractId(): string } | null; + topic: ReadonlyArray; + value: xdr.ScVal; +} + +/** + * Turn one RPC event into an indexer row, or null when it must not be + * persisted (not a `contract` event, or emitted by another contract). + * Unknown and invalid topics are kept with their raw XDR; nothing throws. + */ +export function mapRpcEvent( + event: RpcEventLike, + contractId: string, + fallbackIndex: number, +): IndexerEventInput | null { + if (event.type !== "contract") { + return null; + } + if (event.contractId && event.contractId.contractId() !== contractId) { + return null; + } + + const decoded = decodePredictifyEvent({ topic: event.topic, value: event.value }); + if (decoded.status === "decoded") { + indexerEventsDecodedTotal.inc({ event_type: decoded.eventType }); + } else if (decoded.status === "unknown") { + indexerEventsDecodedTotal.inc({ event_type: "unknown" }); + } else { + indexerEventsInvalidTotal.inc({ event_type: decoded.eventType.slice("invalid:".length) }); + logger.warn( + { txHash: event.txHash, ledger: event.ledger, eventType: decoded.eventType, errors: decoded.errors }, + "indexer: contract event failed validation; stored as invalid", + ); + } + + return { + ledger: event.ledger, + txHash: event.txHash, + opIndex: parseEventOpIndex(event.id, fallbackIndex), + eventType: decoded.eventType, + marketId: decoded.marketId, + payload: eventDataOf(decoded), + }; +} + export function createSorobanRpcClient(): SorobanRpcClient { const server = new rpc.Server(env.SOROBAN_RPC_URL); @@ -98,24 +153,24 @@ export function createSorobanRpcClient(): SorobanRpcClient { return collected; } if (event.ledger >= startLedger) { - const decoded = decodeEventId(event.id); - if (!decoded) { + const decodedId = decodeEventId(event.id); + if (!decodedId) { logger.warn( { eventId: event.id, txHash: event.txHash, ledger: event.ledger }, "indexer skipping event with malformed event id", ); continue; } - collected.push({ - id: event.id, - ledger: event.ledger, - txHash: event.txHash, - txApplicationOrder: decoded.txApplicationOrder, - opIndex: decoded.opIndex, - eventIndex: decoded.eventIndex, - eventType: event.type, - payload: event.value, - }); + const mapped = mapRpcEvent(event, env.PREDICTIFY_CONTRACT_ID, decodedId.opIndex); + if (mapped) { + collected.push({ + ...mapped, + id: event.id, + txApplicationOrder: decodedId.txApplicationOrder, + opIndex: decodedId.opIndex, + eventIndex: decodedId.eventIndex, + }); + } } } diff --git a/src/workers/predictionsConfirmer.ts b/src/workers/predictionsConfirmer.ts index c9b9ebf4..598ce53d 100644 --- a/src/workers/predictionsConfirmer.ts +++ b/src/workers/predictionsConfirmer.ts @@ -2,14 +2,17 @@ * predictionsConfirmer.ts * * Background worker that transitions pending predictions to confirmed (or - * failed) by joining against indexer_events on txHash. + * failed) by joining against indexer_events. * - * After the indexer ingests a bet_placed event, the corresponding pending row + * After the indexer ingests a bet_plc event, the corresponding pending row * in predictions must transition to confirmed. This worker: * * 1. Polls on a configurable interval (PREDICTION_CONFIRM_INTERVAL_MS). - * 2. Finds pending predictions where a matching indexer_event exists (same - * txHash) and batch-updates them to "confirmed". + * 2. Finds pending predictions where a matching indexer_event exists and + * batch-updates them to "confirmed". An event matches only when it is a + * canonical `bet_plc` event in the prediction's tx whose market_id equals + * the prediction's market and whose bettor equals the user's address + * (see `confirmingEventExists`). Unknown/invalid/other events never count. * 3. For unmatched pending predictions, increments `confirmAttempts`. After * MAX_CONFIRM_ATTEMPTS (default 3) the row transitions to "failed" with a * descriptive lastError. @@ -26,7 +29,7 @@ * • The webhook emitter is injected as a constructor dependency so unit * tests can mock the webhook path without module-level imports. * • The repo uses DISTINCT when joining indexer_events to guard against - * duplicate prediction IDs when a txHash has multiple event rows. + * duplicate prediction IDs when a txHash has multiple bet_plc rows. * • The worker is structured as an injectable service class (for unit tests) * and a standalone main() entry point (for production). */ @@ -37,7 +40,7 @@ import { drizzle } from "drizzle-orm/node-postgres"; import { env } from "../config/env"; import { logger } from "../config/logger"; import { pool } from "../db/client"; -import { predictions, indexerEvents, markets } from "../db/schema"; +import { predictions, indexerEvents, markets, users } from "../db/schema"; import type { Db } from "../db"; // ─── Constants ─────────────────────────────────────────────────────────────── @@ -81,8 +84,9 @@ export const defaultWebhookEmitter: WebhookEmitter = async ( */ export interface PredictionsConfirmerRepo { /** - * Returns the unique IDs of pending predictions that have a matching txHash - * in indexer_events. These should be transitioned to "confirmed". + * Returns the unique IDs of pending predictions that have a matching + * canonical bet_plc event in indexer_events (same txHash, market and + * bettor). These should be transitioned to "confirmed". * Uses DISTINCT to avoid duplicates when a txHash has multiple event rows. */ findConfirmedCandidateIds(batchSize: number): Promise; @@ -264,6 +268,30 @@ export class PredictionsConfirmerService { // ─── Drizzle repository ──────────────────────────────────────────────────── +/** Topic symbol of the bet-placed event (events.rs: `bet_plc`). */ +export const BET_PLACED_EVENT_TYPE = "bet_plc"; + +/** + * Predicate that is true when `indexer_events` holds the on-chain event + * confirming a prediction: a canonical `bet_plc` row in the prediction's + * transaction, for the prediction's market, placed by the prediction's user. + * The bettor is compared with the user's Stellar address (`data->>'bettor'` + * is a strkey string written by the event decoder). + */ +function confirmingEventExists() { + return sql`EXISTS ( + SELECT 1 FROM ${indexerEvents} + WHERE ${indexerEvents.txHash} = ${predictions.txHash} + AND ${indexerEvents.eventType} = ${BET_PLACED_EVENT_TYPE} + AND ${indexerEvents.canonical} = TRUE + AND ${indexerEvents.marketId} = ${predictions.marketId} + AND ${indexerEvents.data}->>'bettor' = ( + SELECT ${users.stellarAddress} FROM ${users} + WHERE ${users.id} = ${predictions.userId} + ) + )`; +} + /** * Production implementation of PredictionsConfirmerRepo backed by Drizzle ORM. * All three batch mutations run inside a single Drizzle transaction so partial @@ -277,19 +305,16 @@ export class DrizzlePredictionsConfirmerRepo implements PredictionsConfirmerRepo } async findConfirmedCandidateIds(batchSize: number): Promise { - // DISTINCT ensures a prediction ID appears at most once even when its - // txHash has multiple indexer_event rows (different op_index values). + // Only a canonical bet_plc event for this prediction's market and bettor + // confirms it; unknown:*, invalid:* and other topics in the same tx do not. const rows = await this.db .selectDistinct({ id: predictions.id }) .from(predictions) - .innerJoin( - indexerEvents, - eq(predictions.txHash, indexerEvents.txHash), - ) .where( and( eq(predictions.status, "pending"), sql`${predictions.txHash} != ''`, + confirmingEventExists(), ), ) .limit(batchSize); @@ -348,10 +373,7 @@ export class DrizzlePredictionsConfirmerRepo implements PredictionsConfirmerRepo eq(predictions.status, "pending"), sql`${predictions.txHash} != ''`, sql`${predictions.confirmAttempts} < ${maxAttempts}`, - sql`NOT EXISTS ( - SELECT 1 FROM ${indexerEvents} - WHERE ${indexerEvents.txHash} = ${predictions.txHash} - )`, + sql`NOT ${confirmingEventExists()}`, ), ); @@ -366,17 +388,14 @@ export class DrizzlePredictionsConfirmerRepo implements PredictionsConfirmerRepo .update(predictions) .set({ status: "failed", - lastError: `No matching indexer event found after ${maxAttempts} attempts`, + lastError: `No matching bet_plc indexer event found after ${maxAttempts} attempts`, }) .where( and( eq(predictions.status, "pending"), sql`${predictions.txHash} != ''`, sql`${predictions.confirmAttempts} >= ${maxAttempts}`, - sql`NOT EXISTS ( - SELECT 1 FROM ${indexerEvents} - WHERE ${indexerEvents.txHash} = ${predictions.txHash} - )`, + sql`NOT ${confirmingEventExists()}`, ), ) .returning({ id: predictions.id }); diff --git a/tests/indexerEventDecoder.test.ts b/tests/indexerEventDecoder.test.ts new file mode 100644 index 00000000..9164733e --- /dev/null +++ b/tests/indexerEventDecoder.test.ts @@ -0,0 +1,711 @@ +/** + * Tests for src/indexer/eventDecoder.ts and its use in the RPC client + * (issue #979). Fixtures are built with nativeToScVal / xdr using the exact + * topic and payload layouts of contracts/predictify-hybrid/src/events.rs. + */ + +import { Keypair, StrKey, nativeToScVal, xdr } from "@stellar/stellar-sdk"; + +import { + DECODED_TOPICS, + TOPIC_ALIASES, + decodePredictifyEvent, + eventDataOf, + type RawContractEvent, +} from "../src/indexer/eventDecoder"; +import { + IndexerService, + createSorobanRpcClient, + mapRpcEvent, + type RpcEventLike, +} from "../src/services/indexerService"; +import { + indexerEventsDecodedTotal, + indexerEventsInvalidTotal, + register, +} from "../src/metrics/registry"; + +const mockQuery = jest.fn(); +jest.mock("../src/db/client", () => ({ + getPool: () => ({ query: mockQuery }), +})); + +const mockGetEvents: jest.Mock = jest.fn(); +jest.mock("@stellar/stellar-sdk", () => { + const actual = jest.requireActual("@stellar/stellar-sdk"); + return { + ...actual, + rpc: { + ...actual.rpc, + Server: jest.fn().mockImplementation(() => ({ + getEvents: (...args: unknown[]) => mockGetEvents(...args), + getLatestLedger: jest.fn().mockResolvedValue({ sequence: 1 }), + })), + }, + }; +}); + +// ─── Fixture helpers ───────────────────────────────────────────────────────── + +const I128_MAX = "170141183460469231731687303715884105727"; +const MARKET = "mkt_5cae97f9_0"; +const BETTOR = Keypair.fromRawEd25519Seed(Buffer.alloc(32, 1)).publicKey(); +const ADMIN = Keypair.fromRawEd25519Seed(Buffer.alloc(32, 2)).publicKey(); +const CONTRACT_ADDR = StrKey.encodeContract(Buffer.alloc(32, 3)); + +type FieldType = [string, string]; + +function struct(values: Record, types: Record): xdr.ScVal { + const typeMap: Record = {}; + for (const key of Object.keys(types)) typeMap[key] = ["symbol", types[key]]; + return nativeToScVal(values, { type: typeMap }); +} + +const sym = (s: string): xdr.ScVal => xdr.ScVal.scvSymbol(s); +const str = (s: string): xdr.ScVal => nativeToScVal(s, { type: "string" }); + +function betPlaced(overrides: Record = {}, drop: string[] = []): xdr.ScVal { + const values: Record = { + market_id: MARKET, + bettor: BETTOR, + outcome: "Yes", + amount: BigInt(25_000_000), + nonce: BigInt(7), + timestamp: BigInt(1_792_000_000), + ...overrides, + }; + for (const key of drop) delete values[key]; + return struct(values, { + market_id: "symbol", + bettor: "address", + outcome: "string", + amount: "i128", + nonce: "u64", + timestamp: "u64", + }); +} + +function winningsClaimed(amount: bigint): xdr.ScVal { + return struct( + { market_id: MARKET, user: BETTOR, amount, nonce: BigInt(1), timestamp: BigInt(2) }, + { market_id: "symbol", user: "address", amount: "i128", nonce: "u64", timestamp: "u64" }, + ); +} + +function marketResolved(method = "oracle", marketId = MARKET): xdr.ScVal { + return struct( + { + market_id: marketId, + final_outcome: "Yes", + oracle_result: "Yes", + community_consensus: "Yes", + resolution_method: method, + confidence_score: BigInt(95), + nonce: BigInt(3), + timestamp: BigInt(4), + }, + { + market_id: "symbol", + final_outcome: "string", + oracle_result: "string", + community_consensus: "string", + resolution_method: "string", + confidence_score: "i128", + nonce: "u64", + timestamp: "u64", + }, + ); +} + +function marketCreated(marketId = MARKET): xdr.ScVal { + return struct( + { + market_id: marketId, + question: "Will BTC close above 100k?", + outcomes: xdr.ScVal.scvVec([str("Yes"), str("No")]), + admin: ADMIN, + end_time: BigInt(1_793_000_000), + nonce: BigInt(1), + timestamp: BigInt(1_792_000_000), + }, + { + market_id: "symbol", + question: "string", + outcomes: "vec", + admin: "address", + end_time: "u64", + nonce: "u64", + timestamp: "u64", + }, + ); +} + +function betUpdated(payout: bigint | null): xdr.ScVal { + const value = struct( + { + market_id: MARKET, + bettor: BETTOR, + old_status: "Active", + new_status: "Won", + payout_amount: payout === null ? null : payout, + nonce: BigInt(5), + timestamp: BigInt(6), + }, + { + market_id: "symbol", + bettor: "address", + old_status: "string", + new_status: "string", + payout_amount: "i128", + nonce: "u64", + timestamp: "u64", + }, + ); + return value; +} + +const event = (topic: xdr.ScVal[], value: xdr.ScVal): RawContractEvent => ({ topic, value }); + +// ─── Test plan rows 1-6 ────────────────────────────────────────────────────── + +describe("decodePredictifyEvent", () => { + it("row 1: decodes a valid bet_plc payload", () => { + const result = decodePredictifyEvent(event([sym("bet_plc"), sym(MARKET)], betPlaced())); + expect(result).toMatchObject({ + status: "decoded", + eventType: "bet_plc", + marketId: MARKET, + bettor: BETTOR, + outcome: "Yes", + amount: "25000000", + nonce: "7", + timestamp: "1792000000", + }); + }); + + it("row 2: decodes win_clm with amount = i128::MAX exactly (no float rounding)", () => { + const result = decodePredictifyEvent( + event([sym("win_clm"), sym(MARKET)], winningsClaimed(BigInt(I128_MAX))), + ); + expect(result).toMatchObject({ status: "decoded", eventType: "win_clm", user: BETTOR }); + expect((result as { amount: string }).amount).toBe( + "170141183460469231731687303715884105727", + ); + expect(typeof (result as { amount: unknown }).amount).toBe("string"); + }); + + it("decodes a negative i128 as an exact decimal string", () => { + const result = decodePredictifyEvent( + event([sym("win_clm"), sym(MARKET)], winningsClaimed(-BigInt(I128_MAX) - BigInt(1))), + ); + expect((result as { amount: string }).amount).toBe( + "-170141183460469231731687303715884105728", + ); + }); + + it("row 3: extracts marketId and resolutionMethod from a three-element mkt_res topic", () => { + const result = decodePredictifyEvent( + event([sym("mkt_res"), sym(MARKET), str("oracle")], marketResolved("oracle")), + ); + expect(result).toMatchObject({ + status: "decoded", + eventType: "mkt_res", + marketId: MARKET, + resolutionMethod: "oracle", + confidenceScore: "95", + }); + }); + + it("accepts a Symbol-typed resolution method in topic[2]", () => { + const result = decodePredictifyEvent( + event([sym("mkt_res"), sym(MARKET), sym("manual")], marketResolved("manual")), + ); + expect(result).toMatchObject({ status: "decoded", resolutionMethod: "manual" }); + }); + + it("row 4: unknown topic foo_bar becomes unknown:foo_bar and keeps raw XDR", () => { + const payload = nativeToScVal(42, { type: "u32" }); + const result = decodePredictifyEvent(event([sym("foo_bar"), sym(MARKET)], payload)); + expect(result).toMatchObject({ + status: "unknown", + eventType: "unknown:foo_bar", + marketId: MARKET, + }); + const data = eventDataOf(result) as { raw: { topics: string[]; value: string } }; + expect(data.raw.topics).toEqual([sym("foo_bar").toXDR("base64"), sym(MARKET).toXDR("base64")]); + expect(data.raw.value).toBe(payload.toXDR("base64")); + }); + + it("unknown topic without a market topic has a null marketId (e.g. stats_upd)", () => { + const result = decodePredictifyEvent(event([sym("stats_upd")], xdr.ScVal.scvVoid())); + expect(result).toMatchObject({ status: "unknown", eventType: "unknown:stats_upd", marketId: null }); + }); + + it("row 5: bet_plc with the amount field missing is invalid:bet_plc", () => { + const result = decodePredictifyEvent( + event([sym("bet_plc"), sym(MARKET)], betPlaced({}, ["amount"])), + ); + expect(result).toMatchObject({ status: "invalid", eventType: "invalid:bet_plc", marketId: MARKET }); + expect((result as { errors: string[] }).errors.join(" ")).toContain("amount"); + expect((eventDataOf(result) as { raw: { value: string } }).raw.value).toEqual( + betPlaced({}, ["amount"]).toXDR("base64"), + ); + }); + + it("row 5 (batch continues): decoding an invalid event never throws and does not affect the next", () => { + const batch = [ + event([sym("bet_plc"), sym(MARKET)], betPlaced({}, ["amount"])), + event([sym("bet_plc"), sym(MARKET)], betPlaced()), + ]; + const results = batch.map(decodePredictifyEvent); + expect(results.map((r) => r.eventType)).toEqual(["invalid:bet_plc", "bet_plc"]); + }); + + it("row 6: long-name alias bet_placed decodes the same as bet_plc", () => { + const canonical = decodePredictifyEvent(event([sym("bet_plc"), sym(MARKET)], betPlaced())); + const alias = decodePredictifyEvent(event([sym("bet_placed"), sym(MARKET)], betPlaced())); + expect(alias).toEqual(canonical); + }); + + it("every catalog alias normalises to its emitted symbol", () => { + expect(TOPIC_ALIASES).toEqual({ + market_created: "mkt_crt", + bet_placed: "bet_plc", + market_resolved: "mkt_res", + winnings_claimed: "win_clm", + }); + for (const target of Object.values(TOPIC_ALIASES)) { + expect(DECODED_TOPICS).toContain(target); + } + const created = decodePredictifyEvent(event([sym("market_created"), sym(MARKET)], marketCreated())); + expect(created.eventType).toBe("mkt_crt"); + const resolved = decodePredictifyEvent( + event([sym("market_resolved"), sym(MARKET), str("oracle")], marketResolved()), + ); + expect(resolved.eventType).toBe("mkt_res"); + const claimed = decodePredictifyEvent( + event([sym("winnings_claimed"), sym(MARKET)], winningsClaimed(BigInt(1))), + ); + expect(claimed.eventType).toBe("win_clm"); + }); + + it("treats Object.prototype names as unknown topics, not aliases", () => { + const result = decodePredictifyEvent(event([sym("constructor"), sym(MARKET)], xdr.ScVal.scvVoid())); + expect(result).toMatchObject({ status: "unknown", eventType: "unknown:constructor" }); + }); + + it("decodes mkt_crt", () => { + const result = decodePredictifyEvent(event([sym("mkt_crt"), sym(MARKET)], marketCreated())); + expect(result).toMatchObject({ + status: "decoded", + eventType: "mkt_crt", + marketId: MARKET, + question: "Will BTC close above 100k?", + outcomes: ["Yes", "No"], + admin: ADMIN, + endTime: "1793000000", + }); + }); + + it("decodes bet_upd with and without a payout (Option)", () => { + const won = decodePredictifyEvent(event([sym("bet_upd"), sym(MARKET)], betUpdated(BigInt(15_000_000)))); + expect(won).toMatchObject({ eventType: "bet_upd", oldStatus: "Active", newStatus: "Won", payoutAmount: "15000000" }); + const none = decodePredictifyEvent(event([sym("bet_upd"), sym(MARKET)], betUpdated(null))); + expect(none).toMatchObject({ status: "decoded", payoutAmount: null }); + }); + + it("accepts contract (C...) strkeys as addresses", () => { + const result = decodePredictifyEvent( + event([sym("bet_plc"), sym(MARKET)], betPlaced({ bettor: CONTRACT_ADDR })), + ); + expect(result).toMatchObject({ status: "decoded", bettor: CONTRACT_ADDR }); + }); + + it("accepts base64-encoded XDR input as returned by the RPC", () => { + const result = decodePredictifyEvent({ + topic: [sym("bet_plc").toXDR("base64"), sym(MARKET).toXDR("base64")], + value: betPlaced().toXDR("base64"), + }); + expect(result).toMatchObject({ status: "decoded", eventType: "bet_plc", amount: "25000000" }); + }); + + describe("invalid events", () => { + it("rejects a payload whose market_id disagrees with topic[1]", () => { + const result = decodePredictifyEvent( + event([sym("bet_plc"), sym("other_mkt")], betPlaced()), + ); + expect(result).toMatchObject({ status: "invalid", eventType: "invalid:bet_plc", marketId: "other_mkt" }); + expect((result as { errors: string[] }).errors[0]).toContain("does not match topic[1]"); + }); + + it("rejects a known topic with no market topic", () => { + const result = decodePredictifyEvent(event([sym("bet_plc")], betPlaced())); + expect(result).toMatchObject({ status: "invalid", eventType: "invalid:bet_plc", marketId: null }); + }); + + it("rejects a mkt_res whose topic[2] is missing or of the wrong type", () => { + const missing = decodePredictifyEvent(event([sym("mkt_res"), sym(MARKET)], marketResolved())); + expect(missing).toMatchObject({ status: "invalid", eventType: "invalid:mkt_res" }); + const wrongType = decodePredictifyEvent( + event([sym("mkt_res"), sym(MARKET), nativeToScVal(1, { type: "u32" })], marketResolved()), + ); + expect(wrongType).toMatchObject({ status: "invalid", eventType: "invalid:mkt_res" }); + }); + + it("rejects a mkt_res whose topic[2] disagrees with the payload", () => { + const result = decodePredictifyEvent( + event([sym("mkt_res"), sym(MARKET), str("manual")], marketResolved("oracle")), + ); + expect(result).toMatchObject({ status: "invalid", eventType: "invalid:mkt_res" }); + expect((result as { errors: string[] }).errors[0]).toContain("resolution_method"); + }); + + it("rejects a bettor string that is not a strkey (payload carries a Symbol)", () => { + const value = struct( + { + market_id: MARKET, + bettor: "GNOTAREALADDRESS", + outcome: "Yes", + amount: BigInt(1), + nonce: BigInt(1), + timestamp: BigInt(1), + }, + { + market_id: "symbol", + bettor: "symbol", + outcome: "string", + amount: "i128", + nonce: "u64", + timestamp: "u64", + }, + ); + const result = decodePredictifyEvent(event([sym("bet_plc"), sym(MARKET)], value)); + expect(result).toMatchObject({ status: "invalid", eventType: "invalid:bet_plc" }); + expect((result as { errors: string[] }).errors.join(" ")).toContain("strkey"); + }); + + it("rejects amount of the wrong type (u32 instead of i128)", () => { + const value = struct( + { + market_id: MARKET, + bettor: BETTOR, + outcome: "Yes", + amount: 5, + nonce: BigInt(1), + timestamp: BigInt(1), + }, + { + market_id: "symbol", + bettor: "address", + outcome: "string", + amount: "u32", + nonce: "u64", + timestamp: "u64", + }, + ); + const bad = decodePredictifyEvent(event([sym("bet_plc"), sym(MARKET)], value)); + expect(bad).toMatchObject({ status: "invalid", eventType: "invalid:bet_plc" }); + }); + + it("rejects an amount outside the i128 range (u128 above i128::MAX)", () => { + const value = struct( + { + market_id: MARKET, + user: BETTOR, + amount: BigInt(I128_MAX) + BigInt(1), + nonce: BigInt(1), + timestamp: BigInt(1), + }, + { market_id: "symbol", user: "address", amount: "u128", nonce: "u64", timestamp: "u64" }, + ); + const result = decodePredictifyEvent(event([sym("win_clm"), sym(MARKET)], value)); + expect(result).toMatchObject({ status: "invalid", eventType: "invalid:win_clm" }); + expect((result as { errors: string[] }).errors.join(" ")).toContain("out of i128 range"); + }); + + it("rejects a negative nonce (i64 where u64 is required)", () => { + const value = struct( + { market_id: MARKET, user: BETTOR, amount: BigInt(1), nonce: -BigInt(1), timestamp: BigInt(1) }, + { market_id: "symbol", user: "address", amount: "i128", nonce: "i64", timestamp: "u64" }, + ); + const result = decodePredictifyEvent(event([sym("win_clm"), sym(MARKET)], value)); + expect(result).toMatchObject({ status: "invalid", eventType: "invalid:win_clm" }); + expect((result as { errors: string[] }).errors.join(" ")).toContain("out of u64 range"); + }); + + it("rejects an address type that is neither an account nor a contract", () => { + // The installed SDK only knows account and contract addresses, so a future + // address kind is simulated with a stand-in ScVal. + const exotic = { + switch: () => ({ name: "scvAddress", value: 18 }), + address: () => ({ switch: () => ({ name: "scAddressTypeFuture" }) }), + } as unknown as xdr.ScVal; + const value = xdr.ScVal.scvMap([ + new xdr.ScMapEntry({ key: sym("market_id"), val: sym(MARKET) }), + new xdr.ScMapEntry({ key: sym("user"), val: exotic }), + ]); + const result = decodePredictifyEvent(event([sym("win_clm"), sym(MARKET)], value)); + expect(result.status).toBe("invalid"); + }); + + it("never throws even when the payload object itself misbehaves", () => { + const hostile = { + switch: () => { + throw new Error("boom"); + }, + toXDR: () => { + throw new Error("no xdr"); + }, + } as unknown as xdr.ScVal; + const result = decodePredictifyEvent(event([sym("bet_plc"), sym(MARKET)], hostile)); + expect(result).toMatchObject({ status: "invalid", eventType: "invalid:bet_plc" }); + expect((result as { errors: string[] }).errors[0]).toContain("payload conversion failed"); + expect(eventDataOf(result)).toMatchObject({ raw: { value: "" } }); + }); + + it("rejects a payload that is not a map", () => { + const result = decodePredictifyEvent(event([sym("win_clm"), sym(MARKET)], xdr.ScVal.scvVoid())); + expect(result).toMatchObject({ status: "invalid", eventType: "invalid:win_clm" }); + }); + + it("handles an empty topic list", () => { + const result = decodePredictifyEvent(event([], xdr.ScVal.scvVoid())); + expect(result).toMatchObject({ status: "invalid", eventType: "invalid:topic", marketId: null }); + }); + + it("handles topic[0] that is not a Symbol", () => { + const result = decodePredictifyEvent(event([str("bet_plc")], xdr.ScVal.scvVoid())); + expect(result).toMatchObject({ status: "invalid", eventType: "invalid:topic" }); + }); + + it("handles malformed base64 XDR without throwing", () => { + const result = decodePredictifyEvent({ topic: ["not-xdr!!"], value: "also-bad" }); + expect(result).toMatchObject({ status: "invalid", eventType: "invalid:topic" }); + expect(eventDataOf(result)).toMatchObject({ raw: { topics: ["not-xdr!!"], value: "also-bad" } }); + }); + + it("handles a payload ScVal that cannot be converted to a native value", () => { + const result = decodePredictifyEvent( + event([sym("win_clm"), sym(MARKET)], xdr.ScVal.scvLedgerKeyContractInstance()), + ); + expect(result.status).toBe("invalid"); + }); + }); + + describe("properties", () => { + it("is deterministic and does not mutate its input", () => { + const raw = event([sym("bet_plc"), sym(MARKET)], betPlaced()); + const before = raw.value instanceof Object ? (raw.value as xdr.ScVal).toXDR("base64") : ""; + const a = decodePredictifyEvent(raw); + const b = decodePredictifyEvent(raw); + expect(a).toEqual(b); + expect((raw.value as xdr.ScVal).toXDR("base64")).toBe(before); + }); + + it("produces JSON-safe data (no bigint) for every status", () => { + const results = [ + decodePredictifyEvent(event([sym("win_clm"), sym(MARKET)], winningsClaimed(BigInt(I128_MAX)))), + decodePredictifyEvent(event([sym("foo_bar")], xdr.ScVal.scvVoid())), + decodePredictifyEvent(event([sym("bet_plc"), sym(MARKET)], betPlaced({}, ["amount"]))), + ]; + for (const r of results) { + expect(() => JSON.stringify(eventDataOf(r))).not.toThrow(); + } + expect(eventDataOf(results[0])).toEqual({ + user: BETTOR, + amount: I128_MAX, + nonce: "1", + timestamp: "2", + }); + }); + + it("benchmark: decodes 10,000 events in under 500 ms", () => { + const events = [ + event([sym("bet_plc"), sym(MARKET)], betPlaced()), + event([sym("win_clm"), sym(MARKET)], winningsClaimed(BigInt(I128_MAX))), + event([sym("mkt_res"), sym(MARKET), str("oracle")], marketResolved()), + event([sym("mkt_crt"), sym(MARKET)], marketCreated()), + event([sym("foo_bar")], xdr.ScVal.scvVoid()), + ]; + const N = 10_000; + decodePredictifyEvent(events[0]); // warm up + // Measured as process CPU time (user + system), best of 3 full runs, so a + // busy host stealing cycles does not inflate the figure; wall time is + // logged alongside it. + const runs: number[] = []; + const walls: number[] = []; + for (let run = 0; run < 3; run += 1) { + const cpuStart = process.cpuUsage(); + const wallStart = process.hrtime.bigint(); + for (let i = 0; i < N; i += 1) { + decodePredictifyEvent(events[i % events.length]); + } + const cpu = process.cpuUsage(cpuStart); + walls.push(Number(process.hrtime.bigint() - wallStart) / 1e6); + runs.push((cpu.user + cpu.system) / 1000); + } + const ms = Math.min(...runs); + // eslint-disable-next-line no-console + console.log( + `BENCH decodePredictifyEvent x${N}: best cpu ${ms.toFixed(1)} ms (cpu runs: ${runs.map((r) => r.toFixed(1)).join(", ")}; wall runs: ${walls.map((r) => r.toFixed(1)).join(", ")})`, + ); + // Coverage instrumentation (jest --coverage) slows this loop several-fold, + // so the 500 ms budget is enforced on uninstrumented runs and a loose + // sanity bound applies to instrumented ones. + const instrumented = Boolean((globalThis as { __coverage__?: unknown }).__coverage__); + expect(ms).toBeLessThan(instrumented ? 2500 : 500); + }); + }); +}); + +// ─── Row 10: testnet-shaped fixture ────────────────────────────────────────── + +describe("row 10: testnet fixture for mkt_5cae97f9_0", () => { + /** + * NOTE: This fixture is NOT a captured RPC response. The testnet RPC + * (soroban-testnet.stellar.org) returned zero events for contract + * CCWRJXIUBSMFQ4AF2RXXH3AFBW6MEZG4GK4IYES7C4Z3X4PR6K7VAFHC inside its + * retention window (ledgers 5155054-5276013) on 2026-10-19, so the mkt_crt + * event was built from the events.rs layout and serialised to base64 exactly + * as getEvents returns topic/value (see mapRpcEvent for the same shape). + */ + const rpcShaped = { + topic: [sym("mkt_crt").toXDR("base64"), sym("mkt_5cae97f9_0").toXDR("base64")], + value: marketCreated("mkt_5cae97f9_0").toXDR("base64"), + }; + + it("decodes into a mkt_crt for mkt_5cae97f9_0", () => { + const result = decodePredictifyEvent(rpcShaped); + expect(result).toMatchObject({ + status: "decoded", + eventType: "mkt_crt", + marketId: "mkt_5cae97f9_0", + }); + }); +}); + +// ─── Rows 7 + RPC client / persistence wiring ──────────────────────────────── + +describe("RPC client integration", () => { + const CONTRACT_ID = process.env.PREDICTIFY_CONTRACT_ID as string; + const ownContract = { contractId: () => CONTRACT_ID }; + + function rpcEvent(overrides: Partial = {}): RpcEventLike { + return { + id: "0000000001-0000000002", + type: "contract", + ledger: 100, + txHash: "tx-100", + contractId: ownContract, + topic: [sym("bet_plc"), sym(MARKET)], + value: betPlaced(), + ...overrides, + }; + } + + async function counterValue( + counter: typeof indexerEventsDecodedTotal | typeof indexerEventsInvalidTotal, + label: string, + ): Promise { + const metric = await counter.get(); + return metric.values.find((v) => v.labels.event_type === label)?.value ?? 0; + } + + beforeEach(() => { + mockGetEvents.mockReset(); + mockQuery.mockReset(); + }); + + it("row 7: system and diagnostic events are filtered out before persistence", () => { + expect(mapRpcEvent(rpcEvent({ type: "system" }), CONTRACT_ID, 0)).toBeNull(); + expect(mapRpcEvent(rpcEvent({ type: "diagnostic" }), CONTRACT_ID, 0)).toBeNull(); + }); + + it("filters events emitted by a different contract", () => { + const other = { contractId: () => "COTHERCONTRACT" }; + expect(mapRpcEvent(rpcEvent({ contractId: other }), CONTRACT_ID, 0)).toBeNull(); + }); + + it("maps a bet_plc event to topic symbol, marketId and JSON-safe payload", async () => { + const before = await counterValue(indexerEventsDecodedTotal, "bet_plc"); + const row = mapRpcEvent(rpcEvent(), CONTRACT_ID, 0); + expect(row).toEqual({ + ledger: 100, + txHash: "tx-100", + opIndex: 1, // low 12 bits of the TOID in the event id + eventType: "bet_plc", + marketId: MARKET, + payload: { bettor: BETTOR, outcome: "Yes", amount: "25000000", nonce: "7", timestamp: "1792000000" }, + }); + expect(() => JSON.stringify(row)).not.toThrow(); + expect(await counterValue(indexerEventsDecodedTotal, "bet_plc")).toBe(before + 1); + }); + + it("keeps unknown events as unknown: with raw XDR and counts them under 'unknown'", async () => { + const before = await counterValue(indexerEventsDecodedTotal, "unknown"); + const row = mapRpcEvent( + rpcEvent({ topic: [sym("foo_bar"), sym(MARKET)], value: xdr.ScVal.scvVoid() }), + CONTRACT_ID, + 0, + ); + expect(row?.eventType).toBe("unknown:foo_bar"); + expect(row?.marketId).toBe(MARKET); + expect((row?.payload as { raw: { value: string } }).raw.value).toBe( + xdr.ScVal.scvVoid().toXDR("base64"), + ); + expect(await counterValue(indexerEventsDecodedTotal, "unknown")).toBe(before + 1); + }); + + it("keeps invalid events as invalid: and increments the invalid counter", async () => { + const before = await counterValue(indexerEventsInvalidTotal, "bet_plc"); + const row = mapRpcEvent(rpcEvent({ value: betPlaced({}, ["amount"]) }), CONTRACT_ID, 0); + expect(row?.eventType).toBe("invalid:bet_plc"); + expect(row?.payload).toMatchObject({ errors: expect.any(Array), raw: { topics: expect.any(Array) } }); + expect(await counterValue(indexerEventsInvalidTotal, "bet_plc")).toBe(before + 1); + }); + + it("exposes the counters on the metrics registry", async () => { + const text = await register.metrics(); + expect(text).toContain("indexer_events_decoded_total"); + expect(text).toContain("indexer_events_invalid_total"); + }); + + it("getEvents returns decoded rows, drops non-contract events and never throws on bad events", async () => { + mockGetEvents.mockResolvedValueOnce({ + events: [ + { ...rpcEvent(), pagingToken: "p1" }, + { ...rpcEvent({ id: "0000000001-0000000003", type: "diagnostic" }), pagingToken: "p2" }, + { + ...rpcEvent({ id: "0000000001-0000000004", value: betPlaced({}, ["amount"]) }), + pagingToken: "p3", + }, + ], + }); + const client = createSorobanRpcClient(); + const events = await client.getEvents(100, 100); + expect(mockGetEvents).toHaveBeenCalledWith( + expect.objectContaining({ + filters: [{ type: "contract", contractIds: [CONTRACT_ID] }], + }), + ); + expect(events.map((e) => e.eventType)).toEqual(["bet_plc", "invalid:bet_plc"]); + expect(events[0].marketId).toBe(MARKET); + }); + + it("persists market_id and a JSON-safe data column", async () => { + const service = new IndexerService({ getLatestLedger: jest.fn(), getEvents: jest.fn() }); + mockQuery.mockImplementation(async (sql: string) => + sql.includes("INSERT INTO indexer_events") + ? { rowCount: 1, rows: [{ id: "1" }] } + : { rowCount: 0, rows: [] }, + ); + const row = mapRpcEvent(rpcEvent(), CONTRACT_ID, 0); + await service.persistEventsWithRecovery([row!]); + const insert = mockQuery.mock.calls.find(([sql]) => String(sql).includes("INSERT INTO indexer_events")); + const [sql, params] = insert!; + expect(sql).toContain("market_id"); + expect(params[3]).toBe(1); + expect(params.slice(5, 7)).toEqual(["bet_plc", MARKET]); + expect(JSON.parse(params[7])).toMatchObject({ bettor: BETTOR, amount: "25000000" }); + }); +}); diff --git a/tests/predictionsConfirmerJoin.test.ts b/tests/predictionsConfirmerJoin.test.ts new file mode 100644 index 00000000..057541dd --- /dev/null +++ b/tests/predictionsConfirmerJoin.test.ts @@ -0,0 +1,80 @@ +/** + * Confirmer join tests (issue #979, test plan rows 8 and 9). + * + * DrizzlePredictionsConfirmerRepo is driven through a real drizzle instance + * over a fake pg pool that records the SQL it is asked to run. There is no + * Postgres in the unit-test environment, so these tests pin the exact query + * contract (which rows may confirm a prediction); the SQL itself is not + * executed here. + * + * Row 8: a pending prediction whose tx only has an `unknown:*` event must not + * be confirmed -> the confirming join must require event_type 'bet_plc'. + * Row 9: a bet_plc whose bettor is not the prediction's user must not confirm + * -> the join must compare data->>'bettor' with the user's address. + */ + +import { drizzle } from "drizzle-orm/node-postgres"; + +import { DrizzlePredictionsConfirmerRepo } from "../src/workers/predictionsConfirmer"; + +interface Captured { + text: string; + values: unknown[]; +} + +function makeRepo(): { repo: DrizzlePredictionsConfirmerRepo; captured: Captured[] } { + const captured: Captured[] = []; + const pool = { + query: jest.fn(async (q: { text: string; values?: unknown[] } | string, v?: unknown[]) => { + const text = typeof q === "string" ? q : q.text; + const values = v ?? (typeof q === "string" ? [] : (q.values ?? [])); + captured.push({ text, values }); + return { rows: [], rowCount: 0, fields: [] }; + }), + }; + const db = drizzle(pool as never); + return { repo: new DrizzlePredictionsConfirmerRepo(db), captured }; +} + +const normalise = (sql: string): string => sql.replace(/\s+/g, " "); + +describe("DrizzlePredictionsConfirmerRepo confirming-event join", () => { + it("row 8: findConfirmedCandidateIds only matches canonical bet_plc events (unknown:* never confirms)", async () => { + const { repo, captured } = makeRepo(); + await repo.findConfirmedCandidateIds(10); + + expect(captured).toHaveLength(1); + const sql = normalise(captured[0].text); + expect(sql).toContain(`"indexer_events"."event_type" =`); + expect(sql).toContain(`"indexer_events"."canonical" = TRUE`); + expect(captured[0].values).toContain("bet_plc"); + // An event row with another type (e.g. 'unknown:foo_bar') cannot satisfy + // the equality above, and a tx_hash-only match is no longer sufficient. + expect(sql).not.toMatch(/inner join "indexer_events" on "predictions"."tx_hash" = "indexer_events"."tx_hash"\s+where/i); + }); + + it("row 9: findConfirmedCandidateIds requires market and bettor to match the prediction", async () => { + const { repo, captured } = makeRepo(); + await repo.findConfirmedCandidateIds(10); + + const sql = normalise(captured[0].text); + expect(sql).toContain(`"indexer_events"."market_id" = "predictions"."market_id"`); + expect(sql).toContain(`"indexer_events"."data"->>'bettor' = (`); + expect(sql).toContain(`SELECT "users"."stellar_address" FROM "users" WHERE "users"."id" = "predictions"."user_id"`); + }); + + it("incrementAttempts and markFailed use the same predicate, so unconfirmable predictions still age out", async () => { + const { repo, captured } = makeRepo(); + await repo.incrementAttempts(3, 10); + await repo.markFailed(3, 10); + + expect(captured).toHaveLength(2); + for (const { text, values } of captured) { + const sql = normalise(text); + expect(sql).toContain("NOT EXISTS"); + expect(sql).toContain(`"indexer_events"."canonical" = TRUE`); + expect(sql).toContain(`"indexer_events"."data"->>'bettor'`); + expect(values).toContain("bet_plc"); + } + }); +});