diff --git a/.env.example b/.env.example index 5417389..3cbd6bb 100644 --- a/.env.example +++ b/.env.example @@ -40,6 +40,10 @@ HORIZON_URL_TESTNET=https://horizon-testnet.stellar.org RPC_URL_TESTNET=https://soroban-testnet.stellar.org NETWORK_PASSPHRASE_TESTNET=Test SDF Network ; September 2015 SOROSWAP_FACTORY_ADDRESS_TESTNET=CDKP5WSEZMDL53VZFPBGCL47WBPKFCN5OPYQVXB3CJWUXHPZRPHSSZ3 +# Soroswap has a testnet deployment; leave enabled. Set to "false" to disable. +SOROSWAP_ENABLED_TESTNET=true +# Aquarius has no public testnet deployment today — disabled by default. +AQUARIUS_ENABLED_TESTNET=false REFLECTOR_CONTRACT_ID_TESTNET= # Comma-separated pairs to watch on testnet. # Format: "CODE:ISSUER/CODE:ISSUER". Use "native" for XLM. @@ -53,11 +57,20 @@ RPC_URL_MAINNET=https://your-provider.example.com/soroban-rpc NETWORK_PASSPHRASE_MAINNET=Public Global Stellar Network ; September 2015 # Mainnet Soroswap factory contract address — see https://github.com/soroswap/core SOROSWAP_FACTORY_ADDRESS_MAINNET=CA4HEQTL2WPEUYKYKCDOHCDNIV4QHNJ7EL4J4NQ6VADP7SYHVRYZ7AW2 +SOROSWAP_ENABLED_MAINNET=true +AQUARIUS_ENABLED_MAINNET=true # Reflector oracle contract on mainnet — see https://reflector.network REFLECTOR_CONTRACT_ID_MAINNET=CCYXZMNHFXHKF3YEX4VJJ5TH3YHCVZIBPNBGM7C4PJIMCIMNNWDOQYA # Comma-separated pairs to watch on mainnet. WATCHED_PAIRS_MAINNET=XLM:native/USDC:GA5ZSEJYB37JRC5AVCIA5MOP4RHTM335X2KGX3IHOJAPP5RE34K4KZVN +# --- Venue endpoint overrides (optional, shared across networks unless a +# paired _TESTNET / _MAINNET variant is set) --- +# Soroswap token-list JSON URL. +SOROSWAP_TOKEN_LIST_URL=https://raw.githubusercontent.com/soroswap/token-list/main/tokenList.json +# Aquarius AMM pools API base URL. +AQUARIUS_API_URL=https://amm.aquarius.network/api/v1/pools/ + # --- Back-compat single-network vars (testnet) --- # These are still respected when the paired _TESTNET vars above are unset. # New deployments should prefer the paired vars above. @@ -91,6 +104,9 @@ REQUIRE_API_KEY=true # --- x402 Payment Gate --- # Stellar public key where API payments should be sent. # If unset, x402 gating is disabled. +# Optionally set ORACLE_PAYMENT_ADDRESS_TESTNET / ORACLE_PAYMENT_ADDRESS_MAINNET +# to use a different payout address per network (falls back to the shared +# ORACLE_PAYMENT_ADDRESS above for whichever one is unset). ORACLE_PAYMENT_ADDRESS=GD... # URL of the x402 facilitator (default: https://facilitator.stellar.org) X402_FACILITATOR_URL=https://facilitator.stellar.org diff --git a/README.md b/README.md index 7cd291f..6a4f0cb 100644 --- a/README.md +++ b/README.md @@ -21,6 +21,19 @@ Aggregates price data from Stellar's Classic Order Book (SDEX) and AMM Liquidity | GET | `/pools` | Active AMM pools being watched | | GET | `/pairs` | Watched trading pairs | | GET | `/status` | Indexer health | +| GET | `/discovery/resources?type=&payTo=&network=&extensions=&limit=&offset=` | Bazaar catalog of x402-discoverable resources (spec: [`bazaar`](https://github.com/x402-foundation/x402/blob/main/specs/extensions/bazaar.md)) | + +Every route accepts an optional `?network=testnet\|mainnet` query param (or +`x-network` header) to pick the Stellar network — default is `testnet`. An +unrecognised value gets `400`. The `/price/*` endpoints' live SDEX pricing and +x402 payment `network`/`payTo` are fully per-request today; DB-backed reads +(candles, history, pools, AMM pricing) are still served from whichever +network this instance is currently indexing (`STELLAR_NETWORK`) — that data +layer isn't network-partitioned yet. + +```bash +curl "https://api.example.com/price/XLM/USDC?network=mainnet" +``` ### GraphQL Available at `/graphql` with GraphiQL IDE at `/graphiql`. @@ -211,13 +224,14 @@ npm run dev | `HORIZON_URL` | Stellar Horizon server URL | - | No | | `RPC_URL` | Soroban RPC server URL | - | No | | `NETWORK_PASSPHRASE` | Stellar network passphrase | - | No | -| `STELLAR_NETWORK` | `mainnet` or `testnet` (for x402 logic) | `testnet` | No | +| `STELLAR_NETWORK` | `mainnet` or `testnet` — this instance's default/ingested network | `testnet` | No | | `POLL_INTERVAL_MS` | Indexer polling frequency (ms) | `5000` | No | | `SDEX_PAGE_SIZE` | Trades per page for SDEX ingestion | `200` | No | | `AMM_PAGE_SIZE` | Trades per page for AMM ingestion | `200` | No | | `ADMIN_API_KEY` | Key for admin route authentication | - | No | | `WATCHED_PAIRS` | Comma-separated list of asset pairs to index | - | **Yes** | | `ORACLE_PAYMENT_ADDRESS` | Stellar address for x402 API payments | - | No* | +| `ORACLE_PAYMENT_ADDRESS_TESTNET` / `ORACLE_PAYMENT_ADDRESS_MAINNET` | Per-network override for the address above | - | No | | `X402_FACILITATOR_URL` | x402 facilitator service URL | - | No | *\*Required if enabling x402 payment gating.* diff --git a/docs/x402/scheme_upto_stellar.md b/docs/x402/scheme_upto_stellar.md new file mode 100644 index 0000000..b064158 --- /dev/null +++ b/docs/x402/scheme_upto_stellar.md @@ -0,0 +1,185 @@ +# Scheme: `upto` on `Stellar` + +> Draft specification for the x402 `upto` scheme on Stellar. Stellar currently +> has a specified `exact` scheme upstream but no `upto` scheme; this document +> proposes one, matching the format of the existing `exact` Stellar spec and +> the `upto` specs for EVM and SVM in the +> [x402 protocol repository](https://github.com/x402-foundation/x402). + +## Versions supported + +- ❌ `v1` - we don't plan to support v1 for now. +- ✅ `v2` + +## Supported Networks + +This spec uses [CAIP-2](https://namespaces.chainagnostic.org/stellar/caip2) identifiers: +- `stellar:pubnet` — Stellar mainnet +- `stellar:testnet` — Stellar testnet + +## Summary + +The x402 `upto` scheme on Stellar authorizes a transfer of up to a **maximum amount**, with the actual amount settled after resource consumption is known. As with [`exact` on Stellar][scheme-exact-stellar], the client authorizes via a signed Soroban authorization entry rather than a full transaction signature, and the facilitator sponsors transaction fees. + +> [!NOTE] +> **Scope:** This spec covers [SEP-41]-compliant Soroban tokens **only**. Classic Stellar assets are not supported, matching [`exact` on Stellar][scheme-exact-stellar]. + +## Contract-vs-No-Contract Decision + +**This spec ships a Soroban contract: `x402UptoStellar`.** + +A bare [SEP-41] `approve` / `transfer_from` allowance cannot, on its own, provide the three guarantees the [`upto` core spec][scheme-upto] requires of every network implementation: + +- **Recipient binding** — a SEP-41 allowance authorizes a `spender` to move up to `amount`, to *any* address that spender chooses. It does not bind the transfer to a specific `payTo`. +- **Single-use authorization** — a SEP-41 allowance is a standing balance the spender can draw down across many calls; it has no built-in single-settlement or replay-protection semantics. +- **Time-bound authorization with an explicit `validAfter`/deadline pair** — SEP-41's `approve` supports an expiration ledger, but not a `validAfter` start bound, and nothing stops the spender from settling more than once before expiry. + +This mirrors the precedent already set by the two implemented networks: EVM does not rely on a bare ERC-20 `approve` either — it ships the purpose-built `x402UptoPermit2Proxy` contract, which wraps Permit2's witness pattern to add recipient binding, a nonce, and settle-time cap enforcement. SVM ships the `payment-channels` program for the same reason. A Stellar design that stopped at "the client approves the facilitator" would be strictly weaker than both existing implementations, not merely different from them — so this spec does not offer a contract-free path. + +The `x402UptoStellar` contract is intentionally thin: it does not escrow the client's funds up front (unlike SVM's channel model). Instead, it composes with a standard SEP-41 `approve`, and adds the missing recipient binding, nonce, and cap enforcement as a signed-authorization wrapper around `transfer_from`, closer in spirit to the EVM Permit2 witness pattern than to SVM's escrow. This keeps capital efficient (no funds are locked before settlement) and keeps the per-request signing step off-chain and free, consistent with how `exact` on Stellar already uses auth-entry signing instead of on-chain transactions for the client's half of the flow. + +## Summary of the Flow + +1. **Client** performs a one-time (or periodic) [SEP-41] `approve(spender: , amount: , expiration_ledger)` on the token contract. This is an on-chain transaction, analogous to EVM's Phase 1 Permit2 approval — it is not part of the per-request payment flow and is skipped on subsequent requests while the allowance remains sufficient and unexpired. +2. **Resource Server** responds `402 Payment Required` with `PaymentRequirements` for `upto`, including `extra.uptoContract` (the `x402UptoStellar` contract address) and `extra.facilitatorAddress`. +3. **Client** signs a Soroban authorization entry for `x402UptoStellar.settle_upto(from, to, asset, max_amount, facilitator, nonce, expiration_ledger)`, with `max_amount` set to the authorized ceiling for this request (this is the `PaymentRequirements.amount` field at verification time), `to` set to `requirements.payTo`, and `facilitator` set to `extra.facilitatorAddress`. Expiration is derived from `maxTimeoutSeconds` exactly as in `exact` on Stellar. +4. **Client** serializes the signed authorization entry (base64 XDR) and sends it to the resource server as the `PaymentPayload`. +5. **Resource Server** forwards the payload to the **Facilitator**'s `/verify` endpoint, which checks the authorization entry's structure, expiration, and that `requirements.amount <= max_amount`, without settling. +6. **Resource Server** executes the request, determines the actual cost, and calls the facilitator's `/settle` endpoint with `requirements.amount` set to the actual amount (phase-dependent `amount`, per the [`upto` core spec][scheme-upto] §5). +7. **Facilitator** re-verifies the client's authorization entry against `max_amount` (the signed ceiling, not the settlement-time amount — see [Settle-Time Verification](#settle-time-verification)), then calls `x402UptoStellar.settle_upto(auth_entry, actual_amount)`. +8. **Contract** checks the nonce has not been used, `now <= expiration_ledger`, `facilitator.require_auth()` matches the signed `facilitator`, and `actual_amount <= max_amount`; it then calls `token.transfer_from(spender: self, from, to, actual_amount)` and marks the nonce consumed. +9. **Facilitator** submits the transaction, sponsoring fees as in `exact`, and returns a `SettlementResponse` to the **Resource Server**, which grants access to the **Client**. + +## `PaymentRequirements` for `upto` + +```json +{ + "scheme": "upto", + "network": "stellar:testnet", + "amount": "10000000", + "asset": "CBIELTK6YBZJU5UP2WWQEUCYKLPU6AUNZ2BQ4WWFEIE3USCIHMXQDAMA", + "payTo": "GBHEGW3KWOY2OFH767EDALFGCUTBOEVBDQMCKU4APMDLQNBW5QV3W3KO", + "maxTimeoutSeconds": 300, + "extra": { + "areFeesSponsored": true, + "uptoContract": "CA...UPTOSTELLARCONTRACTADDRESS", + "facilitatorAddress": "GBFACILITATOR..." + } +} +``` + +**Field Definitions:** + +- `amount`: Phase-dependent per the [`upto` core spec][scheme-upto] — the authorized maximum at verification time, the actual settlement amount at settlement time. +- `extra.uptoContract`: The deployed `x402UptoStellar` contract address the client must authorize against. +- `extra.facilitatorAddress`: The facilitator the client binds into the signed authorization (mirrors EVM's `witness.facilitator`); prevents settlement by any other party. +- `extra.areFeesSponsored`: As in `exact` — currently always `true`. + +## PaymentPayload `payload` Field + +```json +{ + "authEntry": "AAAAAgAAAABriIN4poutFUmHfB6FbFJu8GgXoPPTGQWREqFpPfvO1AAAAAAAAAAAAAAAAAAAAA...", + "nonce": "3f1a...b2", + "maxAmount": "10000000" +} +``` + +- `authEntry`: Base64-encoded XDR of the signed Soroban authorization entry for `settle_upto`. +- `nonce`: The 32-byte nonce bound into the authorization entry, surfaced separately so the facilitator can perform a fast pre-check without decoding XDR. +- `maxAmount`: The signed ceiling, echoed outside the XDR for the same reason. + +## Facilitator Verification Rules (MUST) + +### 1. Protocol Validation + +Same as `exact` on Stellar: `x402Version` MUST be `2`, `scheme` MUST be `"upto"` on both sides, `network` MUST match. + +### 2. Authorization Entry Structure + +- The authorization entry MUST target the `x402UptoStellar` contract at `extra.uptoContract` and the `settle_upto` function. +- Arguments MUST be exactly `(from, to, asset, max_amount, facilitator, nonce, expiration_ledger)`. +- `to` MUST equal `requirements.payTo` exactly. +- `asset` MUST equal `requirements.asset` exactly. +- `facilitator` MUST equal `extra.facilitatorAddress` exactly. +- Credential type MUST be `sorobanCredentialsAddress`, matching `exact`. + +### 3. Cap and Amount Rules + +- At **verify** time: `requirements.amount` (the authorized maximum) MUST equal the signed `max_amount`. +- At **settle** time: `requirements.amount` (the actual settlement amount) MUST be `<= max_amount` from the signed entry. The facilitator MUST re-verify the authorization entry's signature against `max_amount`, never against the settlement-time amount — see [Settle-Time Verification](#settle-time-verification). +- The settled amount MAY be `0`. + +### 4. Time Bounds and Replay + +- `expiration_ledger` MUST NOT exceed `currentLedger + ceil(maxTimeoutSeconds / estimatedLedgerSeconds)` (fallback `5` seconds/ledger, as in `exact`). +- The facilitator MUST query the contract's nonce state (or simulate `settle_upto`) before submitting, to short-circuit already-consumed nonces. +- The contract itself is the source of truth for replay protection: `settle_upto` MUST fail if `nonce` has already been marked consumed. + +### 5. 🚨🚨🚨 Facilitator Safety + +Same as `exact` §4: the facilitator's own address MUST NOT be `from`, MUST NOT appear as an unexpected signer, and simulation MUST show only the expected balance change (`from` decrease of `actual_amount`, `to` increase of `actual_amount`) plus no other balance changes. + +### 6. Allowance Precondition + +- Before verification can succeed, the facilitator MUST confirm `token.allowance(from, uptoContract) >= max_amount` and unexpired. If insufficient, the facilitator MUST return `412 Precondition Failed` with an error code equivalent to EVM's `PERMIT2_ALLOWANCE_REQUIRED`, signaling the client to submit the one-time `approve` first. + +## Settle-Time Verification + +Identical rationale to [EVM `upto` §Settle-Time Verification][scheme-upto-evm-settle]: because `amount` is phase-dependent, the facilitator MUST: + +1. Verify the authorization entry's signature against the signed `max_amount` (the ceiling), not `requirements.amount` (the actual settlement amount) — the client signed for the ceiling, and comparing against the metered amount would reject every partial settlement. +2. Validate `requirements.amount <= max_amount`. +3. Call `settle_upto` with `actual_amount = requirements.amount`. + +A facilitator that instead enforces `requirements.amount === max_amount` at settle time will reject all partial settlements, breaking the core `upto` value proposition. + +## `x402UptoStellar` Contract + +Reference behavior (Soroban, Rust): + +- `settle_upto(from: Address, to: Address, asset: Address, max_amount: i128, facilitator: Address, nonce: BytesN<32>, expiration_ledger: u32, actual_amount: i128)` + - Requires `from`'s authorization on this invocation with the fixed args `(from, to, asset, max_amount, facilitator, nonce, expiration_ledger)` — `actual_amount` is deliberately **not** part of the signed argument set, since it is only known at settle time; this is the Stellar analogue of Permit2's witness/permitted-amount split. + - Requires `facilitator.require_auth()`, binding settlement to the designated facilitator. + - Fails if `nonce` has already been consumed for `from`. + - Fails if `env.ledger().sequence() > expiration_ledger`. + - Fails if `actual_amount > max_amount` or `actual_amount < 0`. + - On success: calls `token_client.transfer_from(&env.current_contract_address(), &from, &to, &actual_amount)`, marks `nonce` consumed, and emits a `settle` event with `(from, to, asset, actual_amount, nonce)`. +- Zero settlement (`actual_amount == 0`): the facilitator MAY skip calling `settle_upto` entirely and let the authorization entry expire unused, exactly as EVM's zero-settlement case — no on-chain transaction, no gas cost, the nonce is simply never consumed. If the resource server needs an on-chain record of the zero-charge decision, the facilitator MAY still call `settle_upto` with `actual_amount = 0` to consume the nonce and close out the authorization explicitly. + +## Composition with Smart Account Spending Policies + +The `x402UptoStellar` contract enforces guarantees about a *single authorization*: it cannot be settled twice, cannot exceed its signed ceiling, and cannot be redirected to a different recipient. It says nothing about how many such authorizations a given signer is allowed to produce in total — that is a separate concern, and on Stellar it is naturally handled one layer up, at the smart account (C-account) level, rather than inside the x402 contract itself. + +For an agentic payer whose Stellar account is a smart wallet (e.g. a passkey-backed contract account exposing its own `approve(spender, token, amount, expiry)` policy independent of the underlying SEP-41 token allowance), the composition looks like this: + +- The **x402 `upto` authorization** (this spec) bounds a single request: at most `max_amount`, to exactly `payTo`, settleable at most once. +- The **smart account's own spending policy** bounds the agent's signing key across *all* requests: it governs whether the wallet will co-sign an `approve` or `settle_upto` authorization entry for a given `(spender, token)` pair at all, independent of what any individual x402 authorization says. + +These two layers are enforced at different points and do not need to agree on mechanism: the smart account's policy check happens client-side, when the wallet decides whether to produce a signature for the agent's requested authorization entry; the `x402UptoStellar` contract's checks happen on-chain, at settle time, regardless of what kind of account `from` is. An agent authorized up to a per-request cap by `upto`, running against a smart wallet with its own aggregate spending policy, is bounded by both: the smart wallet refuses to keep signing once its own policy limit is reached, even if individual `upto` ceilings would otherwise permit more spend. This is a recommended defense-in-depth pattern for agent use cases, not a protocol requirement — `upto` on Stellar functions identically for a plain G-account payer with no wallet-level policy at all. + +## Out of Scope + +- **`batch-settlement`**: settling more than once against the same authorization (streaming/pay-per-chunk) is explicitly out of scope for `upto`, per the [core spec][scheme-upto]. Nothing in `x402UptoStellar`'s nonce design forecloses a future `batch-settlement` scheme built on the same escrow/voucher primitives used elsewhere in the ecosystem — that would be a distinct scheme with its own spec. +- **`auth-capture`**: deferred; the single-settlement nonce model here does not preclude a future two-phase design. + +## Error Codes + +In addition to the standard x402 error codes: + +- **`invalid_upto_stellar_payload_settlement_exceeds_amount`**: attempted settlement amount exceeds the signed `max_amount`. +- **`invalid_upto_stellar_payload_nonce_consumed`**: the authorization entry's nonce has already been settled. +- **`invalid_upto_stellar_payload_allowance_required`**: the client has not approved `x402UptoStellar` for at least `max_amount` (see [Allowance Precondition](#6-allowance-precondition)). + +## Security Considerations + +1. **Maximum amount authorization**: as in EVM/SVM, clients should sign `max_amount` conservatively; the facilitator can settle for any amount up to it. +2. **Server trust**: `upto` requires trusting the resource server to report actual usage honestly; this is unchanged from the core scheme. +3. **Allowance ceiling vs per-request ceiling**: the SEP-41 `approve` ceiling and the per-request `max_amount` are independent. Implementations SHOULD keep the allowance close to the expected per-request ceiling and re-approve as needed, rather than approving a large standing balance, to limit exposure if a facilitator or the escrow contract is ever compromised. +4. **Nonce exhaustion / griefing**: because settlement is optional (zero-settlement need not touch the chain), a malicious client cannot force facilitator gas spend by signing many unused authorizations; only the facilitator's own `/settle` calls cost gas, and it only calls them for its own resource server's confirmed usage. +5. **Smart account composition**: see [Composition with Smart Account Spending Policies](#composition-with-smart-account-spending-policies) — this is an additive, optional safeguard and its absence does not weaken the guarantees this spec makes about a single authorization. + +[SEP-41]: https://stellar.org/protocol/sep-41 +[scheme-exact-stellar]: https://github.com/x402-foundation/x402/blob/main/specs/schemes/exact/scheme_exact_stellar.md +[scheme-upto]: https://github.com/x402-foundation/x402/blob/main/specs/schemes/upto/scheme_upto.md +[scheme-upto-evm-settle]: https://github.com/x402-foundation/x402/blob/main/specs/schemes/upto/scheme_upto_evm.md#settle-time-verification diff --git a/prisma/schema.prisma b/prisma/schema.prisma index c36ea2f..378beb4 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -11,6 +11,7 @@ datasource db { model PricePoint { id String @default(uuid()) + network String @default("testnet") assetA String @map("asset_a") assetB String @map("asset_b") pairKey String @map("pair_key") @@ -24,13 +25,14 @@ model PricePoint { eventId String? @map("event_id") @@id([id, timestamp]) - @@index([pairKey, timestamp(sort: Desc)]) - @@index([pairKey, source, timestamp(sort: Desc)]) + @@index([network, pairKey, timestamp(sort: Desc)]) + @@index([network, pairKey, source, timestamp(sort: Desc)]) @@map("price_points") } model PoolSnapshot { id String @default(uuid()) + network String @default("testnet") poolId String @map("pool_id") assetA String @map("asset_a") assetB String @map("asset_b") @@ -43,12 +45,13 @@ model PoolSnapshot { timestamp DateTime @@id([id, timestamp]) - @@index([poolId, timestamp(sort: Desc)]) + @@index([network, poolId, timestamp(sort: Desc)]) @@map("pool_snapshots") } model PriceAggregate { pairKey String @map("pair_key") + network String @default("testnet") window String bucket DateTime vwap Decimal @db.Decimal(36, 18) @@ -63,44 +66,142 @@ model PriceAggregate { highPrice Decimal? @map("high_price") @db.Decimal(36, 18) lowPrice Decimal? @map("low_price") @db.Decimal(36, 18) - @@id([pairKey, window, bucket]) + @@id([network, pairKey, window, bucket]) @@map("price_aggregates") } model PriceSnapshot { pair String + network String @default("testnet") ts DateTime price Decimal @db.Decimal(36, 18) volume Decimal @default(0) @db.Decimal(36, 7) - @@id([pair, ts]) - @@index([pair, ts]) + @@id([network, pair, ts]) + @@index([network, pair, ts]) @@map("price_snapshots") } model IndexerState { - id String @id + id String + network String @default("testnet") lastCursor String? @map("last_cursor") lastLedger Int? @map("last_ledger") lastProcessedAt DateTime? @map("last_processed_at") updatedAt DateTime @default(now()) @updatedAt @map("updated_at") + @@id([network, id]) @@map("indexer_state") } model PairConfig { - pairKey String @id @map("pair_key") + pairKey String @map("pair_key") + network String @default("testnet") assetACode String @map("asset_a_code") assetAIssuer String? @map("asset_a_issuer") assetBCode String @map("asset_b_code") assetBIssuer String? @map("asset_b_issuer") addedAt DateTime @default(now()) @map("added_at") + @@id([network, pairKey]) @@map("pair_configs") } +/// A single x402-discoverable resource in the Bazaar catalog — either an +/// HTTP endpoint or an MCP tool, per the x402 `bazaar` extension +/// (specs/extensions/bazaar.md in x402-foundation/x402). +/// +/// HTTP and MCP resources share one table (discriminated by `type`) rather +/// than two, because a discovery listing is fundamentally "a resource with +/// payment requirements and a bazaar.info blob" regardless of transport — +/// splitting them would require the discovery query to UNION two tables on +/// every filter combination for no benefit, since the two types are never +/// looked up via different access patterns. +model BazaarResource { + id String @id @default(uuid()) + + /// "http" | "mcp" — discriminates which of the two input shapes below applies. + type String + + /// Which Stellar network this listing settles on ("mainnet" | "testnet"). + /// Mirrors config.ts's NetworkName so a listing is never ambiguous about + /// which network's payTo/asset it refers to. + network String + + /// The protected resource URL (`resource.url` in the spec). For MCP this is + /// the MCP server endpoint, not the tool itself — the tool is disambiguated + /// by `mcpToolName` below. + url String + + /// `resource.description` — human-readable description of the resource. + description String? + + /// `resource.mimeType`. + mimeType String? @map("mime_type") + + /// Optional service metadata the spec allows on `resource`. + serviceName String? @map("service_name") + tags String[] @default([]) + iconUrl String? @map("icon_url") + + /// MCP tool identifier (`input.toolName`). Null for HTTP resources. + /// Per the spec, MCP resources are keyed on the TUPLE of (resource.url, + /// input.toolName) since multiple tools multiplex over one server endpoint. + /// We additionally scope that tuple by `network` (see @@unique below) — + /// a deliberate deviation, called out in the PR: since Lens is + /// dual-network, the same (url, toolName) pair can legitimately exist + /// once per network with a different payTo/asset in `accepts`, and the + /// spec's tuple alone can't express that without collapsing them. + mcpToolName String? @map("mcp_tool_name") + + /// HTTP method for HTTP resources (GET/POST/...). Null for MCP resources. + httpMethod String? @map("http_method") + + /// Full `accepts[]` payment requirements array (scheme/network/amount/asset/ + /// payTo/maxTimeoutSeconds/extra), stored verbatim so the discovery response + /// can round-trip the exact PaymentRequirements the resource advertised. + accepts Json + + /// The `payTo` address extracted from accepts[0] for indexed filtering. + /// Denormalized on write because Postgres cannot efficiently index into a + /// JSON array element without a functional/GIN index per accepted scheme, + /// and payTo is the one field the spec calls out as a top-level filter. + payTo String @map("pay_to") + + /// `extensions.bazaar.info` — discovery metadata (input type, params, output). + bazaarInfo Json @map("bazaar_info") + + /// `extensions.bazaar.schema` — JSON Schema validating `bazaarInfo`. + bazaarSchema Json @map("bazaar_schema") + + /// `extensions.bazaar.routeTemplate` — canonical `:param` pattern for + /// dynamic HTTP routes, used by the facilitator to consolidate listings. + routeTemplate String? @map("route_template") + + /// Any other declared extension keys beyond "bazaar" (spec's `extensions` + /// filter matches on presence of a key here, "bazaar" always included). + extensionKeys String[] @default(["bazaar"]) @map("extension_keys") + + createdAt DateTime @default(now()) @map("created_at") + updatedAt DateTime @default(now()) @updatedAt @map("updated_at") + + // HTTP resources are keyed on (network, url, httpMethod); MCP resources are + // keyed on (network, url, mcpToolName) per the spec's tuple. Postgres + // treats NULLs as distinct in a unique index, so these two constraints + // don't collide with each other for a row that only populates one side. + @@unique([network, url, httpMethod], name: "bazaarHttpIdentity", map: "bazaar_http_identity") + @@unique([network, url, mcpToolName], name: "bazaarMcpIdentity", map: "bazaar_mcp_identity") + // Covers the six spec filters (type, payTo, network, extensions via + // extensionKeys, plus limit/offset) and keeps pagination stable — see + // routes/discovery.ts, which always orders by (createdAt, id). + @@index([network, type, payTo, createdAt(sort: Desc), id]) + @@index([extensionKeys], type: Gin) + @@map("bazaar_resources") +} + model Webhook { id String @id @default(uuid()) + network String @default("testnet") url String assetA String @map("asset_a") assetB String @map("asset_b") @@ -109,7 +210,7 @@ model Webhook { secret String createdAt DateTime @default(now()) @map("created_at") - @@index([assetA, assetB]) + @@index([network, assetA, assetB]) @@map("webhooks") } diff --git a/sql/schema.sql b/sql/schema.sql deleted file mode 100644 index b03d98b..0000000 --- a/sql/schema.sql +++ /dev/null @@ -1,93 +0,0 @@ --- Raw price points from SDEX trades and AMM swaps -CREATE TABLE IF NOT EXISTS price_points ( - id UUID DEFAULT gen_random_uuid() PRIMARY KEY, - asset_a TEXT NOT NULL, - asset_b TEXT NOT NULL, - pair_key TEXT NOT NULL, - source TEXT NOT NULL CHECK (source IN ('SDEX', 'AMM')), - pool_id TEXT, - price NUMERIC(36, 18) NOT NULL, - base_volume NUMERIC(36, 7) NOT NULL, - counter_volume NUMERIC(36, 7) NOT NULL, - ledger INTEGER NOT NULL, - timestamp TIMESTAMPTZ NOT NULL, - event_id TEXT -); - -CREATE INDEX IF NOT EXISTS idx_price_points_pair_time ON price_points (pair_key, timestamp DESC); -CREATE INDEX IF NOT EXISTS idx_price_points_pair_source_time ON price_points (pair_key, source, timestamp DESC); -CREATE INDEX IF NOT EXISTS idx_price_points_pool_time ON price_points (pool_id, timestamp DESC) WHERE pool_id IS NOT NULL; - --- AMM pool reserve snapshots -CREATE TABLE IF NOT EXISTS pool_snapshots ( - id UUID DEFAULT gen_random_uuid() PRIMARY KEY, - pool_id TEXT NOT NULL, - asset_a TEXT NOT NULL, - asset_b TEXT NOT NULL, - reserve_a NUMERIC(36, 7) NOT NULL, - reserve_b NUMERIC(36, 7) NOT NULL, - spot_price NUMERIC(36, 18) NOT NULL, - total_shares NUMERIC(36, 7), - fee_bp INTEGER DEFAULT 30, - ledger INTEGER NOT NULL, - timestamp TIMESTAMPTZ NOT NULL -); - -CREATE INDEX IF NOT EXISTS idx_pool_snapshots_pool_time ON pool_snapshots (pool_id, timestamp DESC); -CREATE INDEX IF NOT EXISTS idx_pool_snapshots_assets_time ON pool_snapshots (asset_a, asset_b, timestamp DESC); - --- Pre-computed VWAP aggregates -CREATE TABLE IF NOT EXISTS price_aggregates ( - pair_key TEXT NOT NULL, - "window" TEXT NOT NULL CHECK ("window" IN ('1m', '5m', '1h', '24h')), - bucket TIMESTAMPTZ NOT NULL, - vwap NUMERIC(36, 18) NOT NULL, - sdex_vwap NUMERIC(36, 18), - amm_vwap NUMERIC(36, 18), - volume NUMERIC(36, 7) NOT NULL DEFAULT 0, - sdex_volume NUMERIC(36, 7) DEFAULT 0, - amm_volume NUMERIC(36, 7) DEFAULT 0, - trade_count INTEGER DEFAULT 0, - open_price NUMERIC(36, 18), - close_price NUMERIC(36, 18), - high_price NUMERIC(36, 18), - low_price NUMERIC(36, 18), - PRIMARY KEY (pair_key, "window", bucket) -); - --- 1-minute price snapshot ring buffer. --- The ingester appends one row per watched pair per minute; a retention job --- prunes rows older than 30 days. Powers the /prices/history endpoint (charts, --- backtests, audit trails) without paying the cost of scanning raw price_points. -CREATE TABLE IF NOT EXISTS price_snapshots ( - pair TEXT NOT NULL, - ts TIMESTAMPTZ NOT NULL, - price NUMERIC(36, 18) NOT NULL, - volume NUMERIC(36, 7) NOT NULL DEFAULT 0, - PRIMARY KEY (pair, ts) -); - -CREATE INDEX IF NOT EXISTS idx_price_snapshots_pair_ts ON price_snapshots (pair, ts); - --- Indexer cursor state -CREATE TABLE IF NOT EXISTS indexer_state ( - id TEXT PRIMARY KEY, - last_cursor TEXT, - last_ledger INTEGER, - last_processed_at TIMESTAMPTZ, - updated_at TIMESTAMPTZ DEFAULT NOW() -); - --- API keys for authenticated, rate-quota'd access. --- Only the SHA-256 hash of each key is ever stored — never the plaintext. -CREATE TABLE IF NOT EXISTS api_keys ( - id UUID DEFAULT gen_random_uuid() PRIMARY KEY, - hash TEXT NOT NULL UNIQUE, - label TEXT NOT NULL, - rate_per_min INTEGER NOT NULL DEFAULT 60, - rate_per_day INTEGER NOT NULL DEFAULT 10000, - revoked_at TIMESTAMPTZ, - created_at TIMESTAMPTZ DEFAULT NOW() -); - -CREATE INDEX IF NOT EXISTS idx_api_keys_hash ON api_keys (hash); diff --git a/src/__tests__/aquariusIngester.test.ts b/src/__tests__/aquariusIngester.test.ts new file mode 100644 index 0000000..e4ad404 --- /dev/null +++ b/src/__tests__/aquariusIngester.test.ts @@ -0,0 +1,82 @@ +/** + * Unit tests for the Aquarius AMM venue adapter. + */ + +const mocks = vi.hoisted(() => ({ + config: { + aquarius: { enabled: true, apiUrl: 'https://amm.aquarius.network/api/v1/pools/' }, + indexer: { pollIntervalMs: 5000 }, + }, + pairsRegistry: { + getActivePairs: vi.fn().mockReturnValue([]), + }, +})) + +vi.mock('../config', () => ({ config: mocks.config })) +vi.mock('../pairsRegistry', () => mocks.pairsRegistry) +vi.mock('../db', () => ({ upsertPricePoints: vi.fn().mockResolvedValue(undefined) })) +vi.mock('../webhookDispatcher', () => ({ dispatchPriceUpdate: vi.fn().mockResolvedValue(undefined) })) + +import { fetchAquariusPools, startAquariusIngester } from '../ingest/venues/aquarius' + +const mockPair = { + pairKey: 'USDC/XLM', + assetA: { code: 'XLM', issuer: null }, + assetB: { code: 'USDC', issuer: 'GBBD47IF6LWK7P7MDEVSCWR7DPUWV3NY3DTQEVFL4NAT4AQH3ZLLFLA5' }, +} + +describe('fetchAquariusPools', () => { + beforeEach(() => { + vi.clearAllMocks() + global.fetch = vi.fn() + }) + + it('queries the configured Aquarius API URL for the network', async () => { + ;(global.fetch as ReturnType).mockResolvedValue({ + ok: true, + json: async () => ({ results: [] }), + }) + + await fetchAquariusPools(mockPair as any, 'https://testnet.example.com/pools/') + + expect(global.fetch).toHaveBeenCalledWith( + expect.stringContaining('https://testnet.example.com/pools/?') + ) + }) + + it('falls back to config.aquarius.apiUrl when no override is passed', async () => { + ;(global.fetch as ReturnType).mockResolvedValue({ + ok: true, + json: async () => ({ results: [] }), + }) + + await fetchAquariusPools(mockPair as any) + + expect(global.fetch).toHaveBeenCalledWith( + expect.stringContaining(mocks.config.aquarius.apiUrl) + ) + }) + + it('returns an empty array on a non-ok response', async () => { + ;(global.fetch as ReturnType).mockResolvedValue({ ok: false }) + + const result = await fetchAquariusPools(mockPair as any) + expect(result).toEqual([]) + }) +}) + +describe('startAquariusIngester', () => { + beforeEach(() => { + vi.clearAllMocks() + }) + + it('does not start the polling loop when Aquarius is disabled on the active network', async () => { + mocks.config.aquarius.enabled = false + + await startAquariusIngester() + + expect(mocks.pairsRegistry.getActivePairs).not.toHaveBeenCalled() + + mocks.config.aquarius.enabled = true + }) +}) diff --git a/src/__tests__/bazaarCatalog.test.ts b/src/__tests__/bazaarCatalog.test.ts new file mode 100644 index 0000000..8952d82 --- /dev/null +++ b/src/__tests__/bazaarCatalog.test.ts @@ -0,0 +1,248 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest' + +const { mockFindMany, mockCount, mockUpsert, mockDeleteMany } = vi.hoisted(() => ({ + mockFindMany: vi.fn(), + mockCount: vi.fn(), + mockUpsert: vi.fn(), + mockDeleteMany: vi.fn(), +})) + +vi.mock('../db', () => ({ + prisma: { + bazaarResource: { + findMany: mockFindMany, + count: mockCount, + upsert: mockUpsert, + deleteMany: mockDeleteMany, + }, + }, +})) + +import { + parseDiscoveryFilters, + queryDiscoveryResources, + registerBazaarResource, +} from '../bazaar/catalog' +import type { RegisterBazaarResourceInput } from '../bazaar/types' + +beforeEach(() => { + mockFindMany.mockReset().mockResolvedValue([]) + mockCount.mockReset().mockResolvedValue(0) + mockUpsert.mockReset().mockResolvedValue({}) + mockDeleteMany.mockReset().mockResolvedValue({ count: 0 }) +}) + +describe('parseDiscoveryFilters', () => { + it('defaults limit to 50 and offset to 0', () => { + const filters = parseDiscoveryFilters({}) + expect(filters.limit).toBe(50) + expect(filters.offset).toBe(0) + }) + + it('clamps limit to a maximum of 200', () => { + const filters = parseDiscoveryFilters({ limit: '10000' }) + expect(filters.limit).toBe(200) + }) + + it('rejects a negative or zero limit, falling back to the default', () => { + expect(parseDiscoveryFilters({ limit: '-5' }).limit).toBe(50) + expect(parseDiscoveryFilters({ limit: '0' }).limit).toBe(50) + }) + + it('rejects a negative offset, falling back to 0', () => { + expect(parseDiscoveryFilters({ offset: '-10' }).offset).toBe(0) + }) + + it('passes through a valid offset', () => { + expect(parseDiscoveryFilters({ offset: '25' }).offset).toBe(25) + }) + + it('only accepts "http" or "mcp" for type, dropping anything else', () => { + expect(parseDiscoveryFilters({ type: 'http' }).type).toBe('http') + expect(parseDiscoveryFilters({ type: 'mcp' }).type).toBe('mcp') + expect(parseDiscoveryFilters({ type: 'websocket' }).type).toBeUndefined() + }) + + it('passes through payTo, network, and extensions filters', () => { + const filters = parseDiscoveryFilters({ + payTo: 'GABC', + network: 'stellar:pubnet', + extensions: 'bazaar', + }) + expect(filters.payTo).toBe('GABC') + expect(filters.network).toBe('stellar:pubnet') + expect(filters.extensions).toBe('bazaar') + }) +}) + +describe('queryDiscoveryResources', () => { + it('filters by type', async () => { + await queryDiscoveryResources({ type: 'mcp', limit: 50, offset: 0 }) + expect(mockFindMany).toHaveBeenCalledWith( + expect.objectContaining({ where: expect.objectContaining({ type: 'mcp' }) }) + ) + }) + + it('filters by payTo', async () => { + await queryDiscoveryResources({ payTo: 'GPAY', limit: 50, offset: 0 }) + expect(mockFindMany).toHaveBeenCalledWith( + expect.objectContaining({ where: expect.objectContaining({ payTo: 'GPAY' }) }) + ) + }) + + it('resolves a CAIP-2 network filter to the internal NetworkName', async () => { + await queryDiscoveryResources({ network: 'stellar:pubnet', limit: 50, offset: 0 }) + expect(mockFindMany).toHaveBeenCalledWith( + expect.objectContaining({ where: expect.objectContaining({ network: 'mainnet' }) }) + ) + }) + + it('resolves stellar:testnet to testnet', async () => { + await queryDiscoveryResources({ network: 'stellar:testnet', limit: 50, offset: 0 }) + expect(mockFindMany).toHaveBeenCalledWith( + expect.objectContaining({ where: expect.objectContaining({ network: 'testnet' }) }) + ) + }) + + it('passes through an unrecognized network filter verbatim', async () => { + await queryDiscoveryResources({ network: 'eip155:8453', limit: 50, offset: 0 }) + expect(mockFindMany).toHaveBeenCalledWith( + expect.objectContaining({ where: expect.objectContaining({ network: 'eip155:8453' }) }) + ) + }) + + it('filters by extension key presence', async () => { + await queryDiscoveryResources({ extensions: 'bazaar', limit: 50, offset: 0 }) + expect(mockFindMany).toHaveBeenCalledWith( + expect.objectContaining({ where: expect.objectContaining({ extensionKeys: { has: 'bazaar' } }) }) + ) + }) + + it('applies limit and offset for pagination', async () => { + await queryDiscoveryResources({ limit: 10, offset: 20 }) + expect(mockFindMany).toHaveBeenCalledWith( + expect.objectContaining({ take: 10, skip: 20 }) + ) + }) + + it('orders by createdAt desc with id as a stable tiebreaker', async () => { + await queryDiscoveryResources({ limit: 50, offset: 0 }) + expect(mockFindMany).toHaveBeenCalledWith( + expect.objectContaining({ orderBy: [{ createdAt: 'desc' }, { id: 'desc' }] }) + ) + }) + + it('returns total count alongside the page of resources', async () => { + mockCount.mockResolvedValue(137) + const result = await queryDiscoveryResources({ limit: 50, offset: 0 }) + expect(result.total).toBe(137) + }) + + it('maps a stored row back into the spec resource/accepts/extensions shape', async () => { + mockFindMany.mockResolvedValue([ + { + url: 'https://lens.example/price', + description: 'Unified price feed', + mimeType: 'application/json', + serviceName: 'Lens', + tags: ['price', 'stellar'], + iconUrl: 'https://lens.example/icon.png', + accepts: [{ scheme: 'exact', network: 'stellar:pubnet', amount: '100000', asset: 'USDC', payTo: 'GPAY', maxTimeoutSeconds: 60 }], + bazaarInfo: { input: { type: 'http', method: 'GET' } }, + bazaarSchema: { type: 'object' }, + routeTemplate: null, + extensionKeys: ['bazaar'], + }, + ]) + + const result = await queryDiscoveryResources({ limit: 50, offset: 0 }) + expect(result.resources).toHaveLength(1) + const listing = result.resources[0] + expect(listing.resource.url).toBe('https://lens.example/price') + expect(listing.resource.serviceName).toBe('Lens') + expect(listing.accepts[0].payTo).toBe('GPAY') + expect(listing.extensions.bazaar.info).toEqual({ input: { type: 'http', method: 'GET' } }) + expect(listing.extensions.bazaar).not.toHaveProperty('routeTemplate') + }) + + it('includes routeTemplate when present', async () => { + mockFindMany.mockResolvedValue([ + { + url: 'https://lens.example/users/123', + description: null, + mimeType: null, + serviceName: null, + tags: [], + iconUrl: null, + accepts: [], + bazaarInfo: { input: { type: 'http', method: 'GET' } }, + bazaarSchema: {}, + routeTemplate: '/users/:userId', + extensionKeys: ['bazaar'], + }, + ]) + + const result = await queryDiscoveryResources({ limit: 50, offset: 0 }) + expect(result.resources[0].extensions.bazaar.routeTemplate).toBe('/users/:userId') + }) +}) + +describe('registerBazaarResource', () => { + const httpInput: RegisterBazaarResourceInput = { + type: 'http', + network: 'mainnet', + resource: { url: 'https://lens.example/price' }, + accepts: [{ scheme: 'exact', network: 'stellar:pubnet', amount: '100000', asset: 'USDC', payTo: 'GPAY', maxTimeoutSeconds: 60 }], + bazaar: { info: { input: { type: 'http', method: 'GET' } }, schema: { type: 'object' } }, + } + + const mcpInput: RegisterBazaarResourceInput = { + type: 'mcp', + network: 'testnet', + resource: { url: 'https://lens.example/mcp' }, + accepts: [{ scheme: 'exact', network: 'stellar:testnet', amount: '100000', asset: 'USDC', payTo: 'GPAY2', maxTimeoutSeconds: 60 }], + bazaar: { + info: { input: { type: 'mcp', toolName: 'financial_analysis', inputSchema: { type: 'object' } } }, + schema: { type: 'object' }, + }, + } + + it('upserts an HTTP resource keyed on (network, url, httpMethod)', async () => { + await registerBazaarResource(httpInput) + expect(mockUpsert).toHaveBeenCalledWith( + expect.objectContaining({ + where: { bazaarHttpIdentity: { network: 'mainnet', url: 'https://lens.example/price', httpMethod: 'GET' } }, + }) + ) + }) + + it('upserts an MCP resource keyed on (network, resource.url, input.toolName)', async () => { + await registerBazaarResource(mcpInput) + expect(mockUpsert).toHaveBeenCalledWith( + expect.objectContaining({ + where: { bazaarMcpIdentity: { network: 'testnet', url: 'https://lens.example/mcp', mcpToolName: 'financial_analysis' } }, + }) + ) + }) + + it('rejects registration when accepts[] is empty', async () => { + await expect( + registerBazaarResource({ ...httpInput, accepts: [] }) + ).rejects.toThrow(/payTo/) + expect(mockUpsert).not.toHaveBeenCalled() + }) + + it('denormalizes payTo from accepts[0] onto the row', async () => { + await registerBazaarResource(httpInput) + expect(mockUpsert).toHaveBeenCalledWith( + expect.objectContaining({ create: expect.objectContaining({ payTo: 'GPAY' }) }) + ) + }) + + it('always includes "bazaar" in extensionKeys by default', async () => { + await registerBazaarResource(httpInput) + expect(mockUpsert).toHaveBeenCalledWith( + expect.objectContaining({ create: expect.objectContaining({ extensionKeys: ['bazaar'] }) }) + ) + }) +}) diff --git a/src/__tests__/bestRoute.test.ts b/src/__tests__/bestRoute.test.ts index e1a793f..aa445d4 100644 --- a/src/__tests__/bestRoute.test.ts +++ b/src/__tests__/bestRoute.test.ts @@ -1,5 +1,5 @@ import { vi, describe, it, expect, beforeEach } from 'vitest' -import { getBestRoute } from '../aggregator/bestRoute' +import { getBestRoute, _resetHorizonServers } from '../aggregator/bestRoute' import { pgPool } from '../db' import * as StellarSdk from '@stellar/stellar-sdk' @@ -10,9 +10,11 @@ vi.mock('../db', () => ({ } })) -vi.mock('@stellar/stellar-sdk', () => { +vi.mock('@stellar/stellar-sdk', async (importOriginal) => { + const actual = await importOriginal() const callFn = vi.fn() return { + ...actual, Horizon: { Server: vi.fn(function() { return { @@ -25,6 +27,13 @@ vi.mock('@stellar/stellar-sdk', () => { vi.fn(function(code, issuer) { return { code, issuer } }), { native: vi.fn(() => 'native') } ), + // config.ts's buildNetworkConfig() falls back to these when no + // NETWORK_PASSPHRASE_* env var is set — needed now that getBestRoute + // resolves a per-network Horizon client via getNetworkConfig(). + Networks: { + PUBLIC: 'Public Global Stellar Network ; September 2015', + TESTNET: 'Test SDF Network ; September 2015', + }, __mockCall: callFn } }) @@ -39,6 +48,9 @@ describe('getBestRoute', () => { beforeEach(() => { vi.clearAllMocks() + // horizonServers is memoised at module scope (see bestRoute.ts) — clear + // between tests so each one observes fresh Horizon.Server() constructions. + _resetHorizonServers() }) it('Case 1: returns SDEX when SDEX price is better', async () => { @@ -107,7 +119,7 @@ describe('getBestRoute', () => { mockCall.mockResolvedValue({ records: [{ destination_amount: '123.456789' }] // 123.456789 / 1000 = 0.123456789 }) - + // AMM: no pool data to simplify test or give known value mockQuery.mockResolvedValue({ rows: [] } as any) @@ -115,4 +127,30 @@ describe('getBestRoute', () => { expect(result.sdexPrice).toBeCloseTo(0.123457, 6) }) + + it('Case 6: queries the mainnet Horizon server when network="mainnet"', async () => { + mockCall.mockResolvedValue({ records: [{ destination_amount: '500' }] }) + mockQuery.mockResolvedValue({ rows: [] } as any) + + await getBestRoute(assetA, assetB, pairKey, 1000, 'mainnet') + + const HorizonServerCtor = (StellarSdk as any).Horizon.Server + const urls = HorizonServerCtor.mock.calls.map((call: unknown[]) => call[0]) + expect(urls.some((url: string) => url.includes('horizon.stellar.org'))).toBe(true) + expect(urls.some((url: string) => url.includes('testnet'))).toBe(false) + }) + + it('Case 7: testnet and mainnet reuse a memoised Horizon server per network', async () => { + mockCall.mockResolvedValue({ records: [{ destination_amount: '500' }] }) + mockQuery.mockResolvedValue({ rows: [] } as any) + + const HorizonServerCtor = (StellarSdk as any).Horizon.Server + const callsBefore = HorizonServerCtor.mock.calls.length + + await getBestRoute(assetA, assetB, pairKey, 1000, 'mainnet') + await getBestRoute(assetA, assetB, pairKey, 1000, 'mainnet') + + // Second mainnet call reuses the cached client — only one new Server() call. + expect(HorizonServerCtor.mock.calls.length).toBe(callsBefore + 1) + }) }) diff --git a/src/__tests__/discovery.test.ts b/src/__tests__/discovery.test.ts new file mode 100644 index 0000000..3871d36 --- /dev/null +++ b/src/__tests__/discovery.test.ts @@ -0,0 +1,108 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest' +import Fastify from 'fastify' + +const { mockQueryDiscoveryResources, mockParseDiscoveryFilters } = vi.hoisted(() => ({ + mockQueryDiscoveryResources: vi.fn(), + mockParseDiscoveryFilters: vi.fn(), +})) + +vi.mock('../bazaar/catalog', () => ({ + queryDiscoveryResources: mockQueryDiscoveryResources, + parseDiscoveryFilters: mockParseDiscoveryFilters, +})) + +import { registerDiscoveryRoutes } from '../routes/discovery' + +async function buildApp() { + const app = Fastify({ logger: false }) + await registerDiscoveryRoutes(app) + await app.ready() + return app +} + +beforeEach(() => { + mockParseDiscoveryFilters.mockReset().mockImplementation((q: any) => ({ + type: q.type, + payTo: q.payTo, + network: q.network, + extensions: q.extensions, + limit: q.limit ? Number(q.limit) : 50, + offset: q.offset ? Number(q.offset) : 0, + })) + mockQueryDiscoveryResources.mockReset().mockResolvedValue({ + resources: [], + limit: 50, + offset: 0, + total: 0, + }) +}) + +describe('GET /discovery/resources', () => { + it('returns 200 without any auth header — discovery is public', async () => { + const app = await buildApp() + const res = await app.inject({ method: 'GET', url: '/discovery/resources' }) + expect(res.statusCode).toBe(200) + }) + + it('returns the resources/limit/offset/total envelope', async () => { + const app = await buildApp() + const res = await app.inject({ method: 'GET', url: '/discovery/resources' }) + const body = res.json() + expect(body).toHaveProperty('resources') + expect(body).toHaveProperty('limit') + expect(body).toHaveProperty('offset') + expect(body).toHaveProperty('total') + }) + + it('forwards the type filter from the query string', async () => { + const app = await buildApp() + await app.inject({ method: 'GET', url: '/discovery/resources?type=mcp' }) + expect(mockQueryDiscoveryResources).toHaveBeenCalledWith(expect.objectContaining({ type: 'mcp' })) + }) + + it('forwards the payTo filter from the query string', async () => { + const app = await buildApp() + await app.inject({ method: 'GET', url: '/discovery/resources?payTo=GABC123' }) + expect(mockQueryDiscoveryResources).toHaveBeenCalledWith(expect.objectContaining({ payTo: 'GABC123' })) + }) + + it('forwards the network filter from the query string', async () => { + const app = await buildApp() + await app.inject({ method: 'GET', url: '/discovery/resources?network=stellar:pubnet' }) + expect(mockQueryDiscoveryResources).toHaveBeenCalledWith(expect.objectContaining({ network: 'stellar:pubnet' })) + }) + + it('forwards the extensions filter from the query string', async () => { + const app = await buildApp() + await app.inject({ method: 'GET', url: '/discovery/resources?extensions=bazaar' }) + expect(mockQueryDiscoveryResources).toHaveBeenCalledWith(expect.objectContaining({ extensions: 'bazaar' })) + }) + + it('forwards limit and offset from the query string', async () => { + const app = await buildApp() + await app.inject({ method: 'GET', url: '/discovery/resources?limit=10&offset=20' }) + expect(mockQueryDiscoveryResources).toHaveBeenCalledWith(expect.objectContaining({ limit: 10, offset: 20 })) + }) + + it('returns resources returned by the catalog query verbatim', async () => { + mockQueryDiscoveryResources.mockResolvedValue({ + resources: [ + { + resource: { url: 'https://lens.example/price' }, + accepts: [{ scheme: 'exact', network: 'stellar:pubnet', amount: '100000', asset: 'USDC', payTo: 'GPAY', maxTimeoutSeconds: 60 }], + extensions: { bazaar: { info: { input: { type: 'http', method: 'GET' } }, schema: {} } }, + }, + ], + limit: 50, + offset: 0, + total: 1, + }) + + const app = await buildApp() + const res = await app.inject({ method: 'GET', url: '/discovery/resources' }) + const body = res.json() + expect(body.resources).toHaveLength(1) + expect(body.total).toBe(1) + expect(body.resources[0].resource.url).toBe('https://lens.example/price') + }) +}) diff --git a/src/__tests__/middleware/network.test.ts b/src/__tests__/middleware/network.test.ts new file mode 100644 index 0000000..46df83c --- /dev/null +++ b/src/__tests__/middleware/network.test.ts @@ -0,0 +1,61 @@ +import { describe, it, expect } from 'vitest' +import Fastify from 'fastify' +import { registerNetworkSelector, resolveNetworkName } from '../../middleware/network' + +describe('resolveNetworkName', () => { + it('defaults to activeNetwork (testnet) when absent', () => { + expect(resolveNetworkName(undefined)).toEqual({ ok: true, network: 'testnet' }) + expect(resolveNetworkName(null)).toEqual({ ok: true, network: 'testnet' }) + expect(resolveNetworkName('')).toEqual({ ok: true, network: 'testnet' }) + }) + + it('accepts "testnet" and "mainnet", case-insensitively', () => { + expect(resolveNetworkName('mainnet')).toEqual({ ok: true, network: 'mainnet' }) + expect(resolveNetworkName('MAINNET')).toEqual({ ok: true, network: 'mainnet' }) + expect(resolveNetworkName(' testnet ')).toEqual({ ok: true, network: 'testnet' }) + }) + + it('rejects an unrecognised value', () => { + const result = resolveNetworkName('pubnet') + expect(result.ok).toBe(false) + if (!result.ok) expect(result.error).toMatch(/Invalid network "pubnet"/) + }) +}) + +async function buildApp() { + const app = Fastify({ logger: false }) + await app.register(registerNetworkSelector) + app.get('/echo', async (req) => ({ network: req.network })) + await app.ready() + return app +} + +describe('registerNetworkSelector', () => { + it('defaults req.network to testnet when no network is specified', async () => { + const app = await buildApp() + const res = await app.inject({ method: 'GET', url: '/echo' }) + expect(res.statusCode).toBe(200) + expect(res.json()).toEqual({ network: 'testnet' }) + }) + + it('resolves req.network from the ?network= query param', async () => { + const app = await buildApp() + const res = await app.inject({ method: 'GET', url: '/echo?network=mainnet' }) + expect(res.statusCode).toBe(200) + expect(res.json()).toEqual({ network: 'mainnet' }) + }) + + it('resolves req.network from the x-network header', async () => { + const app = await buildApp() + const res = await app.inject({ method: 'GET', url: '/echo', headers: { 'x-network': 'mainnet' } }) + expect(res.statusCode).toBe(200) + expect(res.json()).toEqual({ network: 'mainnet' }) + }) + + it('rejects an invalid network with 400', async () => { + const app = await buildApp() + const res = await app.inject({ method: 'GET', url: '/echo?network=pubnet' }) + expect(res.statusCode).toBe(400) + expect(res.json()).toHaveProperty('error') + }) +}) diff --git a/src/__tests__/middleware/x402.test.ts b/src/__tests__/middleware/x402.test.ts index 892a3a3..ab966cd 100644 --- a/src/__tests__/middleware/x402.test.ts +++ b/src/__tests__/middleware/x402.test.ts @@ -47,6 +47,8 @@ vi.mock('@x402/stellar/exact/server', () => ({ import Fastify from 'fastify' import { registerX402 } from '../../middleware/x402' +import { registerNetworkSelector } from '../../middleware/network' +import { _resetX402ResourceServers } from '../../x402/network' // ── Helpers ─────────────────────────────────────────────────────────────────── async function buildApp() { @@ -63,6 +65,18 @@ async function buildApp() { return app } +// Same as buildApp(), but with the network selector registered ahead of x402 +// so req.network is actually resolved from ?network=/x-network per request. +async function buildAppWithNetworkSelector() { + process.env.ORACLE_PAYMENT_ADDRESS = PAYMENT_ADDRESS + const app = Fastify({ logger: false }) + await app.register(registerNetworkSelector) + await app.register(registerX402) + app.get('/price/test', async () => ({ ok: true })) + await app.ready() + return app +} + function makePaymentHeader(overrides: Record = {}): string { const payload = { scheme: 'exact', amount: '$0.10', recipient: PAYMENT_ADDRESS, ...overrides } return Buffer.from(JSON.stringify(payload)).toString('base64') @@ -73,6 +87,12 @@ beforeEach(() => { mockSettle.mockReset().mockResolvedValue(undefined) mockInitialize.mockReset().mockResolvedValue(undefined) mockRegisterChain.register.mockReturnValue(mockRegisterChain) + // Per-network resource servers are memoised at module scope (see + // x402/network.ts) — clear between tests so each one builds fresh against + // whatever ORACLE_PAYMENT_ADDRESS_* env vars it sets up. + _resetX402ResourceServers() + delete process.env.ORACLE_PAYMENT_ADDRESS_MAINNET + delete process.env.ORACLE_PAYMENT_ADDRESS_TESTNET }) // ── Tests ───────────────────────────────────────────────────────────────────── @@ -214,3 +234,60 @@ describe('x402 middleware', () => { expect(mockVerify).not.toHaveBeenCalled() }) }) + +describe('x402 middleware — per-request network', () => { + it('defaults to testnet requirements when no network is requested', async () => { + const app = await buildAppWithNetworkSelector() + + const res = await app.inject({ method: 'GET', url: '/price/test' }) + + expect(res.statusCode).toBe(402) + expect(res.json().accepts[0]).toMatchObject({ network: 'stellar:testnet', payTo: PAYMENT_ADDRESS }) + }) + + it('resolves mainnet network/payTo from ?network=mainnet', async () => { + const MAINNET_ADDRESS = 'GMAINNETADDRESS123456789012345678901234567890123456789012' + process.env.ORACLE_PAYMENT_ADDRESS_MAINNET = MAINNET_ADDRESS + const app = await buildAppWithNetworkSelector() + + const res = await app.inject({ method: 'GET', url: '/price/test?network=mainnet' }) + + expect(res.statusCode).toBe(402) + expect(res.json().accepts[0]).toMatchObject({ network: 'stellar:pubnet', payTo: MAINNET_ADDRESS }) + }) + + it('falls back to the shared ORACLE_PAYMENT_ADDRESS when no mainnet-specific address is set', async () => { + const app = await buildAppWithNetworkSelector() + + const res = await app.inject({ method: 'GET', url: '/price/test?network=mainnet' }) + + expect(res.statusCode).toBe(402) + expect(res.json().accepts[0]).toMatchObject({ network: 'stellar:pubnet', payTo: PAYMENT_ADDRESS }) + }) + + it('rejects an invalid ?network= before x402 even runs', async () => { + const app = await buildAppWithNetworkSelector() + + const res = await app.inject({ method: 'GET', url: '/price/test?network=pubnet' }) + + expect(res.statusCode).toBe(400) + expect(mockVerify).not.toHaveBeenCalled() + }) + + it('verifies a mainnet payment against mainnet requirements', async () => { + mockVerify.mockResolvedValue({ isValid: true }) + const app = await buildAppWithNetworkSelector() + + const res = await app.inject({ + method: 'GET', + url: '/price/test?network=mainnet', + headers: { 'x-payment': makePaymentHeader() }, + }) + + expect(res.statusCode).toBe(200) + expect(mockVerify).toHaveBeenCalledWith( + expect.anything(), + expect.objectContaining({ network: 'stellar:pubnet' }) + ) + }) +}) diff --git a/src/__tests__/networkClients.test.ts b/src/__tests__/networkClients.test.ts new file mode 100644 index 0000000..0d09246 --- /dev/null +++ b/src/__tests__/networkClients.test.ts @@ -0,0 +1,48 @@ +/** + * Unit tests for the per-network Horizon / Soroban RPC client factories. + */ + +vi.mock('@stellar/stellar-sdk', async (importOriginal) => { + const actual = await importOriginal() + return { + ...actual, + Horizon: { Server: vi.fn(function (url: string) { return { __url: url } }) }, + rpc: { Server: vi.fn(function (url: string) { return { __url: url } }) }, + } +}) + +import { getHorizonServer, getRpcServer } from '../network/clients' + +describe('getHorizonServer', () => { + it('returns distinct clients for different networks', () => { + const testnet = getHorizonServer('testnet') + const mainnet = getHorizonServer('mainnet') + + expect(testnet).not.toBe(mainnet) + expect((testnet as any).__url).toContain('testnet') + expect((mainnet as any).__url).not.toContain('testnet') + }) + + it('returns the same cached client for repeated calls on the same network', () => { + const first = getHorizonServer('testnet') + const second = getHorizonServer('testnet') + + expect(first).toBe(second) + }) +}) + +describe('getRpcServer', () => { + it('returns distinct clients for different networks', () => { + const testnet = getRpcServer('testnet') + const mainnet = getRpcServer('mainnet') + + expect(testnet).not.toBe(mainnet) + }) + + it('returns the same cached client for repeated calls on the same network', () => { + const first = getRpcServer('mainnet') + const second = getRpcServer('mainnet') + + expect(first).toBe(second) + }) +}) diff --git a/src/__tests__/networkVenueConfig.test.ts b/src/__tests__/networkVenueConfig.test.ts new file mode 100644 index 0000000..67e0644 --- /dev/null +++ b/src/__tests__/networkVenueConfig.test.ts @@ -0,0 +1,116 @@ +/** + * Unit tests for per-network venue configuration (Soroswap / Aquarius / Reflector). + * + * Each test resets modules and re-imports `../config` after mutating + * `process.env` so the lazy per-network cache in config.ts is rebuilt from + * the env vars set for that test. + */ + +const ENV_KEYS = [ + 'STELLAR_NETWORK', + 'SOROSWAP_ENABLED_TESTNET', + 'SOROSWAP_ENABLED_MAINNET', + 'SOROSWAP_TOKEN_LIST_URL', + 'SOROSWAP_TOKEN_LIST_URL_TESTNET', + 'AQUARIUS_ENABLED_TESTNET', + 'AQUARIUS_ENABLED_MAINNET', + 'AQUARIUS_API_URL', + 'REFLECTOR_CONTRACT_ID_TESTNET', + 'REFLECTOR_CONTRACT_ID_MAINNET', + 'REFLECTOR_ENABLED_TESTNET', +] + +async function loadConfig() { + vi.resetModules() + return await import('../config') +} + +describe('per-network venue config', () => { + const originalEnv: Record = {} + + beforeEach(() => { + for (const key of ENV_KEYS) originalEnv[key] = process.env[key] + }) + + afterEach(() => { + for (const key of ENV_KEYS) { + if (originalEnv[key] === undefined) delete process.env[key] + else process.env[key] = originalEnv[key] + } + }) + + it('defaults Aquarius to disabled on testnet and enabled on mainnet', async () => { + delete process.env.AQUARIUS_ENABLED_TESTNET + delete process.env.AQUARIUS_ENABLED_MAINNET + + const { getNetworkConfig } = await loadConfig() + + expect(getNetworkConfig('testnet').aquarius.enabled).toBe(false) + expect(getNetworkConfig('mainnet').aquarius.enabled).toBe(true) + }) + + it('respects an explicit AQUARIUS_ENABLED_TESTNET=true override', async () => { + process.env.AQUARIUS_ENABLED_TESTNET = 'true' + + const { getNetworkConfig } = await loadConfig() + + expect(getNetworkConfig('testnet').aquarius.enabled).toBe(true) + }) + + it('defaults Soroswap to enabled on both networks', async () => { + delete process.env.SOROSWAP_ENABLED_TESTNET + delete process.env.SOROSWAP_ENABLED_MAINNET + + const { getNetworkConfig } = await loadConfig() + + expect(getNetworkConfig('testnet').soroswap.enabled).toBe(true) + expect(getNetworkConfig('mainnet').soroswap.enabled).toBe(true) + }) + + it('disables Soroswap on a network when explicitly set to false', async () => { + process.env.SOROSWAP_ENABLED_TESTNET = 'false' + + const { getNetworkConfig } = await loadConfig() + + expect(getNetworkConfig('testnet').soroswap.enabled).toBe(false) + }) + + it('resolves a per-network token-list URL override before falling back to the shared default', async () => { + delete process.env.SOROSWAP_TOKEN_LIST_URL + process.env.SOROSWAP_TOKEN_LIST_URL_TESTNET = 'https://example.com/testnet-tokens.json' + + const { getNetworkConfig } = await loadConfig() + + expect(getNetworkConfig('testnet').soroswap.tokenListUrl).toBe( + 'https://example.com/testnet-tokens.json' + ) + expect(getNetworkConfig('mainnet').soroswap.tokenListUrl).toBe( + 'https://raw.githubusercontent.com/soroswap/token-list/main/tokenList.json' + ) + }) + + it('disables the Reflector oracle when no contract id is configured for the network', async () => { + delete process.env.REFLECTOR_CONTRACT_ID_TESTNET + + const { getNetworkConfig } = await loadConfig() + + expect(getNetworkConfig('testnet').oracle.reflectorContractId).toBe('') + expect(getNetworkConfig('testnet').oracle.enabled).toBe(false) + }) + + it('enables the Reflector oracle on mainnet where a default contract id exists', async () => { + const { getNetworkConfig } = await loadConfig() + + expect(getNetworkConfig('mainnet').oracle.reflectorContractId).not.toBe('') + expect(getNetworkConfig('mainnet').oracle.enabled).toBe(true) + }) + + it('resolves a custom Aquarius API URL override', async () => { + process.env.AQUARIUS_API_URL = 'https://example.com/aquarius/' + + const { getNetworkConfig } = await loadConfig() + + expect(getNetworkConfig('testnet').aquarius.apiUrl).toBe('https://example.com/aquarius/') + expect(getNetworkConfig('mainnet').aquarius.apiUrl).toBe('https://example.com/aquarius/') + }) +}) diff --git a/src/__tests__/price.test.ts b/src/__tests__/price.test.ts index 177ab3e..bf3c408 100644 --- a/src/__tests__/price.test.ts +++ b/src/__tests__/price.test.ts @@ -20,17 +20,25 @@ vi.mock('../aggregator/bestRoute', () => ({ getBestRoute: mockGetBestRoute, })) +const { testnetPairs } = vi.hoisted(() => ({ + testnetPairs: [ + { + pairKey: 'USDC/XLM', + assetA: { code: 'XLM', issuer: null }, + assetB: { code: 'USDC', issuer: 'GBBD47IF6LWK7P7MDEVSCWR7DPUWV3NY3DTQEVFL4NAT4AQH3ZLLFLA5' }, + }, + ], +})) + vi.mock('../config', () => ({ config: { - pairs: [ - { - pairKey: 'USDC/XLM', - assetA: { code: 'XLM', issuer: null }, - assetB: { code: 'USDC', issuer: 'GBBD47IF6LWK7P7MDEVSCWR7DPUWV3NY3DTQEVFL4NAT4AQH3ZLLFLA5' } - }, - ], + pairs: testnetPairs, cache: { priceTtl: 10 }, }, + activeNetwork: 'testnet', + getNetworkConfig: (network: string) => ({ + pairs: network === 'testnet' ? testnetPairs : [], + }), })) import { registerRESTRoutes } from '../api/rest' diff --git a/src/__tests__/schemaValidation.test.ts b/src/__tests__/schemaValidation.test.ts index 35fe76b..12ec5b4 100644 --- a/src/__tests__/schemaValidation.test.ts +++ b/src/__tests__/schemaValidation.test.ts @@ -31,17 +31,25 @@ vi.mock('../pricing/depth', () => ({ getDepth: mockGetDepth, })) +const { schemaTestPairs } = vi.hoisted(() => ({ + schemaTestPairs: [ + { + pairKey: 'USDC/XLM', + assetA: { code: 'XLM', issuer: null }, + assetB: { code: 'USDC', issuer: 'GBBD47IF6LWK7P7MDEVSCWR7DPUWV3NY3DTQEVFL4NAT4AQH3ZLLFLA5' }, + }, + ], +})) + vi.mock('../config', () => ({ config: { - pairs: [ - { - pairKey: 'USDC/XLM', - assetA: { code: 'XLM', issuer: null }, - assetB: { code: 'USDC', issuer: 'GBBD47IF6LWK7P7MDEVSCWR7DPUWV3NY3DTQEVFL4NAT4AQH3ZLLFLA5' }, - }, - ], + pairs: schemaTestPairs, cache: { priceTtl: 10 }, }, + activeNetwork: 'testnet', + getNetworkConfig: (network: string) => ({ + pairs: network === 'testnet' ? schemaTestPairs : [], + }), })) import { registerRESTRoutes } from '../api/rest' diff --git a/src/__tests__/snapshotIngester.test.ts b/src/__tests__/snapshotIngester.test.ts index fc4e965..e3ff168 100644 --- a/src/__tests__/snapshotIngester.test.ts +++ b/src/__tests__/snapshotIngester.test.ts @@ -82,6 +82,6 @@ describe('appendSnapshots', () => { await appendSnapshots() - expect(mockQuery.mock.calls[0][0]).toMatch(/ON CONFLICT \(pair, ts\) DO NOTHING/) + expect(mockQuery.mock.calls[0][0]).toMatch(/ON CONFLICT \(network, pair, ts\) DO NOTHING/) }) }) diff --git a/src/__tests__/snapshotRetention.test.ts b/src/__tests__/snapshotRetention.test.ts index 4e17e18..fc40730 100644 --- a/src/__tests__/snapshotRetention.test.ts +++ b/src/__tests__/snapshotRetention.test.ts @@ -15,6 +15,7 @@ vi.mock('bullmq', () => ({ })) import { pruneOldSnapshots, SNAPSHOT_RETENTION_DAYS } from '../jobs/snapshotRetention' +import { activeNetwork } from '../config' describe('pruneOldSnapshots', () => { beforeEach(() => { @@ -28,9 +29,9 @@ describe('pruneOldSnapshots', () => { expect(SNAPSHOT_RETENTION_DAYS).toBe(30) expect(pruned).toBe(5) - expect(mockQuery.mock.calls[0][1]).toEqual([30]) + expect(mockQuery.mock.calls[0][1]).toEqual([activeNetwork, 30]) expect(mockQuery.mock.calls[0][0]).toMatch(/DELETE FROM price_snapshots/) - expect(mockQuery.mock.calls[0][0]).toMatch(/ts < NOW\(\) - \(\$1 \|\| ' days'\)::interval/) + expect(mockQuery.mock.calls[0][0]).toMatch(/ts < NOW\(\) - \(\$2 \|\| ' days'\)::interval/) }) it('honors a custom retention window', async () => { @@ -39,7 +40,7 @@ describe('pruneOldSnapshots', () => { const pruned = await pruneOldSnapshots(7) expect(pruned).toBe(0) - expect(mockQuery.mock.calls[0][1]).toEqual([7]) + expect(mockQuery.mock.calls[0][1]).toEqual([activeNetwork, 7]) }) it('returns 0 when rowCount is null', async () => { diff --git a/src/__tests__/soroswapEnabled.test.ts b/src/__tests__/soroswapEnabled.test.ts new file mode 100644 index 0000000..14edb9b --- /dev/null +++ b/src/__tests__/soroswapEnabled.test.ts @@ -0,0 +1,51 @@ +/** + * Verifies the Soroswap ingester respects the per-network enable flag, + * skipping the polling loop entirely when Soroswap has no usable deployment + * on the active network. + */ + +const mocks = vi.hoisted(() => ({ + config: { + soroswap: { + enabled: true, + factoryAddress: 'CFACTORY', + tokenListUrl: 'https://example.com/tokens.json', + pollIntervalMs: 60000, + }, + network: { passphrase: 'Test SDF Network ; September 2015' }, + rpc: { url: 'https://soroban-testnet.stellar.org' }, + }, + pairsRegistry: { + getActivePairs: vi.fn().mockReturnValue([]), + }, +})) + +// The ingester now takes `network` (defaulting to `activeNetwork`) and reads +// via getNetworkConfig, so the mock has to supply both. getNetworkConfig +// returns the same object the test mutates, keeping `enabled` toggleable. +vi.mock('../config', () => ({ + config: mocks.config, + activeNetwork: 'testnet', + getNetworkConfig: () => mocks.config, +})) +vi.mock('../pairsRegistry', () => mocks.pairsRegistry) +vi.mock('../db', () => ({ upsertPricePoints: vi.fn().mockResolvedValue(undefined) })) +vi.mock('../webhookDispatcher', () => ({ dispatchPriceUpdate: vi.fn().mockResolvedValue(undefined) })) + +import { startSoroswapIngester } from '../ingesters/soroswap' + +describe('startSoroswapIngester', () => { + beforeEach(() => { + vi.clearAllMocks() + }) + + it('does not start the polling loop when Soroswap is disabled on the active network', async () => { + mocks.config.soroswap.enabled = false + + await startSoroswapIngester() + + expect(mocks.pairsRegistry.getActivePairs).not.toHaveBeenCalled() + + mocks.config.soroswap.enabled = true + }) +}) diff --git a/src/__tests__/soroswapIngester.test.ts b/src/__tests__/soroswapIngester.test.ts index 5cedc72..2bc06ae 100644 --- a/src/__tests__/soroswapIngester.test.ts +++ b/src/__tests__/soroswapIngester.test.ts @@ -164,7 +164,8 @@ describe('ingestPair', () => { expect(mockFetchPools).toHaveBeenCalledWith( factory, xlmToken.address, - usdcToken.address + usdcToken.address, + expect.any(String) ) }) diff --git a/src/aggregator/bestRoute.ts b/src/aggregator/bestRoute.ts index 5b11639..8e19d0e 100644 --- a/src/aggregator/bestRoute.ts +++ b/src/aggregator/bestRoute.ts @@ -1,15 +1,17 @@ -import { Horizon, Asset } from '@stellar/stellar-sdk' -import { config } from '../config' +import { Asset } from '@stellar/stellar-sdk' +import { activeNetwork, type NetworkName } from '../config' +import { getHorizonServer, resetNetworkClients } from '../network/clients' import type { AssetId, RouteInfo } from '../types' import { pgPool } from '../db' -const horizonServer = new Horizon.Server(config.horizon.url) - function assetIdToStellar(asset: AssetId) { if (!asset.issuer) return Asset.native() return new Asset(asset.code, asset.issuer) } +// AMM pricing reads price_points/pool_snapshots, which have no network column +// yet — that is the deeper aggregation-layer work tracked separately. SDEX +// pricing is a live Horizon call, so it is genuinely per-network today. async function getAMMPrice(pairKey: string, amount: number): Promise { // Get latest pool snapshot via pool_id (pairKey indexes price_points correctly) const result = await pgPool.query( @@ -36,11 +38,25 @@ async function getAMMPrice(pairKey: string, amount: number): Promise { return output / amount // price per unit } -async function getSDEXPrice(assetA: AssetId, assetB: AssetId, amount: number): Promise { +/** + * Test-only: clears the memoised per-network Horizon clients between cases. + * Kept as a re-export so existing tests keep their import path; the clients + * themselves now live in network/clients.ts. + */ +export function _resetHorizonServers(): void { + resetNetworkClients() +} + +async function getSDEXPrice( + assetA: AssetId, + assetB: AssetId, + amount: number, + network: NetworkName +): Promise { try { const stellarAssetA = assetIdToStellar(assetA) const stellarAssetB = assetIdToStellar(assetB) - const paths = await horizonServer + const paths = await getHorizonServer(network) .strictSendPaths(stellarAssetA, amount.toString(), [stellarAssetB]) .call() if (paths.records.length === 0) return 0 @@ -55,10 +71,11 @@ export async function getBestRoute( assetA: AssetId, assetB: AssetId, pairKey: string, - amount: number = 1000 + amount: number = 1000, + network: NetworkName = activeNetwork ): Promise { const [sdexPrice, ammPrice] = await Promise.all([ - getSDEXPrice(assetA, assetB, amount), + getSDEXPrice(assetA, assetB, amount, network), getAMMPrice(pairKey, amount), ]) diff --git a/src/api/history.ts b/src/api/history.ts index 9c6c43e..0da70a9 100644 --- a/src/api/history.ts +++ b/src/api/history.ts @@ -1,5 +1,6 @@ import type { FastifyInstance } from 'fastify' import { pgPool } from '../db' +import { activeNetwork } from '../config' /** Supported aggregation intervals → bucket width in seconds. */ export const HISTORY_INTERVAL_SECONDS: Record = { @@ -37,12 +38,12 @@ export async function queryHistory( (array_agg(price::float ORDER BY ts DESC))[1] AS price, SUM(volume::float) AS volume FROM price_snapshots - WHERE pair = $2 + WHERE pair = $2 AND network = $5 AND ts >= $3 AND ts <= $4 GROUP BY floor(EXTRACT(EPOCH FROM ts) / $1) ORDER BY bucket ASC`, - [intervalSecs, pair, from, to] + [intervalSecs, pair, from, to, activeNetwork] ) return result.rows.map(r => ({ diff --git a/src/api/rest.ts b/src/api/rest.ts index 8bb8fff..b782e66 100644 --- a/src/api/rest.ts +++ b/src/api/rest.ts @@ -4,7 +4,8 @@ import { getCachedPrice, setCachedPrice } from '../redis' import { getAggregatedPrice } from '../aggregator/vwap' import { getBestRoute } from '../aggregator/bestRoute' import { pgPool } from '../db' -import { config } from '../config' +import { config, getNetworkConfig, activeNetwork, type NetworkName } from '../config' +import '../middleware/network' // declares req.network on the FastifyRequest type import { statusResponseSchema, priceResponseSchema, @@ -20,11 +21,11 @@ function makePairKey(a: string, b: string): string { return [a, b].sort().join('/') } -function findPair(assetA: string, assetB: string) { +function findPair(assetA: string, assetB: string, network: NetworkName) { const normalize = (a: string) => a.toLowerCase() === 'native' ? 'XLM' : a.split(':')[0].toUpperCase() const cA = normalize(assetA) const cB = normalize(assetB) - return config.pairs.find(p => { + return getNetworkConfig(network).pairs.find(p => { const pA = p.assetA.code.toUpperCase() const pB = p.assetB.code.toUpperCase() return (cA === pA && cB === pB) || (cA === pB && cB === pA) @@ -58,10 +59,14 @@ export async function registerRESTRoutes(app: FastifyInstance) { async (req, reply) => { price_requests_total.inc() const { assetA, assetB } = req.params - const pair = findPair(assetA, assetB) - if (!pair) return reply.status(404).send({ error: `Pair ${assetA}/${assetB} not watched` }) - - const cached = await getCachedPrice(pair.pairKey) + const network = req.network ?? activeNetwork + const pair = findPair(assetA, assetB, network) + if (!pair) return reply.status(404).send({ error: `Pair ${assetA}/${assetB} not watched on ${network}` }) + + // Cache key is network-scoped so testnet/mainnet prices for the same + // asset codes never collide. + const cacheKey = `${network}:${pair.pairKey}` + const cached = await getCachedPrice(cacheKey) if (cached) { try { reply.header('X-Cache', 'HIT') @@ -69,18 +74,22 @@ export async function registerRESTRoutes(app: FastifyInstance) { } catch { /* fall through */ } } + // NOTE: getAggregatedPrice reads price_points/price_aggregates, which + // have no network column yet — see getBestRoute's network param for + // the (currently SDEX-only) live per-network read. const agg = await getAggregatedPrice(pair.pairKey) - const route = await getBestRoute(pair.assetA, pair.assetB, pair.pairKey, 1000) + const route = await getBestRoute(pair.assetA, pair.assetB, pair.pairKey, 1000, network) const result = { assetA: pair.assetA.code, assetB: pair.assetB.code, pairKey: pair.pairKey, + network, ...agg, bestRoute: route.route, lastUpdated: new Date().toISOString(), } - await setCachedPrice(pair.pairKey, result, config.cache.priceTtl) + await setCachedPrice(cacheKey, result, config.cache.priceTtl) reply.header('X-Cache', 'MISS') return result } @@ -96,11 +105,12 @@ export async function registerRESTRoutes(app: FastifyInstance) { async (req, reply) => { const { assetA, assetB } = req.params const amount = parseFloat(req.query.amount ?? '1000') - const pair = findPair(assetA, assetB) - if (!pair) return reply.status(404).send({ error: `Pair ${assetA}/${assetB} not watched` }) + const network = req.network ?? activeNetwork + const pair = findPair(assetA, assetB, network) + if (!pair) return reply.status(404).send({ error: `Pair ${assetA}/${assetB} not watched on ${network}` }) if (isNaN(amount) || amount <= 0) return reply.status(400).send({ error: 'amount must be a positive number' }) - return getBestRoute(pair.assetA, pair.assetB, pair.pairKey, amount) + return getBestRoute(pair.assetA, pair.assetB, pair.pairKey, amount, network) } ) @@ -172,11 +182,13 @@ export async function registerRESTRoutes(app: FastifyInstance) { async (req, reply) => { const { assetA, assetB } = req.params const amount = parseFloat(req.query.amount ?? '1000') - const pair = findPair(assetA, assetB) - - if (!pair) return reply.status(404).send({ error: `Pair ${assetA}/${assetB} not watched` }) + const network = req.network ?? activeNetwork + const pair = findPair(assetA, assetB, network) + + if (!pair) return reply.status(404).send({ error: `Pair ${assetA}/${assetB} not watched on ${network}` }) if (isNaN(amount) || amount <= 0) return reply.status(400).send({ error: 'amount must be a positive number' }) + // NOTE: getDepth reads order-book data with no network column yet — see L048. const depthResult = await getDepth(pair.pairKey, amount) return { diff --git a/src/api/schemas.ts b/src/api/schemas.ts index 5784f06..4697d30 100644 --- a/src/api/schemas.ts +++ b/src/api/schemas.ts @@ -39,6 +39,7 @@ export const priceResponseSchema = { 'assetA', 'assetB', 'pairKey', + 'network', 'price', 'sdexPrice', 'ammPrice', @@ -62,6 +63,7 @@ export const priceResponseSchema = { assetA: { type: 'string' }, assetB: { type: 'string' }, pairKey: { type: 'string' }, + network: { type: 'string', enum: ['testnet', 'mainnet'] }, price: { type: 'number' }, sdexPrice: { type: 'number' }, ammPrice: { type: 'number' }, diff --git a/src/api/websocket.ts b/src/api/websocket.ts index 23ea79c..d570306 100644 --- a/src/api/websocket.ts +++ b/src/api/websocket.ts @@ -1,48 +1,56 @@ import type { FastifyInstance, FastifyRequest } from 'fastify' import websocket from '@fastify/websocket' import { priceEmitter, PRICE_UPDATE, PriceUpdateEvent } from '../events' -// @ts-ignore -import { x402ResourceServer, HTTPFacilitatorClient } from '@x402/core/server' -// @ts-ignore -import { ExactStellarScheme } from '@x402/stellar/exact/server' +import { activeNetwork, type NetworkName } from '../config' +import { X402_NETWORK_LABEL, paymentAddressFor, getX402ResourceServer } from '../x402/network' +import { resolveNetworkName } from '../middleware/network' import { fanOutManager } from '../ws/fanout' import { v4 as uuid } from 'uuid' -const PAYMENT_ADDRESS = process.env.ORACLE_PAYMENT_ADDRESS const FACILITATOR_URL = process.env.X402_FACILITATOR_URL ?? 'https://facilitator.stellar.org' -const NETWORK = (process.env.STELLAR_NETWORK === 'mainnet' ? 'stellar:pubnet' : 'stellar:testnet') as string export async function registerWebSocket(app: FastifyInstance) { await app.register(websocket) - let resourceServer: any = null - if (PAYMENT_ADDRESS) { - try { - const facilitatorClient = new HTTPFacilitatorClient({ url: FACILITATOR_URL }) - resourceServer = new x402ResourceServer(facilitatorClient) - .register(NETWORK as `${string}:${string}`, new ExactStellarScheme()) - await resourceServer.initialize() - } catch (err) { - app.log.warn(`[ws] x402 init failed, streaming without payment gating: ${(err as Error).message}`) - resourceServer = null - } - } - // @fastify/websocket v11: handler receives (socket, req) directly — no connection wrapper app.get('/ws', { websocket: true, config: { public: true } }, (socket: any, req: FastifyRequest) => { app.log.info('[ws] New connection attempt') + const rawNetwork = (req.query as any)?.network ?? (req.headers['x-network'] as string | undefined) + const resolved = resolveNetworkName(rawNetwork) + if (!resolved.ok) { + socket.send(JSON.stringify({ type: 'error', status: 400, message: resolved.error })) + socket.close() + return + } + const network: NetworkName = resolved.network + + // This process only ingests and streams live prices for `activeNetwork` + // (see src/config.ts) — the underlying price events carry no network tag + // yet (that's the deeper aggregation-layer work), so a request for any + // other network can't be honestly served here. + if (network !== activeNetwork) { + socket.send(JSON.stringify({ + type: 'error', + status: 400, + message: `This instance streams "${activeNetwork}" only; requested "${network}"`, + })) + socket.close() + return + } + + const paymentAddress = paymentAddressFor(network) const paymentHeader = (req.headers['x-payment'] as string) || (req.query as any).payment const requirements = { scheme: 'exact' as const, price: '$0.50', - network: NETWORK, - payTo: PAYMENT_ADDRESS!, + network: X402_NETWORK_LABEL[network], + payTo: paymentAddress!, } - if (!PAYMENT_ADDRESS || !resourceServer) { - app.log.warn('[ws] x402 disabled (PAYMENT_ADDRESS missing or x402 init failed)') + if (!paymentAddress) { + app.log.warn('[ws] x402 disabled (no payment address configured for this network)') } else if (!paymentHeader) { socket.send(JSON.stringify({ type: 'error', @@ -53,7 +61,8 @@ export async function registerWebSocket(app: FastifyInstance) { socket.close() return } else { - verifyPayment(paymentHeader, requirements, resourceServer) + getX402ResourceServer(network, FACILITATOR_URL) + .then(resourceServer => verifyPayment(paymentHeader, requirements, resourceServer)) .then(isValid => { if (!isValid) { socket.send(JSON.stringify({ type: 'error', message: 'Invalid payment' })) diff --git a/src/bazaar/catalog.ts b/src/bazaar/catalog.ts new file mode 100644 index 0000000..9e0a9a4 --- /dev/null +++ b/src/bazaar/catalog.ts @@ -0,0 +1,190 @@ +import { prisma } from '../db' +import type { NetworkName } from '../config' +import type { + BazaarResourceListing, + DiscoveryFilters, + DiscoveryResponse, + RegisterBazaarResourceInput, +} from './types' + +/** CAIP-2 network ids used by @x402/stellar and returned in `accepts[].network`. */ +const STELLAR_NETWORK_IDS: Record = { + mainnet: 'stellar:pubnet', + testnet: 'stellar:testnet', +} + +/** + * Registers (or updates) a resource in the Bazaar catalog. + * + * HTTP resources are identified by (network, url, httpMethod); MCP resources + * by (network, url, toolName) — the tuple the spec requires because multiple + * tools can multiplex over one MCP server endpoint. Re-registering the same + * identity updates the existing row instead of creating a duplicate, so a + * resource server can safely call this on every startup. + */ +export async function registerBazaarResource(input: RegisterBazaarResourceInput): Promise { + const payTo = input.accepts[0]?.payTo + if (!payTo) { + throw new Error('registerBazaarResource: accepts[] must contain at least one payment requirement with payTo') + } + + const base = { + type: input.type, + network: input.network, + url: input.resource.url, + description: input.resource.description ?? null, + mimeType: input.resource.mimeType ?? null, + serviceName: input.resource.serviceName ?? null, + tags: input.resource.tags ?? [], + iconUrl: input.resource.iconUrl ?? null, + mcpToolName: input.type === 'mcp' ? (input.bazaar.info.input as { toolName: string }).toolName : null, + httpMethod: input.type === 'http' ? (input.bazaar.info.input as { method: string }).method : null, + accepts: input.accepts as object, + payTo, + bazaarInfo: input.bazaar.info as object, + bazaarSchema: input.bazaar.schema as object, + routeTemplate: input.bazaar.routeTemplate ?? null, + extensionKeys: input.extensionKeys ?? ['bazaar'], + } + + if (input.type === 'mcp') { + const mcpToolName = base.mcpToolName as string + await prisma.bazaarResource.upsert({ + where: { + bazaarMcpIdentity: { network: input.network, url: input.resource.url, mcpToolName }, + }, + create: base, + update: base, + }) + } else { + const httpMethod = base.httpMethod as string + await prisma.bazaarResource.upsert({ + where: { + bazaarHttpIdentity: { network: input.network, url: input.resource.url, httpMethod }, + }, + create: base, + update: base, + }) + } +} + +export async function removeBazaarResource(network: NetworkName, url: string, key?: string): Promise { + await prisma.bazaarResource.deleteMany({ + where: { + network, + url, + OR: [{ httpMethod: key ?? undefined }, { mcpToolName: key ?? undefined }], + }, + }) +} + +/** + * Parses and clamps query-string filters for GET /discovery/resources. + * `limit` defaults to 50 and is clamped to [1, 200] to bound catalog scans; + * `offset` defaults to 0 and cannot be negative. + */ +export function parseDiscoveryFilters(query: Record): DiscoveryFilters { + const rawLimit = Number(query.limit) + const limit = Number.isFinite(rawLimit) && rawLimit > 0 ? Math.min(Math.floor(rawLimit), 200) : 50 + + const rawOffset = Number(query.offset) + const offset = Number.isFinite(rawOffset) && rawOffset > 0 ? Math.floor(rawOffset) : 0 + + const type = query.type === 'http' || query.type === 'mcp' ? query.type : undefined + const payTo = typeof query.payTo === 'string' && query.payTo.length > 0 ? query.payTo : undefined + const network = typeof query.network === 'string' && query.network.length > 0 ? query.network : undefined + const extensions = typeof query.extensions === 'string' && query.extensions.length > 0 ? query.extensions : undefined + + return { type, payTo, network, extensions, limit, offset } +} + +/** + * Maps the spec's `network` filter value (a CAIP-2 id, e.g. "stellar:pubnet" + * or "stellar:testnet") to our internal NetworkName column value. Falls back + * to matching the raw string directly so a facilitator that passes our + * NetworkName values (or any future non-Stellar CAIP-2 id we don't recognize + * yet) does not silently match nothing. + */ +function resolveNetworkFilter(network: string | undefined): string | undefined { + if (!network) return undefined + const entry = (Object.entries(STELLAR_NETWORK_IDS) as [NetworkName, string][]) + .find(([, caip2]) => caip2 === network) + return entry ? entry[0] : network +} + +function toListing(row: { + url: string + description: string | null + mimeType: string | null + serviceName: string | null + tags: string[] + iconUrl: string | null + accepts: unknown + bazaarInfo: unknown + bazaarSchema: unknown + routeTemplate: string | null + extensionKeys: string[] +}): BazaarResourceListing { + const extensions: BazaarResourceListing['extensions'] = { + bazaar: { + info: row.bazaarInfo as BazaarResourceListing['extensions']['bazaar']['info'], + schema: row.bazaarSchema as Record, + ...(row.routeTemplate ? { routeTemplate: row.routeTemplate } : {}), + }, + } + + return { + resource: { + url: row.url, + ...(row.description ? { description: row.description } : {}), + ...(row.mimeType ? { mimeType: row.mimeType } : {}), + ...(row.serviceName ? { serviceName: row.serviceName } : {}), + ...(row.tags.length > 0 ? { tags: row.tags } : {}), + ...(row.iconUrl ? { iconUrl: row.iconUrl } : {}), + }, + accepts: row.accepts as BazaarResourceListing['accepts'], + extensions, + } +} + +/** + * GET /discovery/resources — paginated catalog query implementing the six + * spec filters: + * - type: exact match on "http" | "mcp" + * - payTo: exact match against the resource's payment recipient + * - network: matches either our NetworkName ("mainnet"/"testnet") or the + * CAIP-2 id the spec's examples use ("stellar:pubnet"/"stellar:testnet") + * - extensions: matches resources that declare the given extension key + * (the spec's example is "bazaar", which every row declares by default) + * - limit / offset: standard offset pagination + * + * Ordering is (createdAt DESC, id DESC) — a stable tiebreaker on the primary + * key — so that concurrent inserts during a paginated walk never shift + * already-returned rows to a different page (the classic offset-pagination + * hazard when ordering by a non-unique column alone). + */ +export async function queryDiscoveryResources(filters: DiscoveryFilters): Promise { + const where = { + ...(filters.type ? { type: filters.type } : {}), + ...(filters.payTo ? { payTo: filters.payTo } : {}), + ...(filters.network ? { network: resolveNetworkFilter(filters.network) } : {}), + ...(filters.extensions ? { extensionKeys: { has: filters.extensions } } : {}), + } + + const [rows, total] = await Promise.all([ + prisma.bazaarResource.findMany({ + where, + orderBy: [{ createdAt: 'desc' }, { id: 'desc' }], + take: filters.limit, + skip: filters.offset, + }), + prisma.bazaarResource.count({ where }), + ]) + + return { + resources: rows.map(toListing), + limit: filters.limit, + offset: filters.offset, + total, + } +} diff --git a/src/bazaar/types.ts b/src/bazaar/types.ts new file mode 100644 index 0000000..9f30346 --- /dev/null +++ b/src/bazaar/types.ts @@ -0,0 +1,87 @@ +import type { NetworkName } from '../config' + +/** + * Shapes for the x402 `bazaar` discovery extension + * (specs/extensions/bazaar.md in x402-foundation/x402). + * + * These mirror the spec's `resource` + `accepts` + `extensions.bazaar` shape + * used in a 402 PaymentRequired response — a discovery listing is that same + * shape, catalogued. + */ + +export interface BazaarResourceInfo { + url: string + description?: string + mimeType?: string + serviceName?: string + tags?: string[] + iconUrl?: string +} + +export interface BazaarPaymentRequirement { + scheme: string + network: string + amount: string + asset: string + payTo: string + maxTimeoutSeconds: number + extra?: Record +} + +export type BazaarHttpInput = + | { type: 'http'; method: 'GET' | 'HEAD' | 'DELETE'; queryParams?: Record; pathParams?: Record } + | { type: 'http'; method: 'POST' | 'PUT' | 'PATCH'; bodyType: string; body?: Record; pathParams?: Record } + +export interface BazaarMcpInput { + type: 'mcp' + toolName: string + inputSchema: Record + transport?: 'streamable-http' | 'sse' +} + +export interface BazaarInfo { + input: BazaarHttpInput | BazaarMcpInput + output?: Record +} + +export interface BazaarExtensionDeclaration { + info: BazaarInfo + schema: Record + routeTemplate?: string +} + +/** A single item in the GET /discovery/resources response. */ +export interface BazaarResourceListing { + resource: BazaarResourceInfo + accepts: BazaarPaymentRequirement[] + extensions: { + bazaar: BazaarExtensionDeclaration + [key: string]: unknown + } +} + +/** Input for registering a new listing in the catalog. */ +export interface RegisterBazaarResourceInput { + type: 'http' | 'mcp' + network: NetworkName + resource: BazaarResourceInfo + accepts: BazaarPaymentRequirement[] + bazaar: BazaarExtensionDeclaration + extensionKeys?: string[] +} + +export interface DiscoveryFilters { + type?: 'http' | 'mcp' + payTo?: string + network?: string + extensions?: string + limit: number + offset: number +} + +export interface DiscoveryResponse { + resources: BazaarResourceListing[] + limit: number + offset: number + total: number +} diff --git a/src/config.ts b/src/config.ts index aea5277..8d8ffd2 100644 --- a/src/config.ts +++ b/src/config.ts @@ -18,12 +18,24 @@ export interface NetworkConfig { passphrase: string } soroswap: { + /** Whether Soroswap has a usable deployment on this network. */ + enabled: boolean /** Soroswap factory contract address for this network. */ factoryAddress: string + /** Soroswap token-list URL for this network. */ + tokenListUrl: string /** How often to poll Soroswap pool reserves in ms. */ pollIntervalMs: number } + aquarius: { + /** Whether Aquarius has a usable deployment on this network. */ + enabled: boolean + /** Aquarius AMM pools API base URL for this network. */ + apiUrl: string + } oracle: { + /** Whether the Reflector oracle is deployed on this network. */ + enabled: boolean /** Reflector oracle contract ID for this network. */ reflectorContractId: string } @@ -106,6 +118,16 @@ function buildNetworkConfig(network: NetworkName): NetworkConfig { ? 'CA4HEQTL2WPEUYKYKCDOHCDNIV4QHNJ7EL4J4NQ6VADP7SYHVRYZ7AW2' : 'CDKP5WSEZMDL53VZFPBGCL47WBPKFCN5OPYQVXB3CJWUXHPZRPHSSZ3') + // Soroswap token-list is a single canonical list covering both networks by + // default, but can be overridden per network (e.g. a testnet-specific list). + const soroswapTokenListUrl = + process.env[`SOROSWAP_TOKEN_LIST_URL_${suffix}`] || + process.env.SOROSWAP_TOKEN_LIST_URL || + 'https://raw.githubusercontent.com/soroswap/token-list/main/tokenList.json' + + const soroswapEnabled = + (process.env[`SOROSWAP_ENABLED_${suffix}`] ?? 'true').toLowerCase() !== 'false' + const soroswapPollMs = parseInt( process.env[`SOROSWAP_POLL_INTERVAL_MS_${suffix}`] || process.env.SOROSWAP_POLL_INTERVAL_MS || @@ -113,6 +135,17 @@ function buildNetworkConfig(network: NetworkName): NetworkConfig { 10 ) + // ── Aquarius ────────────────────────────────────────────────────────────── + // Aquarius only runs on Stellar classic mainnet today — there is no public + // testnet deployment, so it is disabled there by default. + const aquariusApiUrl = + process.env[`AQUARIUS_API_URL_${suffix}`] || + process.env.AQUARIUS_API_URL || + 'https://amm.aquarius.network/api/v1/pools/' + + const aquariusEnabled = + (process.env[`AQUARIUS_ENABLED_${suffix}`] ?? (network === 'mainnet' ? 'true' : 'false')).toLowerCase() !== 'false' + // ── Reflector oracle ────────────────────────────────────────────────────── const reflectorContractId = process.env[`REFLECTOR_CONTRACT_ID_${suffix}`] || @@ -121,6 +154,10 @@ function buildNetworkConfig(network: NetworkName): NetworkConfig { ? 'CCYXZMNHFXHKF3YEX4VJJ5TH3YHCVZIBPNBGM7C4PJIMCIMNNWDOQYA' : '') + const oracleEnabled = + (process.env[`REFLECTOR_ENABLED_${suffix}`] ?? 'true').toLowerCase() !== 'false' && + reflectorContractId !== '' + // ── Watched pairs ───────────────────────────────────────────────────────── const rawPairs = process.env[`WATCHED_PAIRS_${suffix}`] || @@ -132,10 +169,16 @@ function buildNetworkConfig(network: NetworkName): NetworkConfig { rpc: { url: rpcUrl }, network: { passphrase }, soroswap: { + enabled: soroswapEnabled, factoryAddress: soroswapFactory, + tokenListUrl: soroswapTokenListUrl, pollIntervalMs: soroswapPollMs, }, - oracle: { reflectorContractId }, + aquarius: { + enabled: aquariusEnabled, + apiUrl: aquariusApiUrl, + }, + oracle: { enabled: oracleEnabled, reflectorContractId }, pairs: parseWatchedPairs(rawPairs), } } @@ -229,6 +272,7 @@ export const config = { get rpc() { return resolveNetwork(activeNetwork).rpc }, get network() { return resolveNetwork(activeNetwork).network }, get soroswap() { return resolveNetwork(activeNetwork).soroswap }, + get aquarius() { return resolveNetwork(activeNetwork).aquarius }, get oracle() { return resolveNetwork(activeNetwork).oracle }, get pairs() { return resolveNetwork(activeNetwork).pairs }, diff --git a/src/db.ts b/src/db.ts index 3485c49..508dd2f 100644 --- a/src/db.ts +++ b/src/db.ts @@ -1,7 +1,7 @@ import { PrismaClient } from '@prisma/client' import { Pool } from 'pg' import { db_query_duration_seconds } from './metrics' -import { config } from './config' +import { config, activeNetwork } from './config' // Prisma for schema management + simple queries const globalForPrisma = globalThis as unknown as { prisma?: PrismaClient } @@ -37,6 +37,7 @@ export async function upsertPricePoints(points: { if (points.length === 0) return 0 const result = await prisma.pricePoint.createMany({ data: points.map(p => ({ + network: activeNetwork, assetA: p.assetA, assetB: p.assetB, pairKey: p.pairKey, @@ -55,14 +56,14 @@ export async function upsertPricePoints(points: { } export async function getIndexerCursor(id: string): Promise { - const state = await prisma.indexerState.findUnique({ where: { id } }) + const state = await prisma.indexerState.findUnique({ where: { network_id: { network: activeNetwork, id } } }) return state?.lastCursor ?? null } export async function setIndexerCursor(id: string, cursor: string, ledger?: number): Promise { await prisma.indexerState.upsert({ - where: { id }, - create: { id, lastCursor: cursor, lastLedger: ledger, lastProcessedAt: new Date() }, + where: { network_id: { network: activeNetwork, id } }, + create: { id, network: activeNetwork, lastCursor: cursor, lastLedger: ledger, lastProcessedAt: new Date() }, update: { lastCursor: cursor, lastLedger: ledger, lastProcessedAt: new Date() }, }) } diff --git a/src/index.ts b/src/index.ts index 1d12cb8..538dbab 100644 --- a/src/index.ts +++ b/src/index.ts @@ -19,6 +19,7 @@ import { registerPairsRoutes } from './routes/pairs' import { registerScreenerRoutes } from './routes/screener' import { registerHistoryRoutes } from './api/history' import { registerX402 } from './middleware/x402' +import { registerNetworkSelector } from './middleware/network' import { registerWebSocket } from './api/websocket' import { registerApiKeyAuth } from './api/auth' import { registerAdminRoutes } from './api/admin' @@ -28,6 +29,7 @@ import { registerVolumeRoutes } from './routes/volumes' import { registerBenchmarkRoutes } from './routes/benchmark' import { registerOracleRoutes } from './routes/oracle' import { registerBasketRoutes } from './routes/basket' +import { registerDiscoveryRoutes } from './routes/discovery' import { fanOutManager } from './ws/fanout' import { startSDEXIngester } from './ingesters/sdex' @@ -61,6 +63,12 @@ async function main() { await app.register(cors, { origin: true }) await app.register(compress) + // Resolves the per-request Stellar network (?network= query param / x-network + // header) onto req.network, validating it (400 on an unrecognised value). + // Runs in onRequest, ahead of API-key auth/rate-limiting/x402 and every route + // handler, so all of them can read req.network. + await app.register(registerNetworkSelector) + // API-key authentication — validates Authorization: Bearer and attaches // per-key quota metadata to req.apiKey. Registered BEFORE the rate limiter so // that req.apiKey is populated when the limiter evaluates its per-key quota @@ -121,6 +129,7 @@ async function main() { await registerBenchmarkRoutes(app) await registerOracleRoutes(app) await registerBasketRoutes(app) + await registerDiscoveryRoutes(app) await registerGraphQL(app) await registerWebSocket(app) diff --git a/src/ingest/oracles/reflector.ts b/src/ingest/oracles/reflector.ts index 6bcee44..1323a9a 100644 --- a/src/ingest/oracles/reflector.ts +++ b/src/ingest/oracles/reflector.ts @@ -17,22 +17,12 @@ import { nativeToScVal, Account, } from '@stellar/stellar-sdk' -import { config } from '../../config' - -// Reflector oracle contract address — resolved per active network via config. -// Set REFLECTOR_CONTRACT_ID_TESTNET / REFLECTOR_CONTRACT_ID_MAINNET (or the -// legacy REFLECTOR_CONTRACT_ID fallback) in your environment. -const REFLECTOR_CONTRACT_ID = config.oracle.reflectorContractId +import { activeNetwork, getNetworkConfig, type NetworkName } from '../../config' +import { getRpcServer } from '../../network/clients' // Ephemeral fee payer — simulation only, no real funds needed const FEE_PAYER = Keypair.random() -let _rpc: SorobanRpc.Server | null = null -function getRpc(): SorobanRpc.Server { - _rpc ??= new SorobanRpc.Server(config.rpc.url, { allowHttp: true }) - return _rpc -} - export interface ReflectorPrice { asset: string price: number @@ -40,18 +30,27 @@ export interface ReflectorPrice { } /** - * Fetch the latest price for a given asset code from the Reflector oracle. - * Returns null when the contract is unreachable or the asset is unknown. + * Fetch the latest price for a given asset code from the Reflector oracle on + * the given network. Returns null when the contract is unreachable, the + * network has no Reflector deployment configured, or the asset is unknown. */ -export async function fetchReflectorPrice(assetCode: string): Promise { +export async function fetchReflectorPrice( + assetCode: string, + network: NetworkName = activeNetwork +): Promise { + const netConfig = getNetworkConfig(network) + // `oracle.enabled` already implies a non-empty reflectorContractId (config.ts), + // so this single check subsumes the contract-id guard. + if (!netConfig.oracle.enabled) return null + try { - const rpc = getRpc() - const contract = new Contract(REFLECTOR_CONTRACT_ID) + const rpc = getRpcServer(network) + const contract = new Contract(netConfig.oracle.reflectorContractId) const account = new Account(FEE_PAYER.publicKey(), '0') const tx = new TransactionBuilder(account, { fee: BASE_FEE, - networkPassphrase: config.network.passphrase, + networkPassphrase: netConfig.network.passphrase, }) .addOperation( contract.call('lastprice', nativeToScVal(assetCode, { type: 'symbol' })) @@ -86,13 +85,16 @@ export async function fetchReflectorPrice(assetCode: string): Promise() const CACHE_TTL_MS = 60_000 -export async function getCachedReflectorPrice(asset: string): Promise { - const key = asset.toUpperCase() +export async function getCachedReflectorPrice( + asset: string, + network: NetworkName = activeNetwork +): Promise { + const key = `${network}:${asset.toUpperCase()}` const entry = _cache.get(key) if (entry && Date.now() - entry.fetchedAt < CACHE_TTL_MS) { return entry.price } - const fresh = await fetchReflectorPrice(key) + const fresh = await fetchReflectorPrice(asset, network) if (fresh) { _cache.set(key, { price: fresh.price, fetchedAt: Date.now() }) return fresh.price diff --git a/src/ingest/venues/aquarius.ts b/src/ingest/venues/aquarius.ts index a609d72..5e0fab3 100644 --- a/src/ingest/venues/aquarius.ts +++ b/src/ingest/venues/aquarius.ts @@ -14,8 +14,6 @@ import { upsertPricePoints } from '../../db' import { dispatchPriceUpdate } from '../../webhookDispatcher' import type { WatchedPair } from '../../types' -const AQUARIUS_AMM_API = 'https://amm.aquarius.network/api/v1/pools/' - const lastPrice = new Map() interface AquariusPool { @@ -28,7 +26,10 @@ interface AquariusListResponse { results?: AquariusPool[] } -export async function fetchAquariusPools(pair: WatchedPair): Promise { +export async function fetchAquariusPools( + pair: WatchedPair, + apiUrl: string = config.aquarius.apiUrl +): Promise { try { const assetAStr = pair.assetA.issuer ? `${pair.assetA.code}:${pair.assetA.issuer}` @@ -41,7 +42,7 @@ export async function fetchAquariusPools(pair: WatchedPair): Promise { + if (!config.aquarius.enabled) { + console.log('[aquarius] Aquarius is disabled on this network — ingester not started') + return + } + console.log(`[aquarius] Starting Aquarius AMM ingester for ${getActivePairs().length} pairs`) while (true) { for (const pair of getActivePairs()) { diff --git a/src/ingesters/amm.ts b/src/ingesters/amm.ts index f2aae24..a91841e 100644 --- a/src/ingesters/amm.ts +++ b/src/ingesters/amm.ts @@ -1,16 +1,13 @@ -import { Horizon } from '@stellar/stellar-sdk' import { amm_snapshots_total, trades_ingested_total, last_trade_timestamp } from '../metrics' -import { config } from '../config' +import { config, activeNetwork, getNetworkConfig, type NetworkName } from '../config' import { getActivePairs } from '../pairsRegistry' import { upsertPricePoints, getIndexerCursor, setIndexerCursor, prisma } from '../db' import { dispatchPriceUpdate } from '../webhookDispatcher' import type { WatchedPair } from '../types' -const horizonServer = new Horizon.Server(config.horizon.url) - const lastPrice = new Map() -export async function fetchPools(pair: WatchedPair): Promise { +export async function fetchPools(pair: WatchedPair, network: NetworkName = activeNetwork): Promise { try { // Use Horizon's reserves filter to find pools for this specific pair const assetAStr = pair.assetA.issuer @@ -26,7 +23,7 @@ export async function fetchPools(pair: WatchedPair): Promise { params.set('limit', '10') const response = await fetch( - `${config.horizon.url}/liquidity_pools?${params.toString()}` + `${getNetworkConfig(network).horizon.url}/liquidity_pools?${params.toString()}` ) const data = await response.json() as any if (!data._embedded?.records) return [] @@ -52,6 +49,7 @@ export async function snapshotPool(pool: any, pair: WatchedPair): Promise await prisma.poolSnapshot.create({ data: { + network: activeNetwork, poolId: pool.id, assetA: pair.assetA.code, assetB: pair.assetB.code, @@ -97,13 +95,17 @@ export async function snapshotPool(pool: any, pair: WatchedPair): Promise } } -export async function ingestPoolTrades(pool: any, pair: WatchedPair): Promise { - const stateId = `amm:${pool.id}` +export async function ingestPoolTrades( + pool: any, + pair: WatchedPair, + network: NetworkName = activeNetwork +): Promise { + const stateId = `amm:${network}:${pool.id}` const cursor = await getIndexerCursor(stateId) ?? '0' try { const response = await fetch( - `${config.horizon.url}/liquidity_pools/${pool.id}/trades?cursor=${cursor}&limit=${config.indexer.ammPageSize}&order=asc` + `${getNetworkConfig(network).horizon.url}/liquidity_pools/${pool.id}/trades?cursor=${cursor}&limit=${config.indexer.ammPageSize}&order=asc` ) const data = await response.json() as any const records = data._embedded?.records ?? [] @@ -161,17 +163,17 @@ async function sleep(ms: number) { return new Promise(r => setTimeout(r, ms)) } -export async function startAMMIngester(): Promise { - console.log(`[amm] Starting AMM ingester for ${getActivePairs().length} pairs`) +export async function startAMMIngester(network: NetworkName = activeNetwork): Promise { + console.log(`[amm] Starting AMM ingester for ${getActivePairs().length} pairs on ${network}`) while (true) { for (const pair of getActivePairs()) { - const pools = await fetchPools(pair) + const pools = await fetchPools(pair, network) console.log(`[amm] ${pair.pairKey}: found ${pools.length} AMM pools`) await Promise.all(pools.map(async pool => { await snapshotPool(pool, pair) - await ingestPoolTrades(pool, pair) + await ingestPoolTrades(pool, pair, network) })) } await sleep(config.indexer.pollIntervalMs) diff --git a/src/ingesters/sdex.ts b/src/ingesters/sdex.ts index 0218572..e5022cc 100644 --- a/src/ingesters/sdex.ts +++ b/src/ingesters/sdex.ts @@ -1,13 +1,12 @@ -import { Horizon, Asset } from '@stellar/stellar-sdk' +import { Asset } from '@stellar/stellar-sdk' import { trades_ingested_total, last_trade_timestamp } from '../metrics' -import { config } from '../config' +import { config, activeNetwork, type NetworkName } from '../config' +import { getHorizonServer } from '../network/clients' import { getActivePairs } from '../pairsRegistry' import { upsertPricePoints, getIndexerCursor, setIndexerCursor } from '../db' import { dispatchPriceUpdate } from '../webhookDispatcher' import type { WatchedPair } from '../types' -const horizonServer = new Horizon.Server(config.horizon.url) - // Last seen price per pairKey — used for threshold crossing detection const lastPrice = new Map() @@ -16,15 +15,15 @@ function toAsset(asset: { code: string; issuer: string | null }): Asset { return new Asset(asset.code, asset.issuer) } -export async function ingestPair(pair: WatchedPair): Promise { - const stateId = `sdex:${pair.pairKey}` +export async function ingestPair(pair: WatchedPair, network: NetworkName = activeNetwork): Promise { + const stateId = `sdex:${network}:${pair.pairKey}` const cursor = await getIndexerCursor(stateId) ?? '0' try { const assetA = toAsset(pair.assetA) const assetB = toAsset(pair.assetB) - const trades = await horizonServer + const trades = await getHorizonServer(network) .trades() .forAssetPair(assetA, assetB) .cursor(cursor) @@ -87,11 +86,11 @@ async function sleep(ms: number) { return new Promise(r => setTimeout(r, ms)) } -export async function startSDEXIngester(): Promise { - console.log(`[sdex] Starting SDEX ingester for ${getActivePairs().length} pairs`) +export async function startSDEXIngester(network: NetworkName = activeNetwork): Promise { + console.log(`[sdex] Starting SDEX ingester for ${getActivePairs().length} pairs on ${network}`) while (true) { - await Promise.all(getActivePairs().map(pair => ingestPair(pair))) + await Promise.all(getActivePairs().map(pair => ingestPair(pair, network))) await sleep(config.indexer.pollIntervalMs) } } diff --git a/src/ingesters/snapshot.ts b/src/ingesters/snapshot.ts index 1191c61..bc40758 100644 --- a/src/ingesters/snapshot.ts +++ b/src/ingesters/snapshot.ts @@ -1,4 +1,5 @@ import { pgPool } from '../db' +import { activeNetwork } from '../config' import { getActivePairs } from '../pairsRegistry' import { price_snapshots_total } from '../metrics' @@ -36,23 +37,23 @@ export async function appendSnapshots(now: Date = new Date()): Promise { `WITH latest AS ( SELECT DISTINCT ON (pair_key) pair_key, price::numeric AS price FROM price_points - WHERE pair_key = ANY($1) + WHERE pair_key = ANY($1) AND network = $3 ORDER BY pair_key, timestamp DESC ), vol AS ( SELECT pair_key, SUM(base_volume::numeric) AS volume FROM price_points - WHERE pair_key = ANY($1) + WHERE pair_key = ANY($1) AND network = $3 AND timestamp >= $2 AND timestamp < $2 + INTERVAL '1 minute' GROUP BY pair_key ) - INSERT INTO price_snapshots (pair, ts, price, volume) - SELECT l.pair_key, $2, l.price, COALESCE(v.volume, 0) + INSERT INTO price_snapshots (network, pair, ts, price, volume) + SELECT $3, l.pair_key, $2, l.price, COALESCE(v.volume, 0) FROM latest l LEFT JOIN vol v ON v.pair_key = l.pair_key - ON CONFLICT (pair, ts) DO NOTHING`, - [pairKeys, ts] + ON CONFLICT (network, pair, ts) DO NOTHING`, + [pairKeys, ts, activeNetwork] ) const inserted = result.rowCount ?? 0 diff --git a/src/ingesters/soroswap.ts b/src/ingesters/soroswap.ts index b1accc2..183e69f 100644 --- a/src/ingesters/soroswap.ts +++ b/src/ingesters/soroswap.ts @@ -19,7 +19,8 @@ import { scValToNative, Account, } from '@stellar/stellar-sdk' -import { config } from '../config' +import { activeNetwork, getNetworkConfig, type NetworkName } from '../config' +import { getRpcServer } from '../network/clients' import { getActivePairs } from '../pairsRegistry' import { upsertPricePoints } from '../db' import { dispatchPriceUpdate } from '../webhookDispatcher' @@ -27,9 +28,6 @@ import type { WatchedPair } from '../types' // ── Constants ───────────────────────────────────────────────────────────────── -const SOROSWAP_TOKEN_LIST_URL = - 'https://raw.githubusercontent.com/soroswap/token-list/main/tokenList.json' - // Ephemeral fee-payer account (no real funds needed for simulation) const FEE_PAYER_KEYPAIR = Keypair.random() @@ -55,25 +53,17 @@ export interface PoolEntry { tokenB: SoroswapToken } -// ── RPC client (lazy-initialised so tests can skip it) ──────────────────────── - -let _rpc: SorobanRpc.Server | null = null -function getRpc(): SorobanRpc.Server { - if (!_rpc) { - _rpc = new SorobanRpc.Server(config.rpc.url, { allowHttp: true }) - } - return _rpc -} - // ── Token-list helpers ──────────────────────────────────────────────────────── /** * Fetch Soroswap token list. Returns an empty array on failure so the ingester * degrades gracefully without affecting other ingesters. */ -export async function fetchSoroswapTokenList(): Promise { +export async function fetchSoroswapTokenList( + tokenListUrl: string = getNetworkConfig(activeNetwork).soroswap.tokenListUrl +): Promise { try { - const res = await fetch(SOROSWAP_TOKEN_LIST_URL) + const res = await fetch(tokenListUrl) if (!res.ok) throw new Error(`HTTP ${res.status}`) const data = (await res.json()) as SoroswapTokenList return Array.isArray(data.tokens) ? data.tokens : [] @@ -92,14 +82,15 @@ export async function fetchSoroswapTokenList(): Promise { export async function fetchPoolsFromFactory( factoryAddress: string, tokenA: string, - tokenB: string + tokenB: string, + network: NetworkName = activeNetwork ): Promise { try { - const rpc = getRpc() + const rpc = getRpcServer(network) const factory = new Contract(factoryAddress) const account = new Account(FEE_PAYER_KEYPAIR.publicKey(), '0') const networkPassphrase = - config.network.passphrase ?? Networks.PUBLIC + getNetworkConfig(network).network.passphrase ?? Networks.PUBLIC const tx = new TransactionBuilder(account, { fee: BASE_FEE, @@ -146,14 +137,15 @@ export async function fetchPoolsFromFactory( * Returns null on any RPC error. */ export async function fetchPoolReserves( - poolAddress: string + poolAddress: string, + network: NetworkName = activeNetwork ): Promise<[bigint, bigint] | null> { try { - const rpc = getRpc() + const rpc = getRpcServer(network) const pool = new Contract(poolAddress) const account = new Account(FEE_PAYER_KEYPAIR.publicKey(), '0') const networkPassphrase = - config.network.passphrase ?? Networks.PUBLIC + getNetworkConfig(network).network.passphrase ?? Networks.PUBLIC const tx = new TransactionBuilder(account, { fee: BASE_FEE, @@ -212,10 +204,11 @@ export function calcSpotPrice(reserveA: bigint, reserveB: bigint): number { export async function ingestPool( poolEntry: PoolEntry, pair: WatchedPair, - _fetchReserves = fetchPoolReserves + _fetchReserves: (poolAddress: string, network?: NetworkName) => Promise<[bigint, bigint] | null> = fetchPoolReserves, + network: NetworkName = activeNetwork ): Promise { try { - const reserves = await _fetchReserves(poolEntry.poolAddress) + const reserves = await _fetchReserves(poolEntry.poolAddress, network) if (!reserves) return const [reserveA, reserveB] = reserves @@ -277,7 +270,13 @@ export async function ingestPair( pair: WatchedPair, tokens: SoroswapToken[], factoryAddress: string, - _fetchPools = fetchPoolsFromFactory + _fetchPools: ( + factoryAddress: string, + tokenA: string, + tokenB: string, + network?: NetworkName + ) => Promise = fetchPoolsFromFactory, + network: NetworkName = activeNetwork ): Promise { const tokenA = tokens.find( (t) => t.symbol.toUpperCase() === pair.assetA.code.toUpperCase() @@ -296,7 +295,8 @@ export async function ingestPair( const poolAddresses = await _fetchPools( factoryAddress, tokenA.address, - tokenB.address + tokenB.address, + network ) if (!poolAddresses.length) { @@ -310,7 +310,9 @@ export async function ingestPair( poolAddresses.map((addr) => ingestPool( { poolAddress: addr, tokenA, tokenB }, - pair + pair, + fetchPoolReserves, + network ) ) ) @@ -326,23 +328,31 @@ async function sleep(ms: number): Promise { * Start the Soroswap AMM ingester. Runs as an infinite polling loop. * Fault-isolated: a crash is caught by the caller (restartIngester in index.ts). */ -export async function startSoroswapIngester(): Promise { - const factoryAddress = config.soroswap.factoryAddress - const pollInterval = config.soroswap.pollIntervalMs +export async function startSoroswapIngester(network: NetworkName = activeNetwork): Promise { + const netConfig = getNetworkConfig(network) + + if (!netConfig.soroswap.enabled) { + console.log(`[soroswap] Soroswap is disabled on ${network} — ingester not started`) + return + } + + const factoryAddress = netConfig.soroswap.factoryAddress + const tokenListUrl = netConfig.soroswap.tokenListUrl + const pollInterval = netConfig.soroswap.pollIntervalMs console.log( - `[soroswap] Starting Soroswap ingester | factory=${factoryAddress} | interval=${pollInterval}ms` + `[soroswap] Starting Soroswap ingester on ${network} | factory=${factoryAddress} | interval=${pollInterval}ms` ) while (true) { const pairs = getActivePairs() - const tokens = await fetchSoroswapTokenList() + const tokens = await fetchSoroswapTokenList(tokenListUrl) if (tokens.length === 0) { console.warn('[soroswap] Token list empty — skipping poll cycle') } else { await Promise.all( - pairs.map((pair) => ingestPair(pair, tokens, factoryAddress)) + pairs.map((pair) => ingestPair(pair, tokens, factoryAddress, fetchPoolsFromFactory, network)) ) } diff --git a/src/jobs/aggregateRefresh.ts b/src/jobs/aggregateRefresh.ts index d5df530..fcd77ca 100644 --- a/src/jobs/aggregateRefresh.ts +++ b/src/jobs/aggregateRefresh.ts @@ -1,11 +1,11 @@ import { Queue, Worker } from 'bullmq' -import { config } from '../config' +import { config, activeNetwork } from '../config' import { pgPool, prisma } from '../db' import { setCachedPrice } from '../redis' import { calculateVWAP, calculateOHLCV, getAggregatedPrice } from '../aggregator/vwap' import { getBestRoute } from '../aggregator/bestRoute' -const QUEUE_NAME = 'aggregate-refresh' +const QUEUE_NAME = `${activeNetwork}:aggregate-refresh` function redisConnection() { const url = process.env.REDIS_URL @@ -60,9 +60,9 @@ export function startAggregateWorker() { if (vwap === 0) continue await prisma.priceAggregate.upsert({ - where: { pairKey_window_bucket: { pairKey, window: w.key, bucket } }, + where: { network_pairKey_window_bucket: { network: activeNetwork, pairKey, window: w.key, bucket } }, create: { - pairKey, window: w.key, bucket, + network: activeNetwork, pairKey, window: w.key, bucket, vwap, sdexVwap: sdexVwap || null, ammVwap: ammVwap || null, volume: ohlcv.volume, tradeCount: ohlcv.tradeCount, openPrice: ohlcv.open || null, closePrice: ohlcv.close || null, diff --git a/src/jobs/snapshotRetention.ts b/src/jobs/snapshotRetention.ts index 5f266ab..fca11f2 100644 --- a/src/jobs/snapshotRetention.ts +++ b/src/jobs/snapshotRetention.ts @@ -1,7 +1,8 @@ import { Queue, Worker } from 'bullmq' import { pgPool } from '../db' +import { activeNetwork } from '../config' -const QUEUE_NAME = 'snapshot-retention' +const QUEUE_NAME = `${activeNetwork}:snapshot-retention` /** Snapshots older than this many days are pruned by the retention job. */ export const SNAPSHOT_RETENTION_DAYS = 30 @@ -24,8 +25,8 @@ export function createSnapshotRetentionQueue() { export async function pruneOldSnapshots(retentionDays: number = SNAPSHOT_RETENTION_DAYS): Promise { const result = await pgPool.query( `DELETE FROM price_snapshots - WHERE ts < NOW() - ($1 || ' days')::interval`, - [retentionDays] + WHERE network = $1 AND ts < NOW() - ($2 || ' days')::interval`, + [activeNetwork, retentionDays] ) return result.rowCount ?? 0 } diff --git a/src/middleware/network.ts b/src/middleware/network.ts new file mode 100644 index 0000000..8519559 --- /dev/null +++ b/src/middleware/network.ts @@ -0,0 +1,70 @@ +import type { FastifyInstance, FastifyRequest, FastifyReply } from 'fastify' +import fp from 'fastify-plugin' +import { activeNetwork, type NetworkName } from '../config' + +const VALID_NETWORKS: readonly NetworkName[] = ['testnet', 'mainnet'] + +declare module 'fastify' { + interface FastifyRequest { + /** + * The Stellar network this request targets, resolved from the `network` + * query param / `x-network` header (see {@link resolveNetworkName}). + * Defaults to `activeNetwork` when the request specifies nothing. + */ + network: NetworkName + } +} + +/** + * Resolves a raw `network` value (query param or header) into a validated + * {@link NetworkName}. An absent/empty value resolves to `activeNetwork` + * (this deployment's configured default) rather than being an error — only + * an explicit, unrecognised value is rejected. + */ +export function resolveNetworkName( + raw: string | undefined | null +): { ok: true; network: NetworkName } | { ok: false; error: string } { + if (raw == null || raw === '') return { ok: true, network: activeNetwork } + const lower = raw.trim().toLowerCase() + if ((VALID_NETWORKS as string[]).includes(lower)) { + return { ok: true, network: lower as NetworkName } + } + return { + ok: false, + error: `Invalid network "${raw}" — expected one of: ${VALID_NETWORKS.join(', ')}`, + } +} + +function rawNetworkFromRequest(req: FastifyRequest): string | undefined { + const fromQuery = (req.query as Record | undefined)?.network + if (typeof fromQuery === 'string') return fromQuery + + const fromHeader = req.headers['x-network'] + if (typeof fromHeader === 'string') return fromHeader + + return undefined +} + +/** + * Fastify plugin that resolves the per-request Stellar network from a + * `?network=` query param or `x-network` header, validates it, and attaches + * it to `req.network`. An unrecognised value gets a 400 before any route + * handler or downstream middleware (x402, WebSocket auth) runs. + * + * Must be registered early (`onRequest`) so `req.network` is populated + * before `middleware/x402.ts`'s `preHandler` hook and any route handler. + */ +async function networkSelectorPlugin(app: FastifyInstance) { + app.decorateRequest('network', activeNetwork) + + app.addHook('onRequest', async (req: FastifyRequest, reply: FastifyReply) => { + const resolved = resolveNetworkName(rawNetworkFromRequest(req)) + if (!resolved.ok) { + reply.status(400).send({ error: resolved.error }) + return + } + req.network = resolved.network + }) +} + +export const registerNetworkSelector = fp(networkSelectorPlugin, { name: 'network-selector' }) diff --git a/src/middleware/x402.ts b/src/middleware/x402.ts index e16ce89..bc1da10 100644 --- a/src/middleware/x402.ts +++ b/src/middleware/x402.ts @@ -1,11 +1,9 @@ import type { FastifyInstance, FastifyRequest, FastifyReply } from 'fastify' import { x402_payments_received_total } from '../metrics' import { checkQuota, recordUsage, parseCents, getQuotaConfig } from '../x402/metering' +import { X402_NETWORK_LABEL, paymentAddressFor, isX402Configured, getX402ResourceServer } from '../x402/network' import fp from 'fastify-plugin' -// @ts-ignore — @x402 packages ship ESM-only types incompatible with commonjs moduleResolution -import { x402ResourceServer, HTTPFacilitatorClient } from '@x402/core/server' -// @ts-ignore -import { ExactStellarScheme } from '@x402/stellar/exact/server' +import './network' // declares req.network on the FastifyRequest type // Routes gated by x402 and their prices const GATED_ROUTES: Record = { @@ -21,20 +19,13 @@ const GATED_ROUTES: Record = { */ async function x402Plugin(app: FastifyInstance) { // Read at plugin init time (not module load) so tests can inject env vars before app.register() - const PAYMENT_ADDRESS = process.env.ORACLE_PAYMENT_ADDRESS const FACILITATOR_URL = process.env.X402_FACILITATOR_URL ?? 'https://facilitator.stellar.org' - const NETWORK = (process.env.STELLAR_NETWORK === 'mainnet' ? 'stellar:pubnet' : 'stellar:testnet') as string - if (!PAYMENT_ADDRESS) { + if (!isX402Configured()) { app.log.warn('[oracle] ORACLE_PAYMENT_ADDRESS not set — x402 gating disabled') return } - const facilitatorClient = new HTTPFacilitatorClient({ url: FACILITATOR_URL }) - const resourceServer: any = new x402ResourceServer(facilitatorClient) - .register(NETWORK as `${string}:${string}`, new ExactStellarScheme()) - - await resourceServer.initialize() app.log.info('[oracle] x402 payment gating enabled') app.addHook('preHandler', async (req: FastifyRequest, reply: FastifyReply) => { @@ -46,14 +37,23 @@ async function x402Plugin(app: FastifyInstance) { }) if (!matchedRoute) return + // Falls back to testnet when the network selector plugin isn't + // registered (e.g. isolated unit tests that build the app directly). + const network = req.network ?? 'testnet' + const paymentAddress = paymentAddressFor(network) + if (!paymentAddress) { + reply.status(402).send({ error: `x402 payments are not configured for network "${network}"` }) + return + } + const { price, description } = GATED_ROUTES[matchedRoute] const paymentHeader = req.headers['x-payment'] as string | undefined const requirements = { scheme: 'exact' as const, price, - network: NETWORK, - payTo: PAYMENT_ADDRESS, + network: X402_NETWORK_LABEL[network], + payTo: paymentAddress, } // No payment header — return 402 with requirements @@ -76,6 +76,7 @@ async function x402Plugin(app: FastifyInstance) { payload = JSON.parse(paymentHeader) } + const resourceServer = await getX402ResourceServer(network, FACILITATOR_URL) const result = await resourceServer.verify(payload, requirements) if (!result.isValid) { reply.status(402).send({ error: 'Payment invalid', reason: result.invalidReason }) diff --git a/src/network/clients.ts b/src/network/clients.ts new file mode 100644 index 0000000..0841d03 --- /dev/null +++ b/src/network/clients.ts @@ -0,0 +1,47 @@ +/** + * Per-network Horizon / Soroban RPC client factories. + * + * Every ingester and aggregator that talks to a Stellar network needs a + * Horizon.Server or SorobanRpc.Server bound to that network's endpoints. + * Previously these were created once as module-level singletons bound to + * whatever network was active at import time, which made it impossible to + * run ingesters against more than one network in the same process. + * + * getHorizonServer(network) / getRpcServer(network) return a client cached + * per network so repeated calls for the same network reuse one instance, + * while different networks always resolve to distinct clients. + */ + +import { Horizon, rpc as SorobanRpc } from '@stellar/stellar-sdk' +import { getNetworkConfig, type NetworkName } from '../config' + +const horizonClients = new Map() +const rpcClients = new Map() + +export function getHorizonServer(network: NetworkName): Horizon.Server { + let client = horizonClients.get(network) + if (!client) { + client = new Horizon.Server(getNetworkConfig(network).horizon.url) + horizonClients.set(network, client) + } + return client +} + +export function getRpcServer(network: NetworkName): SorobanRpc.Server { + let client = rpcClients.get(network) + if (!client) { + client = new SorobanRpc.Server(getNetworkConfig(network).rpc.url, { allowHttp: true }) + rpcClients.set(network, client) + } + return client +} + +/** + * Test-only: drops the memoised clients so a test can rebind the mocked + * Horizon/RPC constructors between cases. Production code never calls this — + * clearing the cache mid-run would silently open new connections. + */ +export function resetNetworkClients(): void { + horizonClients.clear() + rpcClients.clear() +} diff --git a/src/pairsRegistry.ts b/src/pairsRegistry.ts index 1289036..3de2d78 100644 --- a/src/pairsRegistry.ts +++ b/src/pairsRegistry.ts @@ -1,4 +1,4 @@ -import { config } from './config' +import { config, activeNetwork } from './config' import { prisma } from './db' import type { WatchedPair, AssetId } from './types' @@ -41,8 +41,9 @@ export function makePairKey(a: AssetId, b: AssetId): string { /** Persist a new pair to DB so it survives restarts. */ export async function persistPair(pair: WatchedPair): Promise { await prisma.pairConfig.upsert({ - where: { pairKey: pair.pairKey }, + where: { network_pairKey: { network: activeNetwork, pairKey: pair.pairKey } }, create: { + network: activeNetwork, pairKey: pair.pairKey, assetACode: pair.assetA.code, assetAIssuer: pair.assetA.issuer, @@ -55,7 +56,7 @@ export async function persistPair(pair: WatchedPair): Promise { /** Load persisted pairs from DB and merge into the active registry. */ export async function loadPersistedPairs(): Promise { - const rows = await prisma.pairConfig.findMany() + const rows = await prisma.pairConfig.findMany({ where: { network: activeNetwork } }) for (const row of rows) { const pair: WatchedPair = { assetA: { code: row.assetACode, issuer: row.assetAIssuer }, diff --git a/src/redis.ts b/src/redis.ts index 7d6a97b..74e3a56 100644 --- a/src/redis.ts +++ b/src/redis.ts @@ -1,5 +1,5 @@ import Redis from 'ioredis' -import { config } from './config' +import { config, activeNetwork } from './config' export const redis = new Redis(config.redis.url, { maxRetriesPerRequest: 3, @@ -12,7 +12,7 @@ redis.on('error', (err) => { export async function getCachedPrice(pairKey: string): Promise { try { - return await redis.get(`lens:price:${pairKey}`) + return await redis.get(`lens:${activeNetwork}:price:${pairKey}`) } catch { return null } @@ -20,7 +20,7 @@ export async function getCachedPrice(pairKey: string): Promise { export async function setCachedPrice(pairKey: string, data: object, ttlSeconds: number): Promise { try { - await redis.set(`lens:price:${pairKey}`, JSON.stringify(data), 'EX', ttlSeconds) + await redis.set(`lens:${activeNetwork}:price:${pairKey}`, JSON.stringify(data), 'EX', ttlSeconds) } catch { // Redis cache miss is non-fatal } diff --git a/src/routes/discovery.ts b/src/routes/discovery.ts new file mode 100644 index 0000000..6b91b7e --- /dev/null +++ b/src/routes/discovery.ts @@ -0,0 +1,16 @@ +import type { FastifyInstance } from 'fastify' +import { parseDiscoveryFilters, queryDiscoveryResources } from '../bazaar/catalog' + +/** + * Registers GET /discovery/resources — the Bazaar catalog-browsing endpoint + * from the x402 `bazaar` extension (specs/extensions/bazaar.md in + * x402-foundation/x402). Public and un-gated: discovery has to work before a + * client has any payment method configured, the same reasoning that keeps + * GET /supported un-gated (see routes/facilitator.ts). + */ +export async function registerDiscoveryRoutes(app: FastifyInstance) { + app.get('/discovery/resources', { config: { public: true } }, async (req) => { + const filters = parseDiscoveryFilters(req.query as Record) + return queryDiscoveryResources(filters) + }) +} diff --git a/src/routes/webhooks.ts b/src/routes/webhooks.ts index 0e269eb..9eef277 100644 --- a/src/routes/webhooks.ts +++ b/src/routes/webhooks.ts @@ -1,6 +1,7 @@ import type { FastifyInstance } from 'fastify' import { randomBytes } from 'crypto' import { prisma } from '../db' +import { activeNetwork } from '../config' function isValidHttpsUrl(raw: string): boolean { try { @@ -38,6 +39,7 @@ export async function registerWebhookRoutes(app: FastifyInstance) { const webhook = await prisma.webhook.create({ data: { + network: activeNetwork, url, assetA: assetA.toUpperCase(), assetB: assetB.toUpperCase(), diff --git a/src/webhookDispatcher.ts b/src/webhookDispatcher.ts index 03d6d02..a68dbbd 100644 --- a/src/webhookDispatcher.ts +++ b/src/webhookDispatcher.ts @@ -1,4 +1,5 @@ import { prisma } from './db' +import { activeNetwork } from './config' import { buildThresholdAlertPayload, crossesThreshold, deliverJsonWithRetries } from './alerts' export interface PriceUpdate { @@ -13,6 +14,7 @@ export async function dispatchPriceUpdate(update: PriceUpdate): Promise { const webhooks = await prisma.webhook.findMany({ where: { + network: activeNetwork, assetA: assetA.toUpperCase(), assetB: assetB.toUpperCase(), }, diff --git a/src/ws/fanout.ts b/src/ws/fanout.ts index 6ac9608..4954827 100644 --- a/src/ws/fanout.ts +++ b/src/ws/fanout.ts @@ -1,5 +1,6 @@ import { EventEmitter } from 'events' import { redis } from '../redis' +import { activeNetwork } from '../config' import { priceEmitter, PRICE_UPDATE, PriceUpdateEvent } from '../events' import { register, Gauge, Counter } from 'prom-client' @@ -26,7 +27,7 @@ export const ws_messages_dropped_total = new Counter({ }) // ─── Configuration ───────────────────────────────────────────────────────────── -const REDIS_CHANNEL = 'lens:price:updates' +const REDIS_CHANNEL = `lens:${activeNetwork}:price:updates` const DEFAULT_COALESCE_INTERVAL_MS = 100 // max 10 updates/sec per client // ─── Client wrapper with backpressure & coalescing ──────────────────────────── diff --git a/src/x402/network.ts b/src/x402/network.ts new file mode 100644 index 0000000..b2e9aed --- /dev/null +++ b/src/x402/network.ts @@ -0,0 +1,53 @@ +import type { NetworkName } from '../config' +// @ts-ignore — @x402 packages ship ESM-only types incompatible with commonjs moduleResolution +import { x402ResourceServer, HTTPFacilitatorClient } from '@x402/core/server' +// @ts-ignore +import { ExactStellarScheme } from '@x402/stellar/exact/server' + +/** x402 chain identifier per Stellar network. */ +export const X402_NETWORK_LABEL: Record = { + testnet: 'stellar:testnet', + mainnet: 'stellar:pubnet', +} + +/** + * The x402 payment address for a given network. + * + * Resolution order (first non-empty wins), mirroring `config.ts`'s per-network + * env var convention: + * 1. `ORACLE_PAYMENT_ADDRESS_TESTNET` / `ORACLE_PAYMENT_ADDRESS_MAINNET` + * 2. `ORACLE_PAYMENT_ADDRESS` (back-compat with single-network setups) + */ +export function paymentAddressFor(network: NetworkName): string | undefined { + const suffix = network.toUpperCase() + return process.env[`ORACLE_PAYMENT_ADDRESS_${suffix}`] || process.env.ORACLE_PAYMENT_ADDRESS +} + +/** True if x402 gating should be active for at least one network. */ +export function isX402Configured(): boolean { + return Boolean(paymentAddressFor('testnet') || paymentAddressFor('mainnet')) +} + +// One resource server per network, built and initialised lazily on first use +// so a network that's never requested never pays the initialize() cost. +const resourceServers = new Map>() + +export function getX402ResourceServer(network: NetworkName, facilitatorUrl: string): Promise { + let pending = resourceServers.get(network) + if (!pending) { + pending = (async () => { + const facilitatorClient = new HTTPFacilitatorClient({ url: facilitatorUrl }) + const server: any = new x402ResourceServer(facilitatorClient) + .register(X402_NETWORK_LABEL[network], new ExactStellarScheme()) + await server.initialize() + return server + })() + resourceServers.set(network, pending) + } + return pending +} + +/** Test-only: clears the memoised resource servers between test cases. */ +export function _resetX402ResourceServers(): void { + resourceServers.clear() +} diff --git a/tests/aggregator.property.test.ts b/tests/aggregator.property.test.ts index a842e61..1e595d5 100644 --- a/tests/aggregator.property.test.ts +++ b/tests/aggregator.property.test.ts @@ -10,9 +10,11 @@ vi.mock('../src/db', () => ({ }, })) -vi.mock('@stellar/stellar-sdk', () => { +vi.mock('@stellar/stellar-sdk', async (importOriginal) => { + const actual = await importOriginal() const callFn = vi.fn() return { + ...actual, Horizon: { Server: vi.fn(function () { return { @@ -27,6 +29,13 @@ vi.mock('@stellar/stellar-sdk', () => { }), { native: vi.fn(() => 'native') } ), + // config.ts's buildNetworkConfig() falls back to these when no + // NETWORK_PASSPHRASE_* env var is set — needed now that getBestRoute + // resolves a per-network Horizon client via getNetworkConfig(). + Networks: { + PUBLIC: 'Public Global Stellar Network ; September 2015', + TESTNET: 'Test SDF Network ; September 2015', + }, __mockCall: callFn, } }) @@ -42,7 +51,11 @@ describe('Price aggregator property tests', () => { vi.clearAllMocks() }) - it('produces valid route results for random venue prices', async () => { + // 10,000 fast-check runs of getBestRoute now actually execute (previously + // this test failed before running a single iteration — the mocked + // @stellar/stellar-sdk had no Networks export, which getNetworkConfig() + // needs); that volume of real work needs more than the 5s default. + it('produces valid route results for random venue prices', { timeout: 30000 }, async () => { await fc.assert( fc.asyncProperty( fc.float({ min: 0, max: 2000, noNaN: true, noDefaultInfinity: true, noNegativeZero: true }), @@ -106,5 +119,5 @@ describe('Price aggregator property tests', () => { ), { numRuns: 10000 } ) - }) + }, 15000) }) diff --git a/tests/staleness.test.ts b/tests/staleness.test.ts index af2d6ba..2b54f0b 100644 --- a/tests/staleness.test.ts +++ b/tests/staleness.test.ts @@ -20,17 +20,25 @@ vi.mock('../src/aggregator/bestRoute', () => ({ getBestRoute: mockGetBestRoute, })) +const { stalenessTestPairs } = vi.hoisted(() => ({ + stalenessTestPairs: [ + { + pairKey: 'USDC/XLM', + assetA: { code: 'XLM', issuer: null }, + assetB: { code: 'USDC', issuer: 'GBBD47IF6LWK7P7MDEVSCWR7DPUWV3NY3DTQEVFL4NAT4AQH3ZLLFLA5' }, + }, + ], +})) + vi.mock('../src/config', () => ({ config: { - pairs: [ - { - pairKey: 'USDC/XLM', - assetA: { code: 'XLM', issuer: null }, - assetB: { code: 'USDC', issuer: 'GBBD47IF6LWK7P7MDEVSCWR7DPUWV3NY3DTQEVFL4NAT4AQH3ZLLFLA5' }, - }, - ], + pairs: stalenessTestPairs, cache: { priceTtl: 10 }, }, + activeNetwork: 'testnet', + getNetworkConfig: (network: string) => ({ + pairs: network === 'testnet' ? stalenessTestPairs : [], + }), })) import { registerRESTRoutes } from '../src/api/rest'