import { Hono, type Context } from 'hono'
import { AbiError, AbiEvent, type Hex } from 'ox'
import {
  type Address,
  BaseError,
  ContractFunctionRevertedError,
  decodeFunctionData,
  parseAbi,
  parseEventLogs,
} from 'viem'
import { Abis, Actions, Addresses, Tick } from 'viem/tempo'
import * as z from 'zod/mini'

import type * as App from '../../../App.js'
import * as Auth from '../../../internal/Auth.js'
import * as Cache from '../../../internal/Cache.js'
import * as Cursor from '../../../internal/Cursor.js'
import * as ExchangeProvider from '../../../internal/exchange/Provider.js'
import * as OpenApi from '../../../internal/OpenApi.js'
import * as Response from '../../../internal/Response.js'
import * as Schema from '../../../internal/Schema.js'
import * as Store from '../../../internal/Store.js'
import * as Timing from '../../../internal/Timing.js'
import * as Ttl from '../../../internal/Ttl.js'
import * as Value from '../../../internal/Value.js'
import * as VerifiedTokens from '../../../internal/VerifiedTokens.js'
import type * as Viem from '../../../internal/Viem.js'
import * as FxOracle from '../FxOracle.js'
import * as Tokens from './tokens.js'
import * as Valuation from './valuation.js'

// Stablecoin DEX precompile. Pair lifecycle is announced by `PairCreated` from
// this address; the indexer pre-decodes those into the `dex_pairs` table.
const stablecoinDex = Addresses.stablecoinDex.toLowerCase()

const contractPausedSelector = AbiError.getSelector(AbiError.fromAbi(Abis.tip20, 'ContractPaused'))

/** Largest amount accepted by TIP-20 and external uint256 provider contracts. */
const maxProviderAmount = 2n ** 256n - 1n

// `OrderCancelled(uint128 orderId)` — emitted when a maker cancels a resting
// order. Used by the orders endpoint to exclude cancelled orders from the
// resting set.
const orderCancelledSignature = 'event OrderCancelled(uint128 indexed orderId)'

// Topic0 hashes of the DEX order-state events. The swap feed resolves each
// fill's `(token, isBid, tick)` point-in-time from the raw `logs` event
// stream instead of the decoded `dex_orders` table: T5+ flip orders keep
// their `orderId`, mutate `(isBid, tick)` on every flip, and emit only
// `OrderFlipped` (no `OrderPlaced`), so `dex_orders` is both incomplete and
// stale for them.
const orderPlacedTopic = AbiEvent.getSelector(
  'event OrderPlaced(uint128 indexed orderId, address indexed maker, address indexed token, uint128 amount, bool isBid, int16 tick, bool isFlipOrder, int16 flipTick)',
)
const orderFlippedTopic = AbiEvent.getSelector(
  'event OrderFlipped(uint128 indexed orderId, address indexed maker, address indexed token, uint128 amount, bool isBid, int16 tick, int16 flipTick)',
)

// Block-range window for every `dex_fills` GROUP BY batch in the swaps feed.
// An unbounded grouped scan aggregates the entire fill history just to emit the
// top `limit + 1` groups, which TIDX rejects outright (HTTP 422) once the table
// grows large enough — so the scan is *always* bounded to a block window and
// the loop slides the window across batches until the page fills or the feed is
// exhausted. The window must comfortably hold a max-`limit` page (200 groups):
// 100k blocks (~28h) holds ~180k fills on Moderato. A group is keyed by
// `(block_num, tx_hash, taker)`, so a block-range bound never splits one group
// across windows; the value only tunes how many round-trips the sweep takes.
const swapWindowBlocks = 100_000

// Swap entrypoints of the stablecoin DEX. A swap's `mode` (exact-source vs
// exact-destination) is not in the events — it is recovered by decoding the
// transaction calldata (direct calls and AA type-`0x76` inner calls) against
// these signatures.
const swapAbi = parseAbi([
  'function swapExactAmountIn(address tokenIn, address tokenOut, uint128 amountIn, uint128 minAmountOut) returns (uint128)',
  'function swapExactAmountOut(address tokenIn, address tokenOut, uint128 amountOut, uint128 maxAmountIn) returns (uint128)',
])

// Example pair base token used by the OpenAPI route examples. Picked so the
// rendered URL addresses a realistic, well-known pair (USDC.e, quoted in
// pathUSD on-chain). Uses a fresh `Schema.tokenAddress(...)` so the example
// reaches the pipe input that path-param OpenAPI generation reads from.
const exampleBase = '0x20c000000000000000000000b9537d11c60e8b50' as const
const exampleSourceToken = '0x20c0000000000000000000008f5425160ebe5525' as const
const exampleApprovalCall = Actions.token.approve.call({
  amount: 1_000_000n,
  spender: Addresses.stablecoinDex,
  token: exampleSourceToken,
})
const exampleSwapCall = Actions.dex.sell.call({
  amountIn: 1_000_000n,
  minAmountOut: 995_000n,
  tokenIn: exampleSourceToken,
  tokenOut: exampleBase,
})

/**
 * Tick spacing enforced by the stablecoin DEX precompile: orders must be
 * placed at ticks divisible by 10 (a 1 bp price grid). Walking the depth at
 * this stride probes every valid level without wasting RPC calls on slots the
 * DEX rejects at order placement. Not currently exported by `ox/tempo`; the
 * upstream invariant is documented in the Tempo DEX spec.
 */
const tickSpacing = 10

/**
 * Default number of non-empty price levels returned per side when callers
 * omit `levels`. Wide enough to cover a typical depth chart's foreground while
 * keeping the payload compact; well-traded pairs that cluster liquidity near
 * peg fit comfortably inside this cap.
 */
const defaultDepthLevels = 50

/**
 * Hard ceiling on `levels`. The DEX tick range (`Tick.minTick`..`Tick.maxTick`)
 * exposes at most 201 valid positions per side at `tickSpacing=10`, so 200 is
 * an effective full-book cap and bounds the per-request multicall payload.
 */
const maxDepthLevels = 200

/**
 * Maximum number of buckets a single OHLC request may produce. Combined with
 * the OHLC `Interval` and `Window` enums this bounds the aggregation work
 * and the response size: e.g. `interval=1m, window=24h` produces 1440
 * buckets and is rejected as `query_invalid`. Hoisted above the schema
 * namespace because the OHLC `Query` refine references it at validation time.
 */
const maxBuckets = 500

/** Largest amount accepted by the DEX precompile's uint128 swap arguments. */
const maxSwapAmount = 2n ** 128n - 1n

/** Basis-point denominator used for caller-selected swap slippage. */
const slippageDenominator = 10_000n

const tokenFields = ['token.logoUri', 'token.verified'] as const

/** Converts the OHLC `interval` enum to a millisecond duration. */
function intervalMs(interval: '1m' | '5m' | '15m' | '1h' | '4h' | '1d'): number {
  switch (interval) {
    case '1m':
      return 60 * 1000
    case '5m':
      return 5 * 60 * 1000
    case '15m':
      return 15 * 60 * 1000
    case '1h':
      return 60 * 60 * 1000
    case '4h':
      return 4 * 60 * 60 * 1000
    case '1d':
      return 24 * 60 * 60 * 1000
  }
}

/**
 * Formats an ISO-8601 datetime string (e.g. `2026-06-04T12:00:00Z`) into the
 * `YYYY-MM-DD HH:MM:SS` form ClickHouse accepts for `DateTime64(3, 'UTC')`
 * comparisons. The ISO-Z form rejects with a conversion error on ClickHouse
 * `DateTime64`. Input is trusted (validated upstream by Zod / produced by
 * `Date.toISOString()` internally).
 */
function formatClickHouseDateTime(value: string): string {
  return value
    .replace('T', ' ')
    .replace(/\.\d+Z$/, '')
    .replace(/Z$/, '')
}

/** Converts the OHLC `window` enum to a millisecond duration. */
function windowMs(window: '1h' | '24h' | '7d' | '30d'): number {
  switch (window) {
    case '1h':
      return 60 * 60 * 1000
    case '24h':
      return 24 * 60 * 60 * 1000
    case '7d':
      return 7 * 24 * 60 * 60 * 1000
    case '30d':
      return 30 * 24 * 60 * 60 * 1000
  }
}

/** Zod schemas owned by the exchange handlers. */
export namespace schema {
  /**
   * A token in a trading pair: its contract address, plus the trimmed token
   * reference spread in when requested via `include=tokens`. Reference fields
   * are absent on the base (no-`include`) response and best-effort when
   * requested (a token whose metadata is unavailable keeps just `address`).
   */
  export const PairToken = OpenApi.component(
    Schema.describe(
      z.object({
        address: Schema.tokenAddress(Tokens.tokenExample.address).check(
          z.describe(
            'TIP-20 token contract address — the `0x`-prefixed 20-byte identifier for this token.',
          ),
          z.meta({ examples: [Tokens.tokenExample.address] }),
        ),
        currency: z
          .optional(z.string())
          .check(
            z.describe('Human-readable currency code for the token, when known.'),
            z.meta({ examples: [Tokens.tokenExample.currency] }),
          ),
        decimals: z
          .optional(z.number().check(z.int(), z.nonnegative()))
          .check(
            z.describe(
              'Number of decimal places the token uses; stablecoins on Tempo typically use 6.',
            ),
            z.meta({ examples: [Tokens.tokenExample.decimals] }),
          ),
        logoUri: z.optional(z.string()).check(
          z.describe('URL for the token logo image, when one is available.'),
          z.meta({
            examples: [Tokens.tokenExample.logoUri],
          }),
        ),
        name: z
          .optional(z.string())
          .check(
            z.describe('Human-readable token name, such as `USD Coin`.'),
            z.meta({ examples: [Tokens.tokenExample.name] }),
          ),
        symbol: z
          .optional(z.string())
          .check(
            z.describe('Short token ticker symbol, such as `USDC`.'),
            z.meta({ examples: [Tokens.tokenExample.symbol] }),
          ),
        verified: z
          .optional(z.boolean())
          .check(
            z.describe('Whether Tempo has verified this token metadata.'),
            z.meta({ examples: [true] }),
          ),
      }),
      'One side of a trading pair (with metadata when requested via `include=tokens`).',
    ),
    'ExchangePairToken',
  )

  const ValuedAmount = OpenApi.component(
    Schema.describe(
      z.extend(Schema.TokenAmount, {
        valuation: z
          .optional(z.nullable(Valuation.schema.Value))
          .check(
            z.describe(
              'The amount’s nominal value in the requested `valuation.currency`. `null` when the token is ' +
                'unverified, its display currency has no rate, or rates were unavailable.',
            ),
          ),
      }),
      'A token amount carrying its nominal value in the requested denomination.',
    ),
    'ExchangeValuedAmount',
  )

  /** Schemas for creating an executable exchange quote. */
  export namespace createQuote {
    const DestinationTokenAddress = Schema.tokenAddress(exampleBase)
    const SourceTokenAddress = Schema.tokenAddress(exampleSourceToken)

    /** Positive token amount accepted by exchange providers. */
    export const Amount = Schema.DecimalString.check(
      z.refine(
        (value) => {
          const amount = BigInt(value)
          return amount > 0n && amount <= maxProviderAmount
        },
        { error: 'Amount must be a positive uint256.' },
      ),
      z.describe(
        'Positive token amount in the token’s smallest unit, encoded as a decimal string.',
      ),
      z.meta({ examples: ['1000000'] }),
    )

    /** Optional token fields that require sources beyond RPC metadata. */
    export const Include = z
      .enum(tokenFields)
      .check(z.describe('Additional token fields to include in the quote response.'))

    /** Parses comma-separated optional token fields. */
    export const includeQuery = Schema.includeQuery(
      Include,
      'Comma-separated token fields to include, such as `token.logoUri,token.verified`.',
    )

    /** Query parameters selecting the Tempo chain and optional token fields. */
    export const Query = z
      .strictObject({
        chainId: Schema.ChainIdQuery,
        include: includeQuery,
        'valuation.currency': Schema.Denomination,
      })
      .check(z.describe('Query parameters for creating an exchange quote.'))

    /** Quote request with one pinned side and caller-selected slippage. */
    export const Request = OpenApi.component(
      z
        .strictObject({
          account: z
            .optional(Schema.Address)
            .check(
              z.describe('Wallet that will sign and execute the swap.'),
              z.meta({ examples: ['0x1111111111111111111111111111111111111111'] }),
            ),
          amount: Amount.check(
            z.describe(
              'Pinned amount in base units: source amount for `exactSource`, destination amount for `exactDestination`.',
            ),
          ),
          destinationToken: DestinationTokenAddress.check(
            z.describe('TIP-20 token the swap receives.'),
          ),
          mode: z
            .enum(['exactSource', 'exactDestination'])
            .check(
              z.describe('Which side of the swap keeps the requested amount exact.'),
              z.meta({ examples: ['exactSource'] }),
            ),
          slippageBps: z
            .number()
            .check(
              z.int(),
              z.gte(0),
              z.lt(10_000),
              z.describe('Allowed execution slippage in basis points, where 100 is 1%.'),
              z.meta({ examples: [50] }),
            ),
          sourceToken: SourceTokenAddress.check(z.describe('TIP-20 token the swap spends.')),
        })
        .check(
          z.refine((request) => request.sourceToken !== request.destinationToken, {
            error: 'Source and destination tokens must differ.',
          }),
          z.describe('Request for an executable quote from the configured exchange providers.'),
        ),
      'CreateExchangeQuoteRequest',
    )

    /** One call in the unsigned Tempo transaction plan. */
    export const Call = OpenApi.component(
      z
        .object({
          data: Schema.Hex.check(
            z.describe('ABI-encoded call data.'),
            z.meta({ examples: ['0x095ea7b3'] }),
          ),
          to: Schema.Address.check(
            z.describe('Contract or precompile that receives the call.'),
            z.meta({ examples: [Addresses.stablecoinDex] }),
          ),
          value: Schema.Quantity.check(
            z.describe('Native token value sent with the call.'),
            z.meta({ examples: ['0x0'] }),
          ),
        })
        .check(z.describe('One unsigned call in the Tempo transaction plan.')),
      'ExchangeQuoteCall',
    )

    const QuoteToken = Schema.describe(
      z.partial(z.omit(Tokens.schema.Token, { id: true }), {
        logoUri: true,
        verified: true,
      }),
      'A token referenced by an exchange quote, with RPC metadata and optional enrichments.',
    )

    const DestinationToken = OpenApi.component(
      Schema.describe(
        z.extend(QuoteToken, {
          address: DestinationTokenAddress.check(z.describe('TIP-20 token the swap receives.')),
        }),
        'The token received by an exchange quote.',
      ),
      'ExchangeQuoteDestinationToken',
    )

    const SourceToken = OpenApi.component(
      Schema.describe(
        z.extend(QuoteToken, {
          address: SourceTokenAddress.check(z.describe('TIP-20 token the swap spends.')),
        }),
        'The token spent by an exchange quote.',
      ),
      'ExchangeQuoteSourceToken',
    )

    export const Transaction = OpenApi.component(
      z
        .object({
          calls: z.array(Call).check(
            z.minLength(1),
            z.describe('Calls to execute in order.'),
            z.meta({
              examples: [
                [
                  {
                    data: exampleApprovalCall.data,
                    to: exampleApprovalCall.to,
                    value: '0x0',
                  },
                  { data: exampleSwapCall.data, to: exampleSwapCall.to, value: '0x0' },
                ],
              ],
            }),
          ),
          chainId: Schema.ChainId.check(z.describe('Tempo chain where these calls execute.')),
        })
        .check(z.describe('Unsigned Tempo transaction plan.')),
      'ExchangeQuoteTransaction',
    )

    /** Response-level valuation rate provenance. */
    export const Meta = OpenApi.component(
      z
        .object({
          valuation: z
            .optional(Valuation.schema.Pricing)
            .check(
              z.describe(
                'Rate provenance for amount valuations. Present when conversion rates were consulted.',
              ),
            ),
        })
        .check(z.describe('Response-level resources attached to this quote.')),
      'ExchangeQuoteMeta',
    )

    /** Provider-neutral EIP-712 payload to sign before finalizing a quote. */
    export const TypedData = OpenApi.component(
      z
        .object({
          domain: z.record(z.string(), z.unknown()).check(
            z.describe('EIP-712 signing domain.'),
            z.meta({
              examples: [
                {
                  chainId: 4217,
                  name: 'Permit2',
                  verifyingContract: '0x000000000022d473030f116ddee9f6b43ac78ba3',
                },
              ],
            }),
          ),
          message: z.record(z.string(), z.unknown()).check(
            z.describe('Values encoded by the primary EIP-712 type.'),
            z.meta({
              examples: [
                {
                  details: {
                    amount: '1000000',
                    expiration: '4102444800',
                    nonce: '0',
                    token: exampleSourceToken,
                  },
                  sigDeadline: '4102444800',
                  spender: '0x1febb76be10aaf3a1402f04e8e835f2c382f7914',
                },
              ],
            }),
          ),
          primaryType: z
            .string()
            .check(z.minLength(1))
            .check(z.describe('Root EIP-712 type.'), z.meta({ examples: ['PermitSingle'] })),
          types: z
            .record(
              z.string(),
              z.array(
                z.strictObject({
                  name: z.string(),
                  type: z.string(),
                }),
              ),
            )
            .check(
              z.describe('EIP-712 type definitions keyed by type name.'),
              z.meta({
                examples: [
                  {
                    PermitDetails: [
                      { name: 'token', type: 'address' },
                      { name: 'amount', type: 'uint160' },
                      { name: 'expiration', type: 'uint48' },
                      { name: 'nonce', type: 'uint48' },
                    ],
                    PermitSingle: [
                      { name: 'details', type: 'PermitDetails' },
                      { name: 'spender', type: 'address' },
                      { name: 'sigDeadline', type: 'uint256' },
                    ],
                  },
                ],
              }),
            ),
        })
        .check(z.describe('EIP-712 payload the wallet must sign before execution can continue.')),
      'ExchangeQuoteTypedData',
    )

    const Provider = z
      .string()
      .check(z.describe('Provider selected for this route.'), z.meta({ examples: ['nativeDex'] }))

    const Quote = {
      gasEstimate: Schema.DecimalString.check(
        z.describe('Estimated network fee in the chain native token’s base unit.'),
        z.meta({ examples: ['150000000000000'] }),
      ),
      provider: Provider,
    }

    const Approval = OpenApi.component(
      z
        .object({
          calls: z.array(Call).check(
            z.minLength(1),
            z.describe('Approval calls to confirm before requesting a fresh quote.'),
            z.meta({
              examples: [
                [
                  {
                    data: exampleApprovalCall.data,
                    to: exampleApprovalCall.to,
                    value: '0x0',
                  },
                ],
              ],
            }),
          ),
        })
        .check(z.describe('Token approval calls required before requesting a fresh quote.')),
      'ExchangeQuoteApproval',
    )

    const ApprovalRequired = {
      approval: Approval,
      status: z
        .literal('approvalRequired')
        .check(
          z.describe('Confirm the approvals, then request a fresh quote.'),
          z.meta({ examples: ['approvalRequired'] }),
        ),
    }

    const Ready = {
      status: z
        .literal('ready')
        .check(
          z.describe('The returned calls can be executed now.'),
          z.meta({ examples: ['ready'] }),
        ),
      transaction: Transaction,
    }

    const SignatureRequired = {
      continuation: z
        .string()
        .check(
          z.minLength(1),
          z.describe('Opaque state supplied to the finalization endpoint within 30 seconds.'),
          z.meta({ examples: ['eyJwcm92aWRlciI6InVuaXN3YXAifQ'] }),
        ),
      status: z
        .literal('signatureRequired')
        .check(
          z.describe('Sign the typed data before final calldata can be built.'),
          z.meta({ examples: ['signatureRequired'] }),
        ),
      typedData: TypedData,
    }

    const ExactSource = {
      destinationAmount: ValuedAmount,
      destinationAmountMin: ValuedAmount,
      destinationToken: DestinationToken,
      meta: z.optional(Meta).check(z.describe('Valuation rate provenance for this quote.')),
      mode: z
        .literal('exactSource')
        .check(
          z.describe('The source amount is fixed and the destination has a minimum.'),
          z.meta({ examples: ['exactSource'] }),
        ),
      sourceAmount: ValuedAmount,
      sourceToken: SourceToken,
    }

    const ExactDestination = {
      destinationAmount: ValuedAmount,
      destinationToken: DestinationToken,
      meta: z.optional(Meta).check(z.describe('Valuation rate provenance for this quote.')),
      mode: z
        .literal('exactDestination')
        .check(
          z.describe('The destination amount is fixed and the source has a maximum.'),
          z.meta({ examples: ['exactDestination'] }),
        ),
      sourceAmount: ValuedAmount,
      sourceAmountMax: ValuedAmount,
      sourceToken: SourceToken,
    }

    const ExactSourceApprovalRequired = OpenApi.component(
      z
        .object({ ...ApprovalRequired, ...ExactSource, ...Quote })
        .check(z.describe('An exact-source quote that requires token approval.')),
      'ExchangeQuoteExactSourceApprovalRequired',
    )
    const ExactSourceReady = OpenApi.component(
      z
        .object({ ...ExactSource, ...Quote, ...Ready })
        .check(z.describe('An exact-source quote ready to execute.')),
      'ExchangeQuoteExactSourceReady',
    )
    const ExactSourceSignatureRequired = OpenApi.component(
      z
        .object({ ...ExactSource, ...Quote, ...SignatureRequired })
        .check(z.describe('An exact-source quote that requires a signature.')),
      'ExchangeQuoteExactSourceSignatureRequired',
    )
    const ExactDestinationApprovalRequired = OpenApi.component(
      z
        .object({ ...ApprovalRequired, ...ExactDestination, ...Quote })
        .check(z.describe('An exact-destination quote that requires token approval.')),
      'ExchangeQuoteExactDestinationApprovalRequired',
    )
    const ExactDestinationReady = OpenApi.component(
      z
        .object({ ...ExactDestination, ...Quote, ...Ready })
        .check(z.describe('An exact-destination quote ready to execute.')),
      'ExchangeQuoteExactDestinationReady',
    )
    const ExactDestinationSignatureRequired = OpenApi.component(
      z
        .object({ ...ExactDestination, ...Quote, ...SignatureRequired })
        .check(z.describe('An exact-destination quote that requires a signature.')),
      'ExchangeQuoteExactDestinationSignatureRequired',
    )

    /** Provider-neutral quote and the action needed to execute it. */
    export const Response = OpenApi.component(
      z
        .union([
          ExactSourceApprovalRequired,
          ExactSourceReady,
          ExactSourceSignatureRequired,
          ExactDestinationApprovalRequired,
          ExactDestinationReady,
          ExactDestinationSignatureRequired,
        ])
        .check(z.describe('Provider-neutral exchange quote with its next execution action.')),
      'ExchangeQuote',
    )
  }

  /** Schemas for finishing a quote after a provider-requested signature. */
  export namespace executeQuote {
    /** Signed continuation request. */
    export const Request = OpenApi.component(
      z
        .strictObject({
          account: Schema.Address.check(
            z.describe('Wallet that signed and will execute the swap.'),
            z.meta({ examples: ['0x1111111111111111111111111111111111111111'] }),
          ),
          continuation: z
            .string()
            .check(
              z.minLength(1),
              z.maxLength(100_000),
              z.describe('Opaque state returned by the quote request.'),
              z.meta({ examples: ['eyJwcm92aWRlciI6InVuaXN3YXAifQ'] }),
            ),
          provider: z
            .string()
            .check(
              z.minLength(1),
              z.describe('Provider that issued the continuation.'),
              z.meta({ examples: ['uniswap'] }),
            ),
          signature: Schema.Hex.check(
            z.refine((value) => value.length === 132, { error: 'Signature must be 65 bytes.' }),
            z.describe('65-byte EIP-712 signature requested by the quote.'),
            z.meta({ examples: [`0x${'11'.repeat(65)}`] }),
          ),
        })
        .check(z.describe('Request to finalize a provider quote after signing its typed data.')),
      'FinalizeExchangeQuoteRequest',
    )

    /** Final transaction plan. */
    export const Response = OpenApi.component(
      z
        .object({
          provider: z
            .string()
            .check(
              z.describe('Provider that built the transaction.'),
              z.meta({ examples: ['uniswap'] }),
            ),
          transaction: createQuote.Transaction,
        })
        .check(z.describe('Final unsigned transaction plan for a signed provider quote.')),
      'FinalizedExchangeQuote',
    )
  }

  /** Schemas for the getPairs operation. */
  export namespace getPairs {
    /**
     * Optional resources for the pair list: the per-row `tokens` embed plus the
     * response-wide capped `totalCount`. The detail route keeps its own
     * `tokens`-only {@link getPair.Include}.
     */
    export const Include = z
      .enum(['tokens', 'totalCount'])
      .check(
        z.describe(
          'Extra resources you can request with `include`, such as token metadata or a total count.',
        ),
      )

    /**
     * Parses a comma-separated `include` query value into a list of optional
     * resources to embed. Extra lookups (e.g. token metadata) only run when
     * explicitly requested, keeping the base pair page fast.
     */
    export const includeQuery = Schema.includeQuery(
      Include,
      'Comma-separated optional resources to embed, e.g. `tokens,totalCount`.',
    )

    /** Query parameters for trading pair list requests. */
    export const Query = z
      .strictObject({
        chainId: Schema.ChainIdQuery,
        cursor: Schema.Cursor,
        include: includeQuery,
        limit: Schema.Limit,
        order: Schema.Order,
        page: Schema.Page,
        sort: z
          ._default(z.enum(['created', 'liquidity']), 'created')
          .check(
            z.describe(
              'Sort key: `created` orders by when the pair was created, while `liquidity` ranks by the DEX-held base-token balance as a practical liquidity signal.',
            ),
            z.meta({ examples: ['created'] }),
          ),
      })
      .check(
        ...Schema.pageChecks(),
        z.describe('Filters, pagination, and sorting options for listing exchange trading pairs.'),
      )

    /** A single trading pair on the stablecoin DEX. */
    export const Pair = OpenApi.component(
      z
        .object({
          base: PairToken,
          blockNumber: z
            .number()
            .check(
              z.int(),
              z.nonnegative(),
              z.describe('Block number where the pair was created.'),
              z.meta({ examples: [23456789] }),
            ),
          id: Schema.Hash.check(
            z.describe(
              'Stable API id for this pair; it is the same value as the onchain pair key.',
            ),
          ),
          key: Schema.Hash.check(
            z.describe(
              'Stable onchain pair identifier returned by the DEX precompile as `pairKey`.',
            ),
          ),
          liquidity: z
            .optional(
              z
                .string()
                .check(
                  z.regex(/^\d+$/),
                  z.describe(
                    'Liquidity signal for the pair: the base-token balance held by the DEX, returned as a decimal integer string in the token’s smallest units. Present only when `sort=liquidity`.',
                  ),
                ),
            )
            .check(
              z.describe(
                'Liquidity signal for the pair, present only when you request `sort=liquidity`.',
              ),
              z.meta({ examples: ['1000000'] }),
            ),
          quote: PairToken,
          timestamp: z.iso
            .datetime()
            .check(
              z.describe('Block timestamp when the pair was created.'),
              z.meta({ examples: ['2024-01-01T00:00:00Z'] }),
            ),
          transactionHash: Schema.Hash.check(
            z.describe('Transaction hash for the transaction that created the pair.'),
          ),
        })
        .check(z.describe('One trading pair on Tempo’s built-in stablecoin exchange.')),
      'ExchangePair',
    )

    /** Response metadata requested through `include`, such as capped counts. */
    export const Meta = OpenApi.component(Schema.CountMeta, 'ExchangePairListMeta')

    /** Page of trading pairs on the stablecoin DEX, ordered by creation. */
    export const Response = OpenApi.component(
      z
        .object({
          data: z.array(Pair).check(z.describe('Trading pairs returned on this page.')),
          meta: z
            .optional(Meta)
            .check(
              z.describe(
                'Response-level resources requested with `include`, such as `totalCount`.',
              ),
            ),
          nextCursor: Schema.NextCursor,
        })
        .check(
          z.describe('A paginated list of trading pairs on Tempo’s built-in stablecoin exchange.'),
        ),
      'ExchangePairList',
    )
  }

  /** Schemas for the getPair operation (single trading pair detail). */
  export namespace getPair {
    /**
     * Path parameter for pair detail requests. On-chain, a pair's quote is
     * intrinsic to its base token (the precompile resolves it from the TIP-20's
     * `quote_token()`), so the base address alone identifies the pair.
     */
    export const Params = z
      .object({
        base: Schema.tokenAddress(exampleBase).check(
          z.describe('Base token address for the trading pair.'),
        ),
      })
      .check(z.describe('Path parameters for looking up a trading pair.'))

    /** Optional related resource that callers opt into via `include`. */
    export const Include = z
      .enum(['tokens'])
      .check(z.describe('Related resource you can opt into with `include`.'))

    /**
     * Parses a comma-separated `include` query value into a list of optional
     * resources to embed.
     */
    export const includeQuery = Schema.includeQuery(
      Include,
      'Comma-separated optional resources to embed, e.g. `tokens`.',
    )

    /** Query parameters for pair detail requests. */
    export const Query = z
      .strictObject({
        chainId: Schema.ChainIdQuery,
        include: includeQuery,
      })
      .check(z.describe('Query options for fetching one trading pair.'))

    /** A single trading pair, identical in shape to `GET /exchange/pairs` rows. */
    export const Response = getPairs.Pair
  }

  /** Schemas for the getSwaps operation (the swap feed, optionally pair-scoped). */
  export namespace getSwaps {
    /** Optional token fields that require sources beyond RPC metadata. */
    export const Include = z
      .enum(tokenFields)
      .check(z.describe('Additional token fields to include in swap responses.'))

    /** Parses comma-separated optional token fields. */
    export const includeQuery = Schema.includeQuery(
      Include,
      'Comma-separated token fields to include, such as `token.logoUri,token.verified`.',
    )

    /** Query parameters for swap feed requests. */
    export const Query = z
      .strictObject({
        'blockNumber.from': Schema.blockNumberBound('swaps', 'from'),
        'blockNumber.to': Schema.blockNumberBound('swaps', 'to'),
        chainId: Schema.ChainIdQuery,
        cursor: Schema.Cursor,
        include: includeQuery,
        limit: Schema.Limit,
        maker: z
          .optional(Schema.Address)
          .check(
            z.describe(
              'Only include swaps that filled against an order owned by this maker address.',
            ),
          ),
        order: Schema.Order,
        participant: z
          .optional(Schema.Address)
          .check(
            z.describe(
              'Only include swaps where this address participated as the taker or as a maker whose order was filled.',
            ),
          ),
        taker: z
          .optional(Schema.Address)
          .check(z.describe('Only include swaps initiated by this taker address.')),
        'timestamp.from': Schema.timestampBound('swaps', 'from'),
        'timestamp.to': Schema.timestampBound('swaps', 'to'),
        transactionHash: z
          .optional(Schema.Hash)
          .check(z.describe('Only include swaps included in this transaction hash.')),
        'valuation.currency': Schema.Denomination,
      })
      .check(
        z.describe(
          'Query parameters for swap feed requests. Cursor-paginated only — a batched ' +
            'transaction expands one fill group into a variable number of swap rows, so a ' +
            'positional `page` has no stable meaning on this feed.',
        ),
      )

    /** One swap-side token with RPC metadata and optional curated fields. */
    export const SwapSide = OpenApi.component(
      Schema.describe(Tokens.schema.TokenReference, 'A token referenced by one side of a swap.'),
      'ExchangeSwapToken',
    )

    /** A single maker-order fill within a swap, framed from the taker's perspective. */
    export const Fill = OpenApi.component(
      z
        .object({
          destinationAmount: ValuedAmount,
          destinationToken: SwapSide,
          logIndex: z
            .number()
            .check(
              z.int(),
              z.nonnegative(),
              z.describe('Log index of this fill within its block.'),
              z.meta({ examples: [0] }),
            ),
          maker: Schema.Address.check(z.describe('Maker address whose resting order was filled.')),
          orderId: Schema.DecimalString.check(
            z.describe('On-chain maker order id, returned as a decimal string.'),
          ),
          partialFill: z
            .boolean()
            .check(
              z.describe('Whether this fill used only part of the maker order.'),
              z.meta({ examples: [false] }),
            ),
          price: z
            .string()
            .check(
              z.regex(/^\d+(\.\d+)?$/),
              z.describe(
                'Quote-per-base price at fill time, returned as a fixed-decimal string with 5 decimal places. This value is direction-independent and works well for charts.',
              ),
              z.meta({ examples: ['1.00000'] }),
            ),
          sourceAmount: ValuedAmount,
          sourceToken: SwapSide,
        })
        .check(
          z.describe('One maker-order fill within a swap, described from the taker’s perspective.'),
        ),
      'ExchangeSwapFill',
    )

    /**
     * A single logical taker swap on the stablecoin DEX: every maker-order
     * fill one taker's swap produced in one transaction, assembled into a
     * source → destination row. Multi-hop swaps (routed through an
     * intermediate token, e.g. pathUSD) are one row; the intermediate hops
     * are visible via `route` and `fills`.
     */
    export const Swap = OpenApi.component(
      z
        .object({
          blockNumber: z
            .number()
            .check(
              z.int(),
              z.nonnegative(),
              z.describe('Block number where the swap was included.'),
              z.meta({ examples: [23456789] }),
            ),
          destinationAmount: ValuedAmount,
          destinationToken: SwapSide,
          filledAt: z.iso
            .datetime()
            .check(
              z.describe('Block timestamp when the swap was filled onchain.'),
              z.meta({ examples: ['2024-01-01T00:00:00Z'] }),
            ),
          fills: z
            .array(Fill)
            .check(
              z.describe(
                'Maker-order fills that make up this swap, ordered by execution `logIndex`. Exact-destination swaps can execute hops in reverse, so use `route` for the source-to-destination path.',
              ),
            ),
          id: z.string().check(
            z.describe(
              'Stable API id built from the lowercase transaction hash and the first fill log index.',
            ),
            z.meta({
              examples: ['0x515801d7f9a5ac705e793e85904c9c69b3f1694b465cc8fb6ba3f0298dc82665-0'],
            }),
          ),
          logIndex: z
            .number()
            .check(
              z.int(),
              z.nonnegative(),
              z.describe('Block-wide log index for the swap, taken from its first fill.'),
              z.meta({ examples: [0] }),
            ),
          mode: z
            .nullable(z.enum(['exactSource', 'exactDestination']))
            .check(
              z.describe(
                'The amount type fixed by the swap call: `exactSource` for `swapExactAmountIn` or `exactDestination` for `swapExactAmountOut`. This is recovered from transaction calldata and is `null` when the swap did not come from a decodable swap call.',
              ),
              z.meta({ examples: ['exactSource'] }),
            ),
          rate: z
            .string()
            .check(
              z.regex(/^\d+(\.\d+)?$/),
              z.describe(
                'Swap-level effective rate, calculated as destination amount divided by source amount from the taker’s perspective and returned with 5 decimal places. Compare it with `1` to see distance from peg.',
              ),
              z.meta({ examples: ['1.00000'] }),
            ),
          route: z
            .array(Schema.TokenAddress)
            .check(
              z.describe(
                'Token path for the swap, from source through any intermediate tokens to destination. `route.length - 1` is the hop count; two entries means a direct swap.',
              ),
            ),
          sourceAmount: ValuedAmount,
          sourceToken: SwapSide,
          taker: Schema.Address.check(z.describe('Taker address that initiated the swap.')),
          transactionHash: Schema.Hash.check(
            z.describe('Transaction hash for the transaction containing the swap.'),
          ),
        })
        .check(z.describe('One logical taker swap on Tempo’s built-in stablecoin exchange.')),
      'ExchangeSwap',
    )

    /**
     * Page of swaps. A row is one logical taker swap — the fills of one
     * `(transaction, taker)` group chained source → destination — so the
     * unscoped and token-filtered feeds share one shape. `limit` caps the
     * rows per page; a transaction batching several unrelated swaps by one
     * taker may have its rows split across consecutive pages.
     */
    /** Page-level resources: valuation rate provenance. */
    export const Meta = OpenApi.component(
      z
        .object({
          valuation: z
            .optional(Valuation.schema.Pricing)
            .check(
              z.describe(
                'Rate provenance for amount valuations. Present when conversion rates were ' +
                  'consulted; absent when every value was identity-valued or rates were unavailable.',
              ),
            ),
        })
        .check(z.describe('Page-level resources attached to this response.')),
      'ExchangeSwapListMeta',
    )

    export const Response = OpenApi.component(
      z
        .object({
          data: z.array(Swap).check(z.describe('Swaps returned on this page.')),
          meta: z
            .optional(Meta)
            .check(z.describe('Page-level resources, such as valuation rate provenance.')),
          nextCursor: Schema.NextCursor,
        })
        .check(z.describe('A paginated list of swaps on Tempo’s built-in stablecoin exchange.')),
      'ExchangeSwapList',
    )
  }

  /** Schemas for the getPairOhlc operation (candlestick aggregations). */
  export namespace getPairOhlc {
    /**
     * Path parameter mirrors `getPair.Params`: the base address alone
     * identifies the pair (its quote is intrinsic to the base token on-chain).
     */
    export const Params = z
      .object({
        base: Schema.tokenAddress(exampleBase).check(
          z.describe('Base token address for the trading pair.'),
        ),
      })
      .check(z.describe('Path parameters for fetching OHLC candles for a pair.'))

    /** Bucket size for the candlestick aggregation. */
    export const Interval = z
      .enum(['1m', '5m', '15m', '1h', '4h', '1d'])
      .check(z.describe('Time size for each OHLC candle bucket.'), z.meta({ examples: ['1h'] }))

    /** Rolling lookback window for the candlestick aggregation. */
    export const Window = z
      .enum(['1h', '24h', '7d', '30d'])
      .check(
        z.describe('Rolling time window covered by the OHLC candles.'),
        z.meta({ examples: ['24h'] }),
      )

    /**
     * Optional related resource that callers opt into via `include`. Currently
     * only `tokens` (embeds metadata onto `base`/`quote` instead of returning
     * just the addresses).
     */
    export const Include = z
      .enum(['tokens'])
      .check(z.describe('Related resource you can opt into with `include`.'))

    /**
     * Parses a comma-separated `include` query value into a list of optional
     * resources to embed.
     */
    export const includeQuery = Schema.includeQuery(
      Include,
      'Comma-separated optional resources to embed, e.g. `tokens`.',
    )

    /** Query parameters for pair OHLC requests. */
    export const Query = z
      .strictObject({
        chainId: Schema.ChainIdQuery,
        include: includeQuery,
        interval: z._default(Interval, '1h').check(z.meta({ examples: ['1h'] })),
        window: z._default(Window, '24h').check(z.meta({ examples: ['24h'] })),
      })
      .check(
        z.refine((query) => windowMs(query.window) / intervalMs(query.interval) <= maxBuckets, {
          error: `Requested \`window\` / \`interval\` produces more than ${maxBuckets} buckets. Pick a longer \`interval\` or a shorter \`window\`.`,
          path: ['interval'],
        }),
      )
      .check(z.describe('Query options for fetching OHLC candles for a pair.'))

    /**
     * A single candlestick bucket. Open/close are picked from the earliest
     * and latest fill in the bucket (ordered by `(block_num, log_idx)`); high
     * and low are the extrema across all fills in the bucket. Volumes are
     * the sum of per-fill base- and quote-side amounts in base units.
     */
    export const Bucket = OpenApi.component(
      z
        .object({
          close: z
            .string()
            .check(
              z.regex(/^\d+(\.\d+)?$/),
              z.describe(
                'Close price for the bucket: the latest fill rate, returned with 5 decimal places.',
              ),
              z.meta({ examples: ['1.00000'] }),
            ),
          fillCount: z
            .number()
            .check(
              z.int(),
              z.nonnegative(),
              z.describe('Number of fills included in this bucket.'),
              z.meta({ examples: [5] }),
            ),
          high: z
            .string()
            .check(
              z.regex(/^\d+(\.\d+)?$/),
              z.describe(
                'High price for the bucket: the highest fill rate, returned with 5 decimal places.',
              ),
              z.meta({ examples: ['1.01000'] }),
            ),
          id: z.string().check(
            z.describe(
              'Stable API id built from the pair key, candle interval, and bucket start time in Unix seconds.',
            ),
            z.meta({
              examples: [
                '0x515801d7f9a5ac705e793e85904c9c69b3f1694b465cc8fb6ba3f0298dc82665-1h-1704067200',
              ],
            }),
          ),
          low: z
            .string()
            .check(
              z.regex(/^\d+(\.\d+)?$/),
              z.describe(
                'Low price for the bucket: the lowest fill rate, returned with 5 decimal places.',
              ),
              z.meta({ examples: ['0.99000'] }),
            ),
          open: z
            .string()
            .check(
              z.regex(/^\d+(\.\d+)?$/),
              z.describe(
                'Open price for the bucket: the earliest fill rate, returned with 5 decimal places.',
              ),
              z.meta({ examples: ['1.00000'] }),
            ),
          timestamp: z.iso
            .datetime()
            .check(
              z.describe('ISO 8601 timestamp for the start of this candle bucket.'),
              z.meta({ examples: ['2024-01-01T00:00:00Z'] }),
            ),
          volume: z
            .object({
              base: Schema.DecimalString.check(
                z.describe(
                  'Total base-side amount filled in this bucket, in the token’s smallest units.',
                ),
              ),
              quote: Schema.DecimalString.check(
                z.describe(
                  'Total quote-side amount filled in this bucket, reconstructed from each fill’s tick and returned in the token’s smallest units.',
                ),
              ),
            })
            .check(z.describe('Total filled volume in this bucket, split by pair side.')),
        })
        .check(z.describe('One OHLC candle bucket for charting a trading pair.')),
      'ExchangeOhlcBucket',
    )

    /** Page of candlestick buckets for a single trading pair. */
    export const Response = OpenApi.component(
      z
        .object({
          base: PairToken,
          data: z
            .array(Bucket)
            .check(
              z.describe(
                'OHLC candle buckets ordered oldest to newest; empty buckets are omitted.',
              ),
            ),
          interval: Interval,
          quote: PairToken,
          truncated: z
            .boolean()
            .check(
              z.describe(
                '`true` when the fill scan hit its hard cap, meaning OHLC values were computed from only the most recent fills.',
              ),
              z.meta({ examples: [false] }),
            ),
          window: Window,
        })
        .check(z.describe('OHLC candle data for one trading pair over a rolling time window.')),
      'ExchangePairOhlc',
    )
  }

  /** Schemas shared by the order detail and order fills handlers. */
  export namespace order {
    /** Path parameters for any single-order route (`/orders/:orderId`). */
    export const Params = z
      .object({
        orderId: Schema.DecimalString.check(
          z.describe('On-chain order id, returned as a decimal string.'),
          z.meta({ examples: ['1'] }),
        ),
      })
      .check(z.describe('Path parameters for routes that look up one order.'))

    /** The pair the order is placed against. */
    export const Pair = OpenApi.component(
      Schema.describe(
        z.object({
          base: PairToken,
          key: Schema.Hash.check(z.describe('On-chain key for this `(base, quote)` trading pair.')),
          quote: PairToken,
        }),
        'The trading pair the order belongs to.',
      ),
      'ExchangeOrderPair',
    )
  }

  /** Schemas for the getOrder operation. */
  export namespace getOrder {
    /** Path parameters for single-order requests. */
    export const Params = order.Params

    /**
     * Optional related resource that callers opt into via `include`. Currently
     * only `tokens` (embeds metadata for both pair tokens on `pair.base` and
     * `pair.quote`).
     */
    export const Include = z
      .enum(['tokens'])
      .check(z.describe('Related resource you can opt into with `include`.'))

    /**
     * Parses a comma-separated `include` query value into a list of optional
     * resources to embed. Extra lookups (e.g. pair-token metadata) only run
     * when explicitly requested, keeping the base order response fast.
     */
    export const includeQuery = Schema.includeQuery(
      Include,
      'Comma-separated optional resources to embed, e.g. `tokens`.',
    )

    /** Query parameters for single-order requests. */
    export const Query = z
      .strictObject({
        chainId: Schema.ChainIdQuery,
        include: includeQuery,
      })
      .check(z.describe('Query options for fetching one order.'))

    /**
     * A single DEX order. Placement parameters and current state are sourced
     * directly from the on-chain `dex.getOrder` view, so `remaining` reflects
     * live state (not indexer lag) and the response is a single fast RPC.
     *
     * Fills are paginated separately under
     * `GET /exchange/orders/{orderId}/fills`; the per-order fills feed grows
     * over time and would otherwise force this resource to revalidate on
     * every new fill.
     */
    export const Response = OpenApi.component(
      Schema.describe(
        z.object({
          amount: Schema.DecimalString.check(
            z.describe(
              'Initial order size, returned as a decimal integer string in base-token smallest units.',
            ),
          ),
          flipTick: z
            .number()
            .check(
              z.int(),
              z.describe(
                'Replacement `tick` used if this order auto-flips into a counter-order after filling.',
              ),
              z.meta({ examples: [10] }),
            ),
          id: Schema.DecimalString.check(
            z.describe(
              'Stable API id for this order; it is the same value as the onchain order id.',
            ),
          ),
          isBid: z
            .boolean()
            .check(
              z.describe(
                'Order side as a boolean: `true` means the maker buys base, and `false` means the maker sells base.',
              ),
              z.meta({ examples: [true] }),
            ),
          isFlipOrder: z
            .boolean()
            .check(
              z.describe(
                'Whether this order automatically becomes a counter-order after it fills.',
              ),
              z.meta({ examples: [false] }),
            ),
          maker: Schema.Address.check(z.describe('Maker address that owns the order.')),
          mode: z
            .enum(['exactSource', 'exactDestination'])
            .check(
              z.describe(
                'Taker-perspective mode for the order. `exactSource` applies to maker bids where the taker sells base; `exactDestination` applies to maker asks where the taker buys base.',
              ),
              z.meta({ examples: ['exactSource'] }),
            ),
          orderId: Schema.DecimalString.check(
            z.describe('On-chain order id, returned as a decimal string.'),
          ),
          pair: order.Pair,
          price: z
            .string()
            .check(
              z.regex(/^\d+(\.\d+)?$/),
              z.describe(
                'Quote-per-base price implied by `tick`, returned as a fixed-decimal string with 5 decimal places.',
              ),
              z.meta({ examples: ['1.00000'] }),
            ),
          rate: z
            .string()
            .check(
              z.regex(/^\d+(\.\d+)?$/),
              z.describe(
                'Destination-to-source price ratio, returned as a fixed-decimal string with 5 decimal places.',
              ),
              z.meta({ examples: ['1.00000'] }),
            ),
          remaining: Schema.DecimalString.check(
            z.describe(
              'Unfilled base-side amount still resting on the orderbook, returned as a decimal integer string.',
            ),
          ),
          tick: z
            .number()
            .check(
              z.int(),
              z.describe(
                'On-chain signed tick, scaled by `1/priceScale`; use `price` for the decoded ratio.',
              ),
              z.meta({ examples: [0] }),
            ),
        }),
        'A single DEX order with live onchain state.',
      ),
      'ExchangeOrder',
    )
  }

  /** Schemas for the getOrders operation (the resting-order feed, optionally pair-scoped). */
  export namespace getOrders {
    /**
     * Optional resources for the order feed: the per-row `tokens` embed plus
     * the response-wide capped `totalCount` (the number of matching resting
     * orders, a lower bound when `truncated`).
     */
    export const Include = z
      .enum(['tokens', 'totalCount'])
      .check(
        z.describe(
          'Extra resources you can request with `include`, such as token metadata or a total count.',
        ),
      )

    /**
     * Parses a comma-separated `include` query value into a list of optional
     * resources to embed.
     */
    export const includeQuery = Schema.includeQuery(
      Include,
      'Comma-separated optional resources to embed, e.g. `tokens,totalCount`.',
    )

    /** Query parameters for resting-order feed requests. */
    export const Query = z
      .strictObject({
        base: z
          // The shared `TokenAddress` example is pathUSD — the universal
          // *quote* — which is never a valid base; use a real base instead.
          .optional(Schema.tokenAddress(exampleBase))
          .check(
            z.describe(
              'Only include orders for the pair with this base token. The quote token is determined onchain from the base token, so `base` alone selects the pair; returns 404 if no pair exists.',
            ),
          ),
        chainId: Schema.ChainIdQuery,
        cursor: Schema.Cursor,
        include: includeQuery,
        limit: Schema.Limit,
        maker: z
          .optional(Schema.Address)
          .check(z.describe('Only include orders placed by this maker address.')),
        order: Schema.Order,
        page: Schema.Page,
        side: z
          .optional(z.enum(['bid', 'ask']))
          .check(
            z.describe(
              'Limit results to one side of the orderbook: `bid` orders pay quote for base, while `ask` orders sell base for quote.',
            ),
            z.meta({ examples: ['bid'] }),
          ),
        sort: z
          ._default(z.enum(['tick', 'time']), 'tick')
          .check(
            z.describe(
              'Sort key: `tick` orders by price in standard book order, while `time` orders by block number and log index.',
            ),
            z.meta({ examples: ['tick'] }),
          ),
      })
      .check(z.describe('Filters, pagination, and sorting options for listing resting orders.'))

    /**
     * A single resting maker order, sans its pair (rows are shaped here and
     * then annotated with their pair). `amount` is the originally-placed base
     * amount and `remaining` is what's still on the book (`amount` minus the
     * sum of `OrderFilled.amountFilled` over all fills for this order).
     * Orders for which `remaining <= 0` or that have an `OrderCancelled`
     * event are excluded from the response.
     */
    export const RestingOrder = Schema.describe(
      z.object({
        amount: Schema.DecimalString.check(
          z.describe(
            'Original placed amount, returned as a decimal integer string in base-token smallest units.',
          ),
        ),
        blockNumber: z
          .number()
          .check(
            z.int(),
            z.nonnegative(),
            z.describe('Block number where the order was placed.'),
            z.meta({ examples: [23456789] }),
          ),
        id: Schema.DecimalString.check(
          z.describe('Stable API id for this order; it is the same value as the onchain order id.'),
        ),
        logIndex: z
          .number()
          .check(
            z.int(),
            z.nonnegative(),
            z.describe('Log index of the order placement within its block.'),
            z.meta({ examples: [0] }),
          ),
        maker: Schema.Address.check(z.describe('Maker address that owns the order.')),
        orderId: Schema.DecimalString.check(
          z.describe('On-chain order id, returned as a decimal string.'),
        ),
        placedAt: z.iso
          .datetime()
          .check(
            z.describe('Block timestamp when the order was placed.'),
            z.meta({ examples: ['2024-01-01T00:00:00Z'] }),
          ),
        price: z
          .string()
          .check(
            z.regex(/^\d+(\.\d+)?$/),
            z.describe(
              'Quote-per-base limit price implied by `tick`, returned with 5 decimal places. This value is direction-independent; use `rate` for the taker-perspective ratio.',
            ),
            z.meta({ examples: ['1.00000'] }),
          ),
        rate: z
          .string()
          .check(
            z.regex(/^\d+(\.\d+)?$/),
            z.describe(
              'Order limit price as a destination/source ratio from the taker’s perspective, returned with 5 decimal places. Use `price` for orderbook displays.',
            ),
            z.meta({ examples: ['1.00000'] }),
          ),
        remaining: Schema.DecimalString.check(
          z.describe(
            'Unfilled remainder of `amount`, returned as a decimal integer string in base-token smallest units. This is always positive for rows in this response.',
          ),
        ),
        side: z
          .enum(['bid', 'ask'])
          .check(
            z.describe(
              'Orderbook side: `bid` pays quote for base, and `ask` sells base for quote.',
            ),
            z.meta({ examples: ['bid'] }),
          ),
        tick: z
          .number()
          .check(
            z.int(),
            z.describe('Order tick as a signed offset from peg; one tick equals `1/priceScale`.'),
            z.meta({ examples: [0] }),
          ),
        transactionHash: Schema.Hash.check(
          z.describe('Transaction hash for the transaction that placed the order.'),
        ),
      }),
      'One resting maker order on Tempo’s built-in stablecoin exchange.',
    )

    /** A resting maker order annotated with the trading pair it belongs to. */
    export const Order = OpenApi.component(
      Schema.describe(
        z.extend(RestingOrder, {
          pair: order.Pair,
        }),
        'A resting maker order plus its trading pair.',
      ),
      'ExchangeRestingOrder',
    )

    /** Response metadata requested through `include`, such as capped counts. */
    export const Meta = OpenApi.component(Schema.CountMeta, 'ExchangeOrderListMeta')

    /** Page of resting maker orders, each annotated with its trading pair. */
    export const Response = OpenApi.component(
      z
        .object({
          data: z.array(Order).check(z.describe('Resting orders returned on this page.')),
          meta: z
            .optional(Meta)
            .check(
              z.describe(
                'Response-level resources requested with `include`, such as `totalCount`.',
              ),
            ),
          nextCursor: Schema.NextCursor,
          truncated: z
            .boolean()
            .check(
              z.describe(
                '`true` when the `OrderPlaced` scan hit its hard cap, meaning the resting set was computed from only the most recent placements.',
              ),
              z.meta({ examples: [false] }),
            ),
        })
        .check(
          z.describe(
            'A paginated list of resting maker orders on Tempo’s built-in stablecoin exchange.',
          ),
        ),
      'ExchangeOrderList',
    )
  }

  /** Schemas for the getOrderFills operation. */
  export namespace getOrderFills {
    /** Path parameters: inherits the canonical `/orders/:orderId` shape. */
    export const Params = order.Params

    /** Query parameters for paginated order-fill requests. */
    export const Query = z
      .strictObject({
        chainId: Schema.ChainIdQuery,
        cursor: Schema.Cursor,
        include: Schema.totalCountInclude,
        limit: Schema.Limit,
        order: Schema.Order,
        page: Schema.Page,
      })
      .check(
        ...Schema.pageChecks(),
        z.describe('Pagination and sorting options for listing fills for one order.'),
      )

    /**
     * A single fill against the order. The order's pair, maker, side, and
     * rate are constant across all fills, so they live on the `Order`
     * resource rather than being repeated per row.
     */
    export const Fill = OpenApi.component(
      z
        .object({
          amountFilled: Schema.DecimalString.check(
            z.describe(
              'Base-side amount filled by this event, returned as a decimal integer string in base-token smallest units.',
            ),
          ),
          blockNumber: z
            .number()
            .check(
              z.int(),
              z.nonnegative(),
              z.describe('Block number where the fill was included.'),
              z.meta({ examples: [23456789] }),
            ),
          filledAt: z.iso
            .datetime()
            .check(
              z.describe('Block timestamp when the fill landed onchain.'),
              z.meta({ examples: ['2024-01-01T00:00:00Z'] }),
            ),
          id: z.string().check(
            z.describe('Stable API id built from the lowercase transaction hash and log index.'),
            z.meta({
              examples: ['0x515801d7f9a5ac705e793e85904c9c69b3f1694b465cc8fb6ba3f0298dc82665-0'],
            }),
          ),
          logIndex: z
            .number()
            .check(
              z.int(),
              z.nonnegative(),
              z.describe('Log index of this fill within its block.'),
              z.meta({ examples: [0] }),
            ),
          orderId: Schema.DecimalString.check(
            z.describe('On-chain order id, returned as a decimal string.'),
          ),
          partialFill: z
            .boolean()
            .check(
              z.describe('Whether this fill used only part of the maker order.'),
              z.meta({ examples: [false] }),
            ),
          taker: Schema.Address.check(
            z.describe('Taker address that submitted the incoming order.'),
          ),
          transactionHash: Schema.Hash.check(
            z.describe('Transaction hash for the transaction containing the fill.'),
          ),
        })
        .check(z.describe('One `OrderFilled` event against this order.')),
      'ExchangeOrderFill',
    )

    /** Response metadata requested through `include`, such as capped counts. */
    export const Meta = OpenApi.component(Schema.CountMeta, 'ExchangeOrderFillListMeta')

    /** Paginated page of fills for one order. */
    export const Response = OpenApi.component(
      z
        .object({
          data: z.array(Fill).check(z.describe('Fills returned on this page.')),
          meta: z
            .optional(Meta)
            .check(z.describe('Response-level resources requested with `include`.')),
          nextCursor: Schema.NextCursor,
        })
        .check(z.describe('A paginated list of fill events for one DEX order.')),
      'ExchangeOrderFillList',
    )
  }

  /** Schemas for the getPairDepth operation (orderbook depth aggregation). */
  export namespace getPairDepth {
    /**
     * Path parameter mirrors `getPair.Params`: the base address alone
     * identifies the pair (its quote is intrinsic to the base token on-chain).
     */
    export const Params = z
      .object({
        base: Schema.tokenAddress(exampleBase).check(
          z.describe('Base token address for the trading pair.'),
        ),
      })
      .check(z.describe('Path parameters for fetching orderbook depth for a pair.'))

    /**
     * Optional related resource that callers opt into via `include`. Currently
     * only `tokens` (embeds metadata onto `base`/`quote` instead of returning
     * just the addresses).
     */
    export const Include = z
      .enum(['tokens'])
      .check(z.describe('Related resource you can opt into with `include`.'))

    /**
     * Parses a comma-separated `include` query value into a list of optional
     * resources to embed.
     */
    export const includeQuery = Schema.includeQuery(
      Include,
      'Comma-separated optional resources to embed, e.g. `tokens`.',
    )

    /** Query parameters for pair depth requests. */
    export const Query = z
      .strictObject({
        chainId: Schema.ChainIdQuery,
        include: includeQuery,
        levels: z
          ._default(
            z.coerce.number().check(z.int(), z.gte(1), z.lte(maxDepthLevels)),
            defaultDepthLevels,
          )
          .check(
            z.describe(
              `Maximum number of non-empty price levels to return per side (1-${maxDepthLevels}).`,
            ),
            z.meta({ examples: [50] }),
          ),
      })
      .check(z.describe('Query options for fetching orderbook depth for a pair.'))

    /**
     * A single price-level entry in the orderbook depth. `size` is the raw
     * `totalLiquidity` reported by the DEX precompile for the level (resting
     * base-token amount); `cumulativeSize` is the running sum across the same
     * side, starting from the best (peg-adjacent) populated tick and walking
     * outward.
     */
    export const Level = OpenApi.component(
      z
        .object({
          cumulativeSize: Schema.DecimalString.check(
            z.describe(
              'Running sum of `size` from the best populated tick outward to this level, in base-token smallest units.',
            ),
          ),
          id: z.string().check(
            z.describe('Stable API id built from the pair key, orderbook side, and tick.'),
            z.meta({
              examples: [
                '0x515801d7f9a5ac705e793e85904c9c69b3f1694b465cc8fb6ba3f0298dc82665-bid-0',
              ],
            }),
          ),
          price: z
            .string()
            .check(
              z.regex(/^\d+(\.\d+)?$/),
              z.describe('Tick converted to a price ratio with 5 decimal places.'),
              z.meta({ examples: ['1.00000'] }),
            ),
          size: Schema.DecimalString.check(
            z.describe('Resting liquidity at this tick level, in base-token smallest units.'),
          ),
          tick: z
            .number()
            .check(
              z.int(),
              z.describe('Signed price tick (`int16`), always a multiple of the DEX tick spacing.'),
              z.meta({ examples: [0] }),
            ),
        })
        .check(z.describe('One price level in the orderbook depth response.')),
      'ExchangeDepthLevel',
    )

    /** Orderbook depth (cumulative size per tick) for a single trading pair. */
    export const Response = OpenApi.component(
      z
        .object({
          asks: z
            .array(Level)
            .check(
              z.describe(
                'Ask-side levels, where makers sell base and takers buy it, ordered from best lowest ask outward.',
              ),
            ),
          base: PairToken,
          bids: z
            .array(Level)
            .check(
              z.describe(
                'Bid-side levels, where makers buy base and takers sell it, ordered from best highest bid outward.',
              ),
            ),
          quote: PairToken,
        })
        .check(
          z.describe('Orderbook depth for one trading pair, with cumulative size at each tick.'),
        ),
      'ExchangePairDepth',
    )
  }
}

/**
 * Creates exchange handlers. The first endpoint is `GET /pairs`, which lists
 * trading pairs on the stablecoin DEX precompile by replaying its `PairCreated`
 * events through the indexer.
 *
 * Two sort orders are supported:
 * - `sort=created` (default): order by pair creation `(block_num, log_idx)`.
 * - `sort=liquidity`: rank by the DEX precompile's escrow balance of the pair's
 *   base token, a per-pair liquidity proxy. Each base address participates in
 *   exactly one pair (the on-chain DEX keys orderbooks by base, so the same
 *   base never appears in two pairs), which lets a single
 *   `token_balances_snapshot` row attribute cleanly to one pair. Pairs whose
 *   DEX-escrow balance is zero are excluded from the liquidity ranking.
 */
export function exchanges(options: exchanges.Options = {}) {
  const oracle = options.fx?.oracle ?? FxOracle.ecb()
  return new Hono<App.Environment>()
    .on(
      ['POST', 'QUERY'],
      '/v1/exchange/quotes',
      Auth.policy({
        apiKey: { scopes: ['data:read'] },
        inheritOverridesFrom: 'POST',
        mpp: true,
        public: true,
      }),
      OpenApi.validate('query', schema.createQuote.Query, {
        code: 'query_invalid',
        message: 'Invalid query parameters',
      }),
      OpenApi.validate('json', schema.createQuote.Request, {
        code: 'body_invalid',
        message: 'Invalid request body',
      }),
      OpenApi.describeRoute({
        description:
          'Attempts exchange providers in priority order and returns the first executable quote. This endpoint does not submit a transaction.',
        operationId: 'createExchangeQuote',
        responses: OpenApi.responses({
          errors: {
            400: {
              codes: [
                'body_invalid',
                'chain_id_invalid',
                'chain_id_unsupported',
                'query_invalid',
                'quote_amount_out_of_range',
              ],
            },
            404: {
              codes: ['quote_not_available'],
              description:
                'No configured provider can price an executable exchange for these terms.',
            },
            502: 'The Tempo RPC could not quote the swap or resolve token metadata.',
          },
          success: {
            description: 'A provider-neutral quote and its next execution action.',
            schema: schema.createQuote.Response,
          },
        }),
        summary: 'Create quote',
        tags: ['Exchange'],
      }),
      async (c) => {
        if (Auth.narrowAccess) return Auth.paidAccessError(c)
        if (OpenApi.narrowValidation)
          return OpenApi.validationError(c, {
            code: 'query_invalid',
            message: 'Invalid query parameters',
          })
        if (OpenApi.narrowValidation)
          return OpenApi.validationError(c, {
            code: 'body_invalid',
            message: 'Invalid request body',
          })

        const request = c.req.valid('json')
        const query = c.req.valid('query')
        const chainId = query.chainId ?? c.get('chainId')
        const client = c.get('getClient')(chainId)
        const amount = BigInt(request.amount)
        const slippageBps = BigInt(request.slippageBps)

        c.header('Cache-Control', 'no-store')
        try {
          // Value quoted amounts via their curated display currency; snapshot
          // or rate failures degrade the values to `null`, never the quote.
          const denomination = query['valuation.currency']
          const snapshot = denomination
            ? await VerifiedTokens.snapshot(c, chainId).catch(() => undefined)
            : undefined
          const rates = denomination
            ? await Valuation.ratesFor(c, {
                currencies: snapshot
                  ? [request.destinationToken, request.sourceToken].flatMap((address) => {
                      const held = snapshot.byAddress.get(address.toLowerCase())?.currency
                      return held === undefined ? [] : [held]
                    })
                  : [],
                denomination,
                oracle,
              })
            : undefined
          const valued = (
            baseUnits: bigint | string,
            address: string,
            token: { currency: string; decimals: number },
          ) => ({
            ...Value.tokenAmount({
              baseUnits,
              currency: token.currency,
              decimals: token.decimals,
            }),
            ...(denomination
              ? {
                  valuation: Valuation.valuationFor({
                    amount: BigInt(baseUnits),
                    denomination,
                    rates,
                    token: snapshot?.byAddress.get(address.toLowerCase()),
                  }),
                }
              : {}),
          })

          if (options.nativeDex === false || amount > maxSwapAmount)
            throw new ExchangeProvider.RouteUnavailableError()

          if (request.mode === 'exactSource') {
            const destinationAmount = await Actions.dex.getSellQuote(client, {
              amountIn: amount,
              tokenIn: request.sourceToken,
              tokenOut: request.destinationToken,
            })
            if (destinationAmount === 0n) throw new ExchangeProvider.RouteUnavailableError()

            const minimumAmount =
              (destinationAmount * (slippageDenominator - slippageBps) + slippageDenominator - 1n) /
              slippageDenominator
            const approve = Actions.token.approve.call(client, {
              amount,
              spender: Addresses.stablecoinDex,
              token: request.sourceToken,
            })
            const swap = Actions.dex.sell.call({
              amountIn: amount,
              minAmountOut: minimumAmount,
              tokenIn: request.sourceToken,
              tokenOut: request.destinationToken,
            })
            const calls = [
              { data: approve.data, to: approve.to, value: 0n },
              { data: swap.data, to: swap.to, value: 0n },
            ] as const
            const gasUnits = await client.estimateGas({
              ...(request.account ? { account: request.account } : {}),
              calls,
            })
            const gasPrice = await client.getGasPrice()
            const tokensByAddress = await Tokens.resolveTokens(c, {
              addresses: [request.destinationToken, request.sourceToken],
              chainId,
              include: query.include,
            })
            const destinationToken = tokensByAddress.get(request.destinationToken)
            const sourceToken = tokensByAddress.get(request.sourceToken)
            if (!destinationToken || !sourceToken)
              throw new Error('Unable to resolve exchange quote token metadata')

            return c.json(
              Response.validated(schema.createQuote.Response, {
                destinationAmount: valued(
                  destinationAmount,
                  request.destinationToken,
                  destinationToken,
                ),
                destinationAmountMin: valued(
                  minimumAmount,
                  request.destinationToken,
                  destinationToken,
                ),
                destinationToken: {
                  ...destinationToken,
                  address: request.destinationToken,
                },
                ...(rates ? { meta: { valuation: Valuation.pricing(rates, oracle) } } : {}),
                gasEstimate: gasCost({ gasPrice, gasUnits }),
                mode: request.mode,
                provider: 'nativeDex',
                sourceAmount: valued(request.amount, request.sourceToken, sourceToken),
                sourceToken: {
                  ...sourceToken,
                  address: request.sourceToken,
                },
                status: 'ready',
                transaction: {
                  calls: calls.map((call) => ({ ...call, value: '0x0' as const })),
                  chainId,
                },
              }),
              200,
            )
          }

          const sourceAmount = await Actions.dex.getBuyQuote(client, {
            amountOut: amount,
            tokenIn: request.sourceToken,
            tokenOut: request.destinationToken,
          })
          if (sourceAmount === 0n) throw new ExchangeProvider.RouteUnavailableError()

          const maximumAmount =
            (sourceAmount * (slippageDenominator + slippageBps) + slippageDenominator - 1n) /
            slippageDenominator
          if (maximumAmount > maxSwapAmount) throw new NativeDexAmountOutOfRangeError()

          const approve = Actions.token.approve.call(client, {
            amount: maximumAmount,
            spender: Addresses.stablecoinDex,
            token: request.sourceToken,
          })
          const swap = Actions.dex.buy.call({
            amountOut: amount,
            maxAmountIn: maximumAmount,
            tokenIn: request.sourceToken,
            tokenOut: request.destinationToken,
          })
          const calls = [
            { data: approve.data, to: approve.to, value: 0n },
            { data: swap.data, to: swap.to, value: 0n },
          ] as const
          const gasUnits = await client.estimateGas({
            ...(request.account ? { account: request.account } : {}),
            calls,
          })
          const gasPrice = await client.getGasPrice()
          const tokensByAddress = await Tokens.resolveTokens(c, {
            addresses: [request.destinationToken, request.sourceToken],
            chainId,
            include: query.include,
          })
          const destinationToken = tokensByAddress.get(request.destinationToken)
          const sourceToken = tokensByAddress.get(request.sourceToken)
          if (!destinationToken || !sourceToken)
            throw new Error('Unable to resolve exchange quote token metadata')

          return c.json(
            Response.validated(schema.createQuote.Response, {
              destinationAmount: valued(request.amount, request.destinationToken, destinationToken),
              destinationToken: {
                ...destinationToken,
                address: request.destinationToken,
              },
              ...(rates ? { meta: { valuation: Valuation.pricing(rates, oracle) } } : {}),
              gasEstimate: gasCost({ gasPrice, gasUnits }),
              mode: request.mode,
              provider: 'nativeDex',
              sourceAmount: valued(sourceAmount, request.sourceToken, sourceToken),
              sourceAmountMax: valued(maximumAmount, request.sourceToken, sourceToken),
              sourceToken: {
                ...sourceToken,
                address: request.sourceToken,
              },
              status: 'ready',
              transaction: {
                calls: calls.map((call) => ({ ...call, value: '0x0' as const })),
                chainId,
              },
            }),
            200,
          )
        } catch (cause) {
          if (cause instanceof Valuation.UnsupportedDenominationError)
            return Response.error(c, { code: 'query_invalid', message: cause.message, status: 400 })
          const amountOutOfRange =
            cause instanceof NativeDexAmountOutOfRangeError ||
            (request.mode === 'exactSource' && isQuoteAmountOutOfRange(cause))
          const providers = c.get('providers').filter(ExchangeProvider.is)
          if (amountOutOfRange && providers.length === 0)
            return Response.error(c, {
              code: 'quote_amount_out_of_range',
              message: 'Quote amount exceeds the native DEX limit',
              status: 400,
            })
          if (
            !(cause instanceof ExchangeProvider.RouteUnavailableError) &&
            !isQuoteUnavailable(cause) &&
            !amountOutOfRange
          )
            return Response.upstream(c, cause)

          if (!request.account) return quoteNotAvailable(c)

          for (const provider of providers)
            try {
              const quote = await provider.quote(
                {
                  account: request.account,
                  amount: request.amount,
                  chainId,
                  destinationToken: request.destinationToken,
                  mode: request.mode,
                  slippageBps: request.slippageBps,
                  sourceToken: request.sourceToken,
                },
                c.req.raw.signal,
              )
              if (quote.status === 'ready')
                await validateProviderCalls(client, {
                  account: request.account,
                  calls: quote.transaction.calls,
                  execution: providerExecution({ quote, request }),
                })
              const [gasPrice, gasUnits] = await Promise.all([
                client.getGasPrice(),
                quote.status === 'approvalRequired'
                  ? client.estimateGas({
                      account: request.account,
                      calls: quote.approval.calls.map((call) => ({
                        data: call.data,
                        to: call.to,
                        value: BigInt(call.value),
                      })),
                    })
                  : Promise.resolve(BigInt(quote.gasUnits)),
              ])
              return c.json(
                Response.validated(
                  schema.createQuote.Response,
                  await providerQuoteResponse(c, {
                    chainId,
                    gasPrice,
                    gasUnits,
                    include: query.include,
                    oracle,
                    query,
                    quote,
                    request,
                    provider,
                  }),
                ),
                200,
              )
            } catch (error) {
              if (error instanceof ExchangeProvider.RouteUnavailableError) continue
              const failure = ExchangeProvider.failure(error, {
                chainId,
                operation: 'quote',
                provider,
              })
              if (failure) c.set('providerFailure', failure)
              return Response.upstream(c, error)
            }

          return quoteNotAvailable(c)
        }
      },
    )
    .post(
      '/v1/exchange/quotes/execute',
      Auth.policy({ apiKey: { scopes: ['data:read'] }, mpp: true, public: true }),
      OpenApi.validate('query', schema.createQuote.Query, {
        code: 'query_invalid',
        message: 'Invalid query parameters',
      }),
      OpenApi.validate('json', schema.executeQuote.Request, {
        code: 'body_invalid',
        message: 'Invalid request body',
      }),
      OpenApi.describeRoute({
        description:
          'Finishes a provider quote after the wallet supplies the requested typed-data signature. This endpoint does not submit a transaction.',
        operationId: 'finalizeExchangeQuote',
        responses: OpenApi.responses({
          errors: {
            400: {
              codes: [
                'body_invalid',
                'chain_id_invalid',
                'chain_id_unsupported',
                'query_invalid',
                'swap_continuation_invalid',
                'swap_provider_invalid',
              ],
            },
            502: 'The selected provider could not build the swap right now.',
          },
          success: {
            description: 'The final unsigned calls needed to execute the swap.',
            schema: schema.executeQuote.Response,
          },
        }),
        summary: 'Finalize quote',
        tags: ['Exchange'],
      }),
      async (c) => {
        if (Auth.narrowAccess) return Auth.paidAccessError(c)
        if (OpenApi.narrowValidation)
          return OpenApi.validationError(c, {
            code: 'query_invalid',
            message: 'Invalid query parameters',
          })
        if (OpenApi.narrowValidation)
          return OpenApi.validationError(c, {
            code: 'body_invalid',
            message: 'Invalid request body',
          })

        const request = c.req.valid('json')
        const chainId = c.req.valid('query').chainId ?? c.get('chainId')
        const provider = c
          .get('providers')
          .filter(ExchangeProvider.is)
          .find((candidate) => candidate.id === request.provider && candidate.finalizeQuote)
        if (!provider?.finalizeQuote)
          return Response.error(c, {
            code: 'swap_provider_invalid',
            message: 'The selected swap provider is not configured',
            status: 400,
          })

        c.header('Cache-Control', 'no-store')
        try {
          const client = c.get('getClient')(chainId)
          const result = await provider.finalizeQuote(
            {
              account: request.account,
              chainId,
              continuation: request.continuation,
              signature: request.signature,
            },
            c.req.raw.signal,
          )
          await validateProviderCalls(client, {
            account: request.account,
            calls: result.calls,
            execution: result.execution,
          })
          return c.json(
            Response.validated(schema.executeQuote.Response, {
              provider: provider.id,
              transaction: { calls: result.calls, chainId },
            }),
            200,
          )
        } catch (cause) {
          if (cause instanceof ExchangeProvider.ProviderContinuationError)
            return Response.error(c, {
              code: 'swap_continuation_invalid',
              message: 'Invalid swap continuation',
              status: 400,
            })
          const failure = ExchangeProvider.failure(cause, {
            chainId,
            operation: 'finalize',
            provider,
          })
          if (failure) c.set('providerFailure', failure)
          return Response.upstream(c, cause)
        }
      },
    )
    .get(
      '/v1/exchange/swaps',
      Auth.policy({ apiKey: { scopes: ['data:read'] }, mpp: true, public: true }),
      OpenApi.validate('query', schema.getSwaps.Query, {
        code: 'query_invalid',
        message: 'Invalid query parameters',
      }),
      OpenApi.describeRoute({
        description:
          'List recent swaps on Tempo’s built-in stablecoin exchange. A swap trades one token for another and may contain multiple maker-order fills.',
        operationId: 'getSwaps',
        responses: OpenApi.responses({
          errors: {
            400: {
              codes: ['chain_id_invalid', 'chain_id_unsupported', 'query_invalid'],
            },
            502: 'The indexer or Tempo RPC could not serve the exchange data right now.',
          },
          success: {
            description: 'A page of swaps from Tempo’s built-in stablecoin exchange.',
            schema: schema.getSwaps.Response,
          },
        }),
        summary: 'List swaps',
        tags: ['Exchange'],
      }),
      Cache.response<App.Environment>({
        // Exchanges are a newest-first feed; refresh quickly. The `v3` cache
        // segment isolates this response shape from earlier swap pages.
        cacheControl: Cache.policies.feed,
        name: 'tempo-api:exchange:v3:swaps',
        key: (c) => Cache.urlKey(c, schema.getSwaps.Query),
      }),
      async (c) => {
        if (Auth.narrowAccess) return Auth.paidAccessError(c)
        if (OpenApi.narrowValidation)
          return OpenApi.validationError(c, {
            code: 'query_invalid',
            message: 'Invalid query parameters',
          })

        const options = c.req.valid('query')
        const chainId = options.chainId ?? c.get('chainId')

        // Cursor pages carry current valuations and curated verification data.
        if (options.cursor !== undefined) Cache.setPolicy(c, Cache.policies.metadata)

        try {
          const page = await getSwaps(c, {
            chainId,
            cursor: options.cursor,
            fromBlock: options['blockNumber.from'],
            fromTimestamp: options['timestamp.from'],
            limit: options.limit,
            maker: options.maker,
            order: options.order,
            participant: options.participant,
            taker: options.taker,
            toBlock: options['blockNumber.to'],
            toTimestamp: options['timestamp.to'],
            transactionHash: options.transactionHash,
          })

          // Value amounts via their curated display currency; snapshot or rate
          // failures degrade the values to `null`, never the feed.
          const denomination = options['valuation.currency']
          // Mode, verified-token, and RPC metadata reads depend only on the
          // assembled page, so overlap them before enriching its rows.
          const [, snapshot, tokensByAddress] = await Promise.all([
            resolveSwapModes(c, { chainId, swaps: page.data }),
            denomination !== undefined && page.data.length > 0
              ? VerifiedTokens.snapshot(c, chainId).catch(() => undefined)
              : undefined,
            Tokens.resolveTokens(c, {
              addresses: page.data.flatMap((swap) => swap.route),
              chainId,
              include: options.include,
            }),
          ])
          const rates = denomination
            ? await Valuation.ratesFor(c, {
                currencies: snapshot
                  ? page.data.flatMap((swap) =>
                      swap.route.flatMap((address) => {
                        const held = snapshot.byAddress.get(address.toLowerCase())?.currency
                        return held === undefined ? [] : [held]
                      }),
                    )
                  : [],
                denomination,
                oracle,
              })
            : undefined
          const enrichSide = (side: getSwaps.Side) => {
            const token = tokensByAddress.get(side.address)
            if (!token) throw new Error(`Token metadata unavailable for ${side.address}`)
            return {
              amount: {
                ...Value.tokenAmount({
                  baseUnits: side.amount,
                  currency: token.currency,
                  decimals: token.decimals,
                }),
                ...(denomination
                  ? {
                      valuation: Valuation.valuationFor({
                        amount: BigInt(side.amount),
                        denomination,
                        rates,
                        token: snapshot?.byAddress.get(side.address.toLowerCase()),
                      }),
                    }
                  : {}),
              },
              token: {
                address: side.address,
                currency: token.currency,
                decimals: token.decimals,
                ...(token.logoUri !== undefined ? { logoUri: token.logoUri } : {}),
                name: token.name,
                symbol: token.symbol,
                ...(token.verified !== undefined ? { verified: token.verified } : {}),
              },
            }
          }
          const data = page.data.map((swap) => {
            const destination = enrichSide(swap.destinationToken)
            const source = enrichSide(swap.sourceToken)
            return {
              ...swap,
              destinationAmount: destination.amount,
              destinationToken: destination.token,
              fills: swap.fills.map((fill) => {
                const destination_fill = enrichSide(fill.destinationToken)
                const source_fill = enrichSide(fill.sourceToken)
                return {
                  ...fill,
                  destinationAmount: destination_fill.amount,
                  destinationToken: destination_fill.token,
                  sourceAmount: source_fill.amount,
                  sourceToken: source_fill.token,
                }
              }),
              sourceAmount: source.amount,
              sourceToken: source.token,
            }
          })

          return c.json(
            Response.validated(schema.getSwaps.Response, {
              data,
              ...(rates ? { meta: { valuation: Valuation.pricing(rates, oracle) } } : {}),
              nextCursor: page.nextCursor,
            }),
            200,
          )
        } catch (cause) {
          if (cause instanceof Valuation.UnsupportedDenominationError)
            return Response.error(c, { code: 'query_invalid', message: cause.message, status: 400 })
          return Response.upstream(c, cause)
        }
      },
    )
    .get(
      '/v1/exchange/pairs',
      Auth.policy({ apiKey: { scopes: ['data:read'] }, mpp: true, public: true }),
      OpenApi.validate('query', schema.getPairs.Query, {
        code: 'query_invalid',
        message: 'Invalid query parameters',
      }),
      OpenApi.describeRoute({
        description: 'List trading pairs available on Tempo’s built-in stablecoin exchange.',
        operationId: 'getPairs',
        responses: OpenApi.responses({
          errors: {
            400: {
              codes: ['chain_id_invalid', 'chain_id_unsupported', 'query_invalid'],
            },
            502: 'The indexer could not serve the exchange data right now.',
          },
          success: {
            description: 'A page of exchange trading pairs.',
            schema: schema.getPairs.Response,
          },
        }),
        summary: 'List pairs',
        tags: ['Exchange'],
      }),
      Cache.response<App.Environment>({
        // Pair creation is rare and append-only; `metadata` matches token-listing
        // volatility and amortizes the indexer round-trip well.
        cacheControl: Cache.policies.metadata,
        name: 'tempo-api:exchange:v1',
        key: (c) => Cache.urlKey(c, schema.getPairs.Query),
      }),
      async (c) => {
        if (Auth.narrowAccess) return Auth.paidAccessError(c)
        if (OpenApi.narrowValidation)
          return OpenApi.validationError(c, {
            code: 'query_invalid',
            message: 'Invalid query parameters',
          })

        const options = c.req.valid('query')
        const chainId = options.chainId ?? c.get('chainId')

        // Cursor pages are anchored below the head and effectively immutable, so
        // cache them aggressively; the head page keeps the route's `metadata`
        // default.
        if (options.cursor !== undefined) Cache.setPolicy(c, Cache.policies.immutable)

        try {
          // `totalCount` is opt-in: the pair total is independent of `sort`
          // (both orderings page the same set), so count `dex_pairs` directly.
          // Best-effort: a count failure omits `meta` rather than failing.
          const countPromise = options.include.includes('totalCount')
            ? countPairs(c, { chainId }).catch(() => undefined)
            : undefined

          const page = await getPairs(c, {
            chainId,
            cursor: options.cursor,
            limit: options.limit,
            order: options.order,
            page: options.page,
            sort: options.sort,
          })

          // Token metadata is opt-in. When requested, resolve each unique pair
          // token's metadata concurrently (sharing one verified-token fetch) and
          // spread it into both the `base` and `quote` row fields. Resolution is
          // best-effort: a token whose metadata is unavailable keeps the row's
          // `base`/`quote` as just `address` rather than failing the whole page.
          let data: z.output<typeof schema.getPairs.Pair>[] = page.data.map((pair) => ({
            base: { address: pair.base },
            blockNumber: pair.blockNumber,
            id: pair.key,
            key: pair.key,
            ...(pair.liquidity !== undefined && { liquidity: pair.liquidity }),
            quote: { address: pair.quote },
            timestamp: pair.timestamp,
            transactionHash: pair.transactionHash,
          }))
          if (options.include.includes('tokens') && data.length > 0) {
            const tokensByAddress = await Tokens.resolveTokens(c, {
              addresses: data.flatMap((pair) => [pair.base.address, pair.quote.address]),
              chainId,
              include: tokenFields,
            })
            data = data.map((pair) => {
              const base = tokensByAddress.get(pair.base.address)
              const quote = tokensByAddress.get(pair.quote.address)
              return {
                ...pair,
                base: base ? { ...pair.base, ...base } : pair.base,
                quote: quote ? { ...pair.quote, ...quote } : pair.quote,
              }
            })
          }

          const meta = countPromise ? await countPromise : undefined
          return c.json(
            Response.validated(schema.getPairs.Response, {
              data,
              ...(meta ? { meta } : {}),
              nextCursor: page.nextCursor,
            }),
            200,
          )
        } catch (cause) {
          return Response.upstream(c, cause, { page: options.page })
        }
      },
    )
    .get(
      // Single trading pair detail. The base address alone identifies the
      // pair (its quote is intrinsic to the base token on-chain).
      '/v1/exchange/pairs/:base',
      Auth.policy({ apiKey: { scopes: ['data:read'] }, mpp: true, public: true }),
      OpenApi.validate('param', schema.getPair.Params, {
        code: 'pair_invalid',
        message: 'Invalid base token',
      }),
      OpenApi.validate('query', schema.getPair.Query, {
        code: 'query_invalid',
        message: 'Invalid query parameters',
      }),
      OpenApi.describeRoute({
        description:
          'Get one trading pair by its base token address. On Tempo, the quote token for a pair is determined from the base token onchain.',
        operationId: 'getPair',
        responses: OpenApi.responses({
          errors: {
            400: {
              codes: ['chain_id_invalid', 'chain_id_unsupported', 'pair_invalid', 'query_invalid'],
            },
            404: {
              description: 'No exchange pair was found for that base token.',
              codes: ['pair_not_found'],
            },
            502: 'The indexer could not serve the exchange data right now.',
          },
          success: {
            description: 'Details for one exchange trading pair.',
            schema: schema.getPair.Response,
          },
        }),
        summary: 'Get pair',
        tags: ['Exchange'],
      }),
      Cache.response<App.Environment>({
        // Pair creation context is immutable once created; `metadata` matches
        // the pairs listing.
        cacheControl: Cache.policies.metadata,
        name: 'tempo-api:exchange:v1:pair',
        key: (c) => Cache.urlKey(c, schema.getPair.Query),
      }),
      async (c) => {
        if (Auth.narrowAccess) return Auth.paidAccessError(c)
        if (OpenApi.narrowValidation)
          return OpenApi.validationError(c, {
            code: 'pair_invalid',
            message: 'Invalid request parameters',
          })
        if (OpenApi.narrowValidation)
          return OpenApi.validationError(c, {
            code: 'query_invalid',
            message: 'Invalid request parameters',
          })

        const { base } = c.req.valid('param')
        const options = c.req.valid('query')
        const chainId = options.chainId ?? c.get('chainId')

        try {
          const pairs = await getPairIndex(c, { chainId })
          const target = pairs.get(base)
          if (!target) return pairNotFoundError(c, { base, pairIndex: pairs })

          // Token metadata is opt-in: resolve both pair tokens once and spread
          // onto the `base`/`quote` PairToken slots.
          let baseToken: z.output<typeof schema.PairToken> = { address: target.base }
          let quoteToken: z.output<typeof schema.PairToken> = { address: target.quote }
          if (options.include.includes('tokens')) {
            const tokensByAddress = await Tokens.resolveTokens(c, {
              addresses: [target.base, target.quote],
              chainId,
              include: tokenFields,
            })
            const baseMeta = tokensByAddress.get(target.base)
            const quoteMeta = tokensByAddress.get(target.quote)
            if (baseMeta) baseToken = { ...baseToken, ...baseMeta }
            if (quoteMeta) quoteToken = { ...quoteToken, ...quoteMeta }
          }

          return c.json(
            Response.validated(schema.getPair.Response, {
              base: baseToken,
              blockNumber: target.blockNumber,
              id: target.key,
              key: target.key,
              quote: quoteToken,
              timestamp: target.timestamp,
              transactionHash: target.transactionHash,
            }),
            200,
          )
        } catch (cause) {
          return Response.upstream(c, cause)
        }
      },
    )
    .get(
      // Resting-order feed, optionally scoped to one pair via `base`.
      '/v1/exchange/orders',
      Auth.policy({ apiKey: { scopes: ['data:read'] }, mpp: true, public: true }),
      OpenApi.validate('query', schema.getOrders.Query, {
        code: 'query_invalid',
        message: 'Invalid query parameters',
      }),
      OpenApi.describeRoute({
        description:
          'List resting maker orders on Tempo’s built-in stablecoin exchange. These are open orders waiting in the orderbook.',
        operationId: 'getOrders',
        responses: OpenApi.responses({
          errors: {
            400: {
              codes: ['chain_id_invalid', 'chain_id_unsupported', 'query_invalid'],
            },
            404: {
              description: 'No exchange pair was found for the requested `base` filter.',
              codes: ['pair_not_found'],
            },
            502: 'The indexer could not serve the exchange data right now.',
          },
          success: {
            description: 'A page of resting maker orders.',
            schema: schema.getOrders.Response,
          },
        }),
        summary: 'List orders',
        tags: ['Exchange'],
      }),
      Cache.response<App.Environment>({
        // Order book changes second-to-second; refresh quickly. The handler
        // memoizes the underlying OrderPlaced scan internally for ~15s.
        cacheControl: Cache.policies.state,
        name: 'tempo-api:exchange:v1:all-orders',
        key: (c) => Cache.urlKey(c, schema.getOrders.Query),
      }),
      async (c) => {
        if (Auth.narrowAccess) return Auth.paidAccessError(c)
        if (OpenApi.narrowValidation)
          return OpenApi.validationError(c, {
            code: 'query_invalid',
            message: 'Invalid query parameters',
          })

        const options = c.req.valid('query')
        const chainId = options.chainId ?? c.get('chainId')

        try {
          // `base` selects the single pair it identifies — a pair's quote is
          // intrinsic to its base token on-chain, so the base address alone
          // is a complete selector.
          let pair: getPairsByBase.Pair | undefined
          if (options.base !== undefined) {
            const target = (await getPairsByBase(c, { bases: [options.base], chainId })).get(
              options.base,
            )
            // The 404's quote-token hint scans the full pair index; fetch it
            // only on this miss path so page requests never pay for it.
            if (!target)
              return pairNotFoundError(c, {
                base: options.base,
                pairIndex: await getPairIndex(c, { chainId }),
              })
            pair = target
          }

          const page = await getOrders(c, {
            chainId,
            cursor: options.cursor,
            limit: options.limit,
            maker: options.maker,
            order: options.order,
            page: options.page,
            pair,
            side: options.side,
            sort: options.sort,
          })

          // Token metadata is opt-in. Collect every pair-token address across
          // the page, resolve each once, then spread it onto whichever pair
          // side carries that address. Best-effort: a token whose metadata is
          // unavailable keeps just `address`.
          let data = page.data
          if (options.include.includes('tokens') && data.length > 0) {
            const tokensByAddress = await Tokens.resolveTokens(c, {
              addresses: data.flatMap((o) => [o.pair.base.address, o.pair.quote.address]),
              chainId,
              include: tokenFields,
            })
            data = data.map((o) => {
              const base = tokensByAddress.get(o.pair.base.address)
              const quote = tokensByAddress.get(o.pair.quote.address)
              return {
                ...o,
                pair: {
                  ...o.pair,
                  base: base ? { ...o.pair.base, ...base } : o.pair.base,
                  quote: quote ? { ...o.pair.quote, ...quote } : o.pair.quote,
                },
              }
            })
          }

          // `totalCount` is opt-in and computed in-memory from the resting-order
          // snapshot (no extra query): the matching-order total, capped when the
          // placement scan was truncated (a lower bound).
          const meta = options.include.includes('totalCount')
            ? {
                // `totalCountCapped` when the placement scan was truncated OR the matching
                // set itself reaches the cap — either way the total is a lower bound.
                totalCountCapped: page.truncated || page.total >= Schema.countCap,
                totalCount: Math.min(page.total, Schema.countCap),
              }
            : undefined
          return c.json(
            Response.validated(schema.getOrders.Response, {
              data,
              ...(meta ? { meta } : {}),
              nextCursor: page.nextCursor,
              truncated: page.truncated,
            }),
            200,
          )
        } catch (cause) {
          return Response.upstream(c, cause, { page: options.page })
        }
      },
    )
    .get(
      // Single-order detail. The path-param schema rejects non-decimal
      // `orderId` values, surfacing the canonical `order_invalid` error.
      '/v1/exchange/orders/:orderId',
      Auth.policy({ apiKey: { scopes: ['data:read'] }, mpp: true, public: true }),
      OpenApi.validate('param', schema.getOrder.Params, {
        code: 'order_invalid',
        message: 'Invalid order id',
      }),
      OpenApi.validate('query', schema.getOrder.Query, {
        code: 'query_invalid',
        message: 'Invalid query parameters',
      }),
      OpenApi.describeRoute({
        description:
          'Get one order by id, using live onchain state for fields such as the remaining amount.',
        operationId: 'getOrder',
        responses: OpenApi.responses({
          errors: {
            400: {
              codes: ['chain_id_invalid', 'chain_id_unsupported', 'order_invalid', 'query_invalid'],
            },
            404: {
              description: 'No exchange order was found for that id.',
              codes: ['order_not_found'],
            },
            502: 'The Tempo RPC node could not serve the exchange data right now.',
          },
          success: {
            description: 'Details for one exchange order.',
            schema: schema.getOrder.Response,
          },
        }),
        summary: 'Get order',
        tags: ['Exchange'],
      }),
      Cache.response<App.Environment>({
        // `dex.getOrder` returns live state (`remaining` decreases as fills
        // land), so cache as `state`. The handler upgrades to `immutable` when
        // `remaining === '0'`, since a fully-filled order is terminal — every
        // field this resource exposes is then immutable.
        cacheControl: Cache.policies.state,
        name: 'tempo-api:exchange:v1:order',
        key: (c) => Cache.urlKey(c, schema.getOrder.Query),
      }),
      async (c) => {
        if (Auth.narrowAccess) return Auth.paidAccessError(c)
        if (OpenApi.narrowValidation)
          return OpenApi.validationError(c, {
            code: 'order_invalid',
            message: 'Invalid request parameters',
          })
        if (OpenApi.narrowValidation)
          return OpenApi.validationError(c, {
            code: 'query_invalid',
            message: 'Invalid request parameters',
          })

        const { orderId } = c.req.valid('param')
        const query = c.req.valid('query')
        const chainId = query.chainId ?? c.get('chainId')

        try {
          const order = await getOrder(c, { chainId, orderId })
          if (!order)
            return Response.error(c, {
              code: 'order_not_found',
              message: 'Order not found',
              status: 404,
            })

          // Fully-filled orders are terminal — `remaining`/`amount`/everything
          // else is fixed forever, so they can be cached aggressively.
          if (order.remaining === '0') Cache.setPolicy(c, Cache.policies.immutable)

          // Pair-token metadata is opt-in. The pair only has two tokens so we
          // resolve both concurrently behind a single verified-tokens fetch and
          // one batched `IN (...)` `tokens_created_at` query, matching the
          // pair-swap include path. Resolution is best-effort: a token whose
          // metadata is unavailable keeps just `address`.
          let pair = order.pair
          if (query.include.includes('tokens')) {
            const uniqueTokens = [pair.base.address, pair.quote.address]
            const snapshot = await VerifiedTokens.snapshot(c, chainId)
            const resolved = await Timing.time(c, 'tokens', () =>
              Promise.all(
                uniqueTokens.map((token) =>
                  Tokens.resolveToken(c, {
                    address: token,
                    chainId,
                    snapshot,
                  })
                    .then((metadata) => [token, metadata] as const)
                    .catch(() => [token, undefined] as const),
                ),
              ),
            )
            const tokensByAddress = new Map(resolved)
            const base = tokensByAddress.get(pair.base.address)
            const quote = tokensByAddress.get(pair.quote.address)
            pair = {
              ...pair,
              base: base ? { ...pair.base, ...base } : pair.base,
              quote: quote ? { ...pair.quote, ...quote } : pair.quote,
            }
          }

          return c.json(Response.validated(schema.getOrder.Response, { ...order, pair }), 200)
        } catch (cause) {
          return Response.upstream(c, cause)
        }
      },
    )
    .get(
      // Paginated fill history for a single order.
      '/v1/exchange/orders/:orderId/fills',
      Auth.policy({ apiKey: { scopes: ['data:read'] }, mpp: true, public: true }),
      OpenApi.validate('param', schema.getOrderFills.Params, {
        code: 'order_invalid',
        message: 'Invalid order id',
      }),
      OpenApi.validate('query', schema.getOrderFills.Query, {
        code: 'query_invalid',
        message: 'Invalid query parameters',
      }),
      OpenApi.describeRoute({
        description:
          'List fills for one order. A fill records an execution against the maker order.',
        operationId: 'getOrderFills',
        responses: OpenApi.responses({
          errors: {
            400: {
              codes: ['chain_id_invalid', 'chain_id_unsupported', 'order_invalid', 'query_invalid'],
            },
            502: 'The indexer could not serve the exchange data right now.',
          },
          success: {
            description: 'A page of fills for one order.',
            schema: schema.getOrderFills.Response,
          },
        }),
        summary: 'List order fills',
        tags: ['Exchange'],
      }),
      Cache.response<App.Environment>({
        // Newest-first fill feed — refresh quickly so newly-landed fills appear
        // on the head page. Cursor pages anchor below the head and are
        // effectively immutable; upgraded below.
        cacheControl: Cache.policies.feed,
        name: 'tempo-api:exchange:v1:order:fills',
        key: (c) => Cache.urlKey(c, schema.getOrderFills.Query),
      }),
      async (c) => {
        if (Auth.narrowAccess) return Auth.paidAccessError(c)
        if (OpenApi.narrowValidation)
          return OpenApi.validationError(c, {
            code: 'order_invalid',
            message: 'Invalid request parameters',
          })
        if (OpenApi.narrowValidation)
          return OpenApi.validationError(c, {
            code: 'query_invalid',
            message: 'Invalid request parameters',
          })

        const { orderId } = c.req.valid('param')
        const options = c.req.valid('query')
        const chainId = options.chainId ?? c.get('chainId')

        // Cursor pages anchor below the head on a stable `(block_num, log_idx)`
        // tuple, so they are effectively immutable across requests; only the
        // head page tracks the live tail.
        if (options.cursor !== undefined) Cache.setPolicy(c, Cache.policies.immutable)

        try {
          // `totalCount` (the order's fill total) is opt-in and shares the
          // page's `orderId` filter; run it concurrently with the page.
          // Best-effort: a count failure omits `meta` rather than failing the page.
          const countPromise = options.include.includes('totalCount')
            ? countOrderFills(c, { chainId, orderId }).catch(() => undefined)
            : undefined

          const page = await getOrderFills(c, {
            chainId,
            cursor: options.cursor,
            limit: options.limit,
            order: options.order,
            page: options.page,
            orderId,
          })

          const meta = countPromise ? await countPromise : undefined

          return c.json(
            Response.validated(schema.getOrderFills.Response, {
              data: page.data,
              ...(meta ? { meta } : {}),
              nextCursor: page.nextCursor,
            }),
            200,
          )
        } catch (cause) {
          return Response.upstream(c, cause, { page: options.page })
        }
      },
    )
    .get(
      // OHLC (candlestick) aggregations for a single pair, addressed by its
      // base token.
      '/v1/exchange/pairs/:base/ohlc',
      Auth.policy({ apiKey: { scopes: ['data:read'] }, mpp: true, public: true }),
      OpenApi.validate('param', schema.getPairOhlc.Params, {
        code: 'pair_invalid',
        message: 'Invalid pair tokens',
      }),
      OpenApi.validate('query', schema.getPairOhlc.Query, {
        code: 'query_invalid',
        message: 'Invalid query parameters',
      }),
      OpenApi.describeRoute({
        description:
          'Get OHLC price candles for one trading pair. OHLC means open, high, low, and close over each time bucket for charting.',
        operationId: 'getPairOhlc',
        responses: OpenApi.responses({
          errors: {
            400: {
              codes: ['chain_id_invalid', 'chain_id_unsupported', 'pair_invalid', 'query_invalid'],
            },
            404: {
              description: 'No exchange pair was found for that base token.',
              codes: ['pair_not_found'],
            },
            502: 'The indexer could not serve the exchange data right now.',
          },
          success: {
            description: 'OHLC candle data for the pair.',
            schema: schema.getPairOhlc.Response,
          },
        }),
        summary: 'Get pair OHLC',
        tags: ['Exchange'],
      }),
      Cache.response<App.Environment>({
        // Short-interval candles change second-to-second so the latest bucket
        // is volatile; the per-route handler upgrades the policy for longer
        // intervals where staleness is cheaper.
        cacheControl: Cache.policies.feed,
        name: 'tempo-api:exchange:v1:ohlc',
        key: (c) => Cache.urlKey(c, schema.getPairOhlc.Query),
      }),
      async (c) => {
        if (Auth.narrowAccess) return Auth.paidAccessError(c)
        if (OpenApi.narrowValidation)
          return OpenApi.validationError(c, {
            code: 'pair_invalid',
            message: 'Invalid request parameters',
          })
        if (OpenApi.narrowValidation)
          return OpenApi.validationError(c, {
            code: 'query_invalid',
            message: 'Invalid request parameters',
          })

        const { base: baseAddress } = c.req.valid('param')
        const options = c.req.valid('query')
        const chainId = options.chainId ?? c.get('chainId')

        // Hour+ candles change slowly; upgrade them to the `state` tier so we
        // amortize the per-fill scan more aggressively. The 1m/5m/15m tiers
        // keep the route's `feed` default.
        if (options.interval === '1h' || options.interval === '4h' || options.interval === '1d')
          Cache.setPolicy(c, Cache.policies.state)

        try {
          const pairs = await getPairIndex(c, { chainId })
          const target = pairs.get(baseAddress)
          if (!target) return pairNotFoundError(c, { base: baseAddress, pairIndex: pairs })

          const result = await getPairOhlc(c, {
            chainId,
            interval: options.interval,
            pair: target,
            window: options.window,
          })

          // Token metadata is opt-in. Resolve once per pair and spread onto the
          // top-level `base`/`quote` PairToken slots (the inner bucket rows
          // carry no token references, so no per-row enrichment is needed).
          let base: z.output<typeof schema.PairToken> = { address: target.base }
          let quote: z.output<typeof schema.PairToken> = { address: target.quote }
          if (options.include.includes('tokens')) {
            const tokensByAddress = await Tokens.resolveTokens(c, {
              addresses: [target.base, target.quote],
              chainId,
              include: tokenFields,
            })
            const baseMeta = tokensByAddress.get(target.base)
            const quoteMeta = tokensByAddress.get(target.quote)
            if (baseMeta) base = { ...base, ...baseMeta }
            if (quoteMeta) quote = { ...quote, ...quoteMeta }
          }

          return c.json(
            Response.validated(schema.getPairOhlc.Response, {
              base,
              data: result.data,
              interval: options.interval,
              quote,
              truncated: result.truncated,
              window: options.window,
            }),
            200,
          )
        } catch (cause) {
          return Response.upstream(c, cause)
        }
      },
    )
    .get(
      // Orderbook depth (cumulative size per tick) for a single pair,
      // addressed by its base token.
      '/v1/exchange/pairs/:base/depth',
      Auth.policy({ apiKey: { scopes: ['data:read'] }, mpp: true, public: true }),
      OpenApi.validate('param', schema.getPairDepth.Params, {
        code: 'pair_invalid',
        message: 'Invalid pair tokens',
      }),
      OpenApi.validate('query', schema.getPairDepth.Query, {
        code: 'query_invalid',
        message: 'Invalid query parameters',
      }),
      OpenApi.describeRoute({
        description:
          'Get orderbook depth for one trading pair. Depth shows resting liquidity at price ticks on both sides of the book.',
        operationId: 'getPairDepth',
        responses: OpenApi.responses({
          errors: {
            400: {
              codes: ['chain_id_invalid', 'chain_id_unsupported', 'pair_invalid', 'query_invalid'],
            },
            404: {
              description: 'No exchange pair was found for that base token.',
              codes: ['pair_not_found'],
            },
            502: 'The Tempo RPC node could not serve the exchange data right now.',
          },
          success: {
            description: 'Orderbook depth data for the pair.',
            schema: schema.getPairDepth.Response,
          },
        }),
        summary: 'Get pair depth',
        tags: ['Exchange'],
      }),
      Cache.response<App.Environment>({
        // Depth shifts on every order placement, fill, or cancellation; `feed`
        // matches the order-of-magnitude staleness traders tolerate on a
        // depth chart while still amortizing concurrent dashboard loads.
        cacheControl: Cache.policies.feed,
        name: 'tempo-api:exchange:v1:depth',
        key: (c) => Cache.urlKey(c, schema.getPairDepth.Query),
      }),
      async (c) => {
        if (Auth.narrowAccess) return Auth.paidAccessError(c)
        if (OpenApi.narrowValidation)
          return OpenApi.validationError(c, {
            code: 'pair_invalid',
            message: 'Invalid request parameters',
          })
        if (OpenApi.narrowValidation)
          return OpenApi.validationError(c, {
            code: 'query_invalid',
            message: 'Invalid request parameters',
          })

        const { base: baseAddress } = c.req.valid('param')
        const options = c.req.valid('query')
        const chainId = options.chainId ?? c.get('chainId')

        try {
          // Pair definitions are immutable after creation, so cache successful
          // point lookups without making unknown pairs stale.
          const target = await Store.memoize(
            async () =>
              (await getPairsByBase(c, { bases: [baseAddress], chainId })).get(baseAddress),
            {
              key: `exchange:v1:${chainId}:pair-by-base:${baseAddress}:v1`,
              store: c.get('store'),
              ttl: Ttl.minutes(5),
            },
          )
          if (!target)
            return pairNotFoundError(c, {
              base: baseAddress,
              pairIndex: await getPairIndex(c, { chainId }),
            })

          // Token metadata is opt-in and resolves once per pair onto the top-level
          // `base`/`quote` slots. Run it alongside the independent depth RPC scan.
          const [{ asks, bids }, tokensByAddress] = await Promise.all([
            getPairDepth(c, {
              chainId,
              levels: options.levels,
              pair: target,
            }),
            options.include.includes('tokens')
              ? Tokens.resolveTokens(c, {
                  addresses: [target.base, target.quote],
                  chainId,
                  include: tokenFields,
                })
              : undefined,
          ])

          let base: z.output<typeof schema.PairToken> = { address: target.base }
          let quote: z.output<typeof schema.PairToken> = { address: target.quote }
          if (tokensByAddress) {
            const baseMeta = tokensByAddress.get(target.base)
            const quoteMeta = tokensByAddress.get(target.quote)
            if (baseMeta) base = { ...base, ...baseMeta }
            if (quoteMeta) quote = { ...quote, ...quoteMeta }
          }

          return c.json(
            Response.validated(schema.getPairDepth.Response, { asks, base, bids, quote }),
            200,
          )
        } catch (cause) {
          return Response.upstream(c, cause)
        }
      },
    )
}

export declare namespace exchanges {
  /** Options for the exchange handlers. */
  type Options = {
    /** FX configuration backing amount valuation. */
    fx?: Valuation.addresses.Fx | undefined
    /** Whether to try Tempo's native stablecoin DEX first. Enabled by default. */
    nativeDex?: boolean | undefined
  }
}

// Hard cap on the curated pair index. Pair creation is rare (one event per
// pair lifetime), so a few thousand bounds today's set comfortably; raise this
// if the index ever exhausts.
const pairIndexLimit = 5000

/**
 * Builds the 404 for a failed pair lookup. The most common mistake is passing
 * the *quote* side of the pair (usually pathUSD) as `base` — every pair quotes
 * in pathUSD directly or transitively, so its address is never a base. Detect
 * that case and point the caller at the right selector instead of returning a
 * bare "not found".
 */
function pairNotFoundError(
  c: Context<App.Environment>,
  options: { base: string; pairIndex: Map<string, getPairs.Row> },
) {
  const { base, pairIndex } = options
  const message = [...pairIndex.values()].some((pair) => pair.quote === base)
    ? 'Pair not found: the address is a quote token; address the pair by its base token (see `GET /exchange/pairs`)'
    : 'Pair not found'
  return Response.error(c, { code: 'pair_not_found', message, status: 404 })
}

async function getPairs(c: Context<App.Environment>, options: getPairs.Options) {
  if (options.sort === 'liquidity') return getPairsByLiquidity(c, options)
  return getPairsByCreation(c, options)
}

/**
 * Exact total count of trading pairs on the stablecoin DEX. Reads the
 * `dex_pairs` materialized table (one row per `PairCreated`) on ClickHouse,
 * pruning on the DEX `address` — the same engine/key the page uses. Because the
 * `address` predicate aligns with the table's sort key, ClickHouse counts via
 * the sparse primary index (no row scan), so the count is exact and cheap with
 * no cap (`totalCountCapped` is always `false`). The total is independent of the
 * page's `sort` (both orderings page the same pair set).
 */
function countPairs(
  c: Context<App.Environment>,
  options: { chainId: z.output<typeof Schema.ChainId> },
): Promise<z.output<typeof Schema.CountMeta>> {
  const { chainId } = options
  const store = c.get('store')
  const tidx = c.get('getTidx')(chainId)
  return Timing.time(c, 'pairs_count', () =>
    Store.memoize(
      async () => {
        const result = await tidx.fetch({
          chainId,
          engine: 'clickhouse',
          query:
            `SELECT count(*) AS total FROM dex_pairs WHERE address = '${stablecoinDex}'` as string,
        })
        const totalCount = Value.toNumber(result.rows[0]?.['total']) ?? 0
        return { totalCountCapped: false, totalCount }
      },
      { key: `exchange:v1:${chainId}:pairs-count`, store, ttl: Ttl.seconds(60) },
    ),
  )
}

async function getPairsByCreation(c: Context<App.Environment>, options: getPairs.Options) {
  const { chainId, limit, order } = options
  const direction = order === 'asc' ? 'ASC' : 'DESC'
  const store = c.get('store')
  const tidx = c.get('getTidx')(chainId)
  // Keyset pagination on `(block_num, log_idx)`: a pair is created exactly
  // once, so this tuple is a stable, unique sort key. A malformed cursor falls
  // back to the head page.
  const cursor = options.cursor ? Cursor.decode(options.cursor, ['int', 'int']) : undefined
  // Bounded positional lane (exclusive with `cursor` at the schema): page 1 is
  // the head page, so it shares the head's SQL and cache entries; deeper pages
  // translate to a SQL OFFSET of `(page - 1) * limit` rows.
  const offset =
    options.page !== undefined && options.page > 1 ? (options.page - 1) * limit : undefined

  return Timing.time(c, 'pairs', () =>
    Store.memoize(
      async () => {
        // Pre-decoded pair rows from the stablecoin DEX precompile. Read the
        // `dex_pairs` materialized table (one row per `PairCreated`, sorted on
        // `(block_num, log_idx)`) instead of decoding the `PairCreated` event
        // signature over `logs` per request, so this is a sort-key read that
        // matches the keyset directly. Filtering by `address` keeps unrelated
        // `PairCreated`-named events (if any) out of the page. Fetch one extra
        // row to detect `hasMore` without a separate count query. The inline
        // query is cast to `string` so TIDX treats it as a dynamic ClickHouse
        // query that needs no event signature for the `dex_pairs` table.
        const filters: string[] = [`address = '${stablecoinDex}'`]
        if (cursor !== undefined)
          filters.push(
            Cursor.keyset([
              { literal: Cursor.literal(cursor[0]!, 'int'), name: 'block_num', order },
              { literal: Cursor.literal(cursor[1]!, 'int'), name: 'log_idx', order },
            ]),
          )
        const where = `WHERE ${filters.join(' AND ')}`

        const result = await tidx.fetch({
          chainId,
          engine: 'clickhouse',
          query: `
            SELECT key, base, quote, tx_hash, block_num, log_idx, block_timestamp
            FROM dex_pairs
            ${where}
            ORDER BY block_num ${direction}, log_idx ${direction}
            LIMIT ${limit + 1}${offset !== undefined ? ` OFFSET ${offset}` : ''}
          ` as string,
        })

        // The next page anchors below the last fetched row's `(block, log_idx)`.
        const page = Cursor.paginate({
          rows: result.rows,
          limit,
          key: (row) => {
            const block = Value.toNumber(row['block_num'])
            const index = Value.toNumber(row['log_idx'])
            return block !== undefined && index !== undefined ? [block, index] : undefined
          },
        })

        const data: getPairs.Row[] = []
        for (const row of page.rows) {
          const parsed = parsePairRow(row)
          if (parsed) data.push(parsed)
        }

        return { data, nextCursor: page.nextCursor }
      },
      {
        key: `exchange:v1:${chainId}:pairs:created:${order}:${cursor ? `cursor:${cursor[0]}:${cursor[1]}` : offset !== undefined ? `page:${options.page}` : 'head'}:${limit}`,
        store,
        ttl: Ttl.seconds(60),
      },
    ),
  )
}

async function getPairsByLiquidity(c: Context<App.Environment>, options: getPairs.Options) {
  const { chainId, limit, order } = options
  const store = c.get('store')
  const tidx = c.get('getTidx')(chainId)
  // Keyset pagination on `(balance, token)`: balance ranks the page (mutable,
  // so this is best-effort across balance changes — the inherent limit of
  // ranking feeds), and `token` is a stable tiebreaker. A malformed cursor
  // falls back to the head page.
  const cursor = options.cursor ? Cursor.decode(options.cursor, ['uint', 'address']) : undefined
  // Bounded positional lane (exclusive with `cursor` at the schema): page 1 is
  // the head page, so it shares the head's SQL and cache entries; deeper pages
  // translate to a SQL OFFSET of `(page - 1) * limit` rows.
  const offset =
    options.page !== undefined && options.page > 1 ? (options.page - 1) * limit : undefined
  const balanceOrder = order === 'asc' ? 'ASC' : 'DESC'

  return Timing.time(c, 'pairs', async () =>
    Store.memoize(
      async () => {
        // One ranked read from `dex_pair_liquidity`, which joins each pair's
        // base to its DEX-escrow balance in ClickHouse — replacing the previous
        // 3× over-fetch of escrow balances + in-memory intersection with the
        // pair index. The cursor keyset keeps the previous `(balance, token)`
        // value semantics under the view's `(liquidity, base)` column names, so
        // pre-existing cursors stay valid. Cast the inline query to `string`
        // (dynamic ClickHouse view, no event signature).
        const keyset =
          cursor !== undefined
            ? ` WHERE ${Cursor.keyset([
                { literal: Cursor.literal(cursor[0]!, 'uint'), name: 'liquidity', order },
                { literal: Cursor.literal(cursor[1]!, 'address'), name: 'base', order: 'asc' },
              ])}`
            : ''
        const result = await tidx.fetch({
          chainId,
          engine: 'clickhouse',
          // `liquidity` is UInt256; SELECT it as a string so values > 2^53
          // survive JSON (the cursor encodes the exact value). Alias avoids
          // shadowing the source column, which TIDX rejects (422).
          query: `
            SELECT key, base, quote, block_num, block_timestamp, tx_hash,
                   toString(liquidity) AS liquidity_str
            FROM dex_pair_liquidity${keyset}
            ORDER BY liquidity ${balanceOrder}, base ASC
            LIMIT ${limit + 1}${offset !== undefined ? ` OFFSET ${offset}` : ''}
          ` as string,
        })

        const page = Cursor.paginate({
          rows: result.rows,
          limit,
          key: (row) => {
            const liquidity = Value.toIntegerString(row['liquidity_str'])
            const base = Schema.Address.safeParse(row['base'])
            return liquidity !== undefined && base.success ? [liquidity, base.data] : undefined
          },
        })

        const data: getPairs.Row[] = []
        for (const row of page.rows) {
          const pair = parsePairRow(row)
          const liquidity = Value.toIntegerString(row['liquidity_str'])
          if (pair && liquidity !== undefined) data.push({ ...pair, liquidity })
        }

        return { data, nextCursor: page.nextCursor }
      },
      {
        key: `exchange:v1:${chainId}:pairs:liquidity:v2:${order}:${cursor ? `cursor:${cursor[0]}:${cursor[1]}` : offset !== undefined ? `page:${options.page}` : 'head'}:${limit}`,
        store,
        ttl: Ttl.seconds(30),
      },
    ),
  )
}

/**
 * Loads the curated pair index: a map from each pair's base address to its
 * pair info (key, quote, creation context). The index is bounded by
 * `pairIndexLimit` and memoized for several minutes since pair creation is
 * rare and append-only. Reused across liquidity-ranking pages.
 */
async function getPairIndex(c: Context<App.Environment>, options: getPairIndex.Options) {
  const { chainId } = options
  const store = c.get('store')
  const tidx = c.get('getTidx')(chainId)

  return Timing.time(c, 'pair_index', () =>
    Store.memoize(
      async () => {
        // Read the pre-decoded `dex_pairs` materialized table instead of decoding
        // the `PairCreated` event signature over `logs` per request. The inline
        // query is cast to `string` so TIDX treats it as a dynamic ClickHouse
        // query that needs no event signature for the `dex_pairs` table.
        const result = await tidx.fetch({
          chainId,
          engine: 'clickhouse',
          query: `
          SELECT key, base, quote, tx_hash, block_num, log_idx, block_timestamp
          FROM dex_pairs
          WHERE address = '${stablecoinDex}'
          ORDER BY block_num DESC, log_idx DESC
          LIMIT ${pairIndexLimit}
        ` as string,
        })
        // Serialize as `[base, pair]` entries so the cache layer round-trips
        // them through JSON cleanly (a `Map` would not survive serialization-
        // backed stores).
        const entries: [string, getPairs.Row][] = []
        for (const row of result.rows) {
          const parsed = parsePairRow(row)
          if (parsed) entries.push([parsed.base, parsed])
        }
        return entries
      },
      {
        key: `exchange:v1:${chainId}:pair-index:v1`,
        store,
        ttl: Ttl.minutes(5),
      },
    ),
  ).then((entries) => new Map(entries))
}

declare namespace getPairIndex {
  type Options = {
    chainId: z.output<typeof Schema.ChainId>
  }
}

/**
 * Base-address chunk size for {@link getPairsByBase}. `tidx.fetch` transmits
 * queries as GET, so an unbounded `IN (...)` could overflow the request URI.
 */
const pairLookupBatchSize = 200

/**
 * Resolves the pairs for a set of book base addresses with one targeted
 * `dex_pairs` lookup, avoiding the full pair-index fetch (1MB+ on large
 * chains).
 */
async function getPairsByBase(c: Context<App.Environment>, options: getPairsByBase.Options) {
  const { bases, chainId } = options
  const pairs = new Map<string, getPairsByBase.Pair>()
  if (bases.length === 0) return pairs
  const tidx = c.get('getTidx')(chainId)

  return Timing.time(c, 'pair_lookup', async () => {
    for (let i = 0; i < bases.length; i += pairLookupBatchSize) {
      const list = bases
        .slice(i, i + pairLookupBatchSize)
        .map((base) => `'${base}'`)
        .join(', ')
      // `LIMIT 1 BY base` with the ascending sort picks each base's oldest
      // row, the row the pair index's last-entry-wins `Map` build selects.
      const result = await tidx.fetch({
        chainId,
        engine: 'clickhouse',
        query: `
          SELECT key, base, quote
          FROM dex_pairs
          WHERE address = '${stablecoinDex}' AND base IN (${list})
          ORDER BY block_num ASC, log_idx ASC
          LIMIT 1 BY base
        ` as string,
      })
      // Drop malformed rows, mirroring `parsePairRow`.
      for (const row of result.rows) {
        const key = Schema.Hash.safeParse(row['key'])
        const base = Schema.TokenAddress.safeParse(row['base'])
        const quote = Schema.TokenAddress.safeParse(row['quote'])
        if (key.success && base.success && quote.success)
          pairs.set(base.data, { base: base.data, key: key.data, quote: quote.data })
      }
    }
    return pairs
  })
}

declare namespace getPairsByBase {
  type Options = {
    /** Book base addresses to resolve (lowercase). */
    bases: readonly string[]
    chainId: z.output<typeof Schema.ChainId>
  }
  /** Pair fields consumed by the swap and order feeds. */
  type Pair = {
    base: z.output<typeof Schema.TokenAddress>
    key: z.output<typeof Schema.Hash>
    quote: z.output<typeof Schema.TokenAddress>
  }
}

/**
 * Resolves one pair by its book key with a targeted `dex_pairs` lookup,
 * covering books the bounded pair index ages out of its newest window.
 */
async function getPairByKey(c: Context<App.Environment>, options: getPairByKey.Options) {
  const { chainId, key } = options
  const tidx = c.get('getTidx')(chainId)

  return Timing.time(c, 'pair_lookup', async () => {
    // The ascending sort picks the key's oldest row, matching the pair
    // index's last-entry-wins `Map` build.
    const result = await tidx.fetch({
      chainId,
      engine: 'clickhouse',
      query: `
        SELECT key, base, quote
        FROM dex_pairs
        WHERE address = '${stablecoinDex}' AND key = '${key}'
        ORDER BY block_num ASC, log_idx ASC
        LIMIT 1
      ` as string,
    })
    const row = result.rows[0]
    if (!row) return undefined
    // Drop malformed rows, mirroring `parsePairRow`.
    const parsedKey = Schema.Hash.safeParse(row['key'])
    const base = Schema.TokenAddress.safeParse(row['base'])
    const quote = Schema.TokenAddress.safeParse(row['quote'])
    if (!parsedKey.success || !base.success || !quote.success) return undefined
    return { base: base.data, key: parsedKey.data, quote: quote.data }
  })
}

declare namespace getPairByKey {
  type Options = {
    chainId: z.output<typeof Schema.ChainId>
    /** Book key (lowercase). */
    key: string
  }
}

/**
 * Validates and shapes one `PairCreated` row into the internal pair shape.
 * Returns `undefined` for malformed rows so the caller can drop them without
 * failing the whole page.
 */
function parsePairRow(row: Record<string, unknown>): getPairs.Row | undefined {
  const key = Schema.Hash.safeParse(row['key'])
  const base = Schema.TokenAddress.safeParse(row['base'])
  const quote = Schema.TokenAddress.safeParse(row['quote'])
  const transactionHash = Schema.Hash.safeParse(row['tx_hash'])
  const blockNumber = Value.toNumber(row['block_num'])
  const timestamp = Value.toIsoDateTime(row['block_timestamp'])
  if (
    !key.success ||
    !base.success ||
    !quote.success ||
    !transactionHash.success ||
    blockNumber === undefined ||
    timestamp === undefined
  )
    return undefined
  return {
    base: base.data,
    blockNumber,
    key: key.data,
    quote: quote.data,
    timestamp,
    transactionHash: transactionHash.data,
  }
}

declare namespace getPairs {
  /** Options for the trading pair page query. */
  type Options = {
    chainId: z.output<typeof Schema.ChainId>
    /** Opaque keyset cursor anchoring the page; omit for the head page. */
    cursor?: string | undefined
    limit: number
    /** 1-indexed page number (positional pagination; exclusive with `cursor`). */
    page?: number | undefined
    order: 'asc' | 'desc'
    sort: 'created' | 'liquidity'
  }

  /** A decoded `PairCreated` row before per-surface shaping. */
  type Row = {
    base: z.output<typeof Schema.TokenAddress>
    blockNumber: number
    key: z.output<typeof Schema.Hash>
    /**
     * DEX-escrow balance of the base token as a decimal integer string.
     * Present only when the page was ranked by liquidity.
     */
    liquidity?: string | undefined
    quote: z.output<typeof Schema.TokenAddress>
    timestamp: string
    transactionHash: z.output<typeof Schema.Hash>
  }
}

/**
 * Resolves a page of logical taker swaps.
 *
 * A swap row aggregates the `OrderFilled` events of one
 * `(block, transaction, taker)` group into one or more leg chains (§ multi-hop
 * swaps route through intermediate books; batched transactions can carry
 * several swaps). Each candidate batch is a grouped ClickHouse query over
 * `dex_fills`; each fill's `(token, isBid, tick)` is then resolved
 * point-in-time from the raw `OrderPlaced`/`OrderFlipped` event stream
 * ({@link resolveOrderStates}), since the fill event itself carries none of
 * it. The swap-level `mode` (transaction calldata, {@link resolveSwapCalls})
 * is deferred to the final page so the candidate scan never pays it for rows
 * that don't make the page.
 *
 * Every filter is pushed into the ClickHouse query — row predicates (`taker`,
 * `transactionHash`, block/timestamp bounds) in `WHERE`, and the
 * complete-fill-set filters (`maker`, `participant`) as HAVING aggregates — so
 * `LIMIT` counts matching groups exactly and no row is dropped after
 * pagination. A batched transaction can still expand one group into several
 * swap rows, so the page is capped at row granularity: the cursor carries
 * `(block, group first_idx, row first logIndex)`, the anchor group is
 * re-fetched on the next page, and its already-returned rows are dropped
 * row-wise — `data` never exceeds `limit`, and a group split across a page
 * boundary resumes mid-group.
 *
 * Because a group expands into one *or more* rows (and the rare un-orientable
 * group into zero), the scan accumulates rows across SQL batches (advancing
 * an internal cursor by whole groups) until it has `limit + 1` (more remain)
 * or the feed is exhausted; `nextCursor` then anchors at the last *included*
 * row. This is a row-count top-up over the group→row fan-out, not a filter —
 * no candidate is discarded after the query runs.
 */
async function getSwaps(c: Context<App.Environment>, options: getSwaps.Options) {
  const {
    chainId,
    cursor: cursorToken,
    fromBlock,
    fromTimestamp,
    limit,
    maker,
    order,
    participant,
    taker,
    toBlock,
    toTimestamp,
    transactionHash,
  } = options
  const tidx = c.get('getTidx')(chainId)
  const direction = order === 'asc' ? 'ASC' : 'DESC'
  // Keyset cursor on `(block_num, group first_idx, row first log_idx)`: the
  // group key scopes the SQL re-fetch, the row key resumes inside a group
  // that was split across a page boundary. A malformed cursor falls back to
  // the head page.
  const cursor = cursorToken ? Cursor.decode(cursorToken, ['int', 'int', 'int']) : undefined

  // `dex_fills` block window bounding one grouped batch (see
  // `swapWindowBlocks`); the head variant derives its lower bound in SQL.
  type BlockWindow = { hi: number; lo: number } | 'indexedHead'

  // Row-level predicates (constant within a group) prune before aggregation.
  // These are batch-invariant; the cursor's per-batch block bound is added in
  // `withCursorSql` below so the internal scan loop can advance it.
  const baseFilters: string[] = [`f.address = '${stablecoinDex}'`]
  if (taker !== undefined) baseFilters.push(`f.taker = '${taker}'`)
  // `tx_hash` is bloom-indexed on `dex_fills`, so this prunes granules.
  if (transactionHash !== undefined) baseFilters.push(`f.tx_hash = '${transactionHash}'`)
  if (fromBlock !== undefined) baseFilters.push(`f.block_num >= ${fromBlock}`)
  if (toBlock !== undefined) baseFilters.push(`f.block_num <= ${toBlock}`)
  // Normalize to ClickHouse's `YYYY-MM-DD HH:MM:SS` form — the ISO-Z form
  // rejects against `DateTime64(3, 'UTC')` with a conversion error.
  if (fromTimestamp !== undefined)
    baseFilters.push(`f.block_timestamp >= '${formatClickHouseDateTime(fromTimestamp)}'`)
  if (toTimestamp !== undefined)
    baseFilters.push(`f.block_timestamp <= '${formatClickHouseDateTime(toTimestamp)}'`)

  // Group-level predicates: evaluated over each group's complete fill set, so
  // "any fill matches" semantics stay exact under SQL pagination. Like
  // `baseFilters`, these are batch-invariant; the cursor's group/row bound is
  // added per batch in `withCursorSql`.
  const baseGroupFilters: string[] = []
  if (maker !== undefined) baseGroupFilters.push(`max(f.maker = '${maker}') = 1`)
  if (participant !== undefined)
    baseGroupFilters.push(`max(f.taker = '${participant}' OR f.maker = '${participant}') = 1`)

  // Per-batch SQL: the internal scan loop advances `activeCursor` across SQL
  // batches, so the cursor's bounds (row-level block bound in WHERE, group/row
  // refinement in HAVING) are rebuilt for each batch rather than baked into the
  // base predicates.
  const withCursorSql = (activeCursor: typeof cursor) => {
    const where = [...baseFilters]
    const group = [...baseGroupFilters]
    if (activeCursor !== undefined) {
      // The cursor's block bound is row-level; its `(block, first_idx)` tuple
      // refinement needs the aggregate and lives in HAVING.
      where.push(
        `f.block_num ${order === 'asc' ? '>=' : '<='} ${Cursor.literal(activeCursor[0]!, 'int')}`,
      )
      // Inclusive at the anchor group: a batch may end mid-group (a batched
      // transaction expands one group into several rows), so the anchor group
      // is re-fetched and its already-returned rows are dropped row-wise after
      // assembly.
      const op = order === 'asc' ? '>' : '<'
      const anchorBlock = Cursor.literal(activeCursor[0]!, 'int')
      const anchorGroup = Cursor.literal(activeCursor[1]!, 'int')
      group.push(
        `((block_num ${op} ${anchorBlock}) OR (block_num = ${anchorBlock} AND first_idx ${op}= ${anchorGroup}))`,
      )
    }
    return { having: group.length > 0 ? `HAVING ${group.join(' AND ')}` : '', where }
  }

  return Timing.time(c, 'swaps', async () => {
    // One grouped query returns complete groups with their fills inline.
    // `toString(...)` forces `uint128`/`uint256` tuple members through
    // ClickHouse's JSON output as strings — raw numbers in tuples lose
    // precision past 2^53. `block_timestamp` is per-block, so `min(...)` is
    // the group timestamp.
    const fetchGroups = (activeCursor: typeof cursor, batchLimit: number, window: BlockWindow) => {
      const { having, where } = withCursorSql(activeCursor)
      const windowFilters =
        window === 'indexedHead'
          ? [
              `f.block_num >= greatest(toInt128((SELECT max(num) FROM blocks)) - ${swapWindowBlocks - 1}, toInt128(0))`,
            ]
          : [`f.block_num >= ${window.lo}`, `f.block_num <= ${window.hi}`]
      return tidx.fetch({
        chainId,
        engine: 'clickhouse',
        query: `
        SELECT f.block_num AS block_num, f.tx_hash AS tx_hash, f.taker AS taker,
               min(f.log_idx) AS first_idx, min(f.block_timestamp) AS ts,
               groupArray(tuple(f.log_idx, toString(f."orderId"), f.maker,
                                toString(f."amountFilled"), f."partialFill")) AS fills
        FROM dex_fills f
        WHERE ${[...where, ...windowFilters].join(' AND ')}
        GROUP BY f.block_num, f.tx_hash, f.taker
        ${having}
        ORDER BY block_num ${direction}, first_idx ${direction}
        LIMIT ${batchLimit}
      ` as string,
      })
    }

    type Candidate = { group: parseGroupRow.Group; swap: getSwaps.Swap }

    // A row's keyset position is `(block, group first_idx, its first fill
    // logIndex)`; rows at or before the active cursor's row anchor were already
    // returned (the anchor group is re-fetched whole) and are dropped here.
    // The drop is purely positional.
    const past = (activeCursor: typeof cursor, blockNumber: number, logIndex: number) => {
      if (activeCursor === undefined) return true
      // Decode validated both fields as `int`, so they are plain numbers.
      const anchorBlock = Number(activeCursor[0])
      const anchorIndex = Number(activeCursor[2])
      if (blockNumber !== anchorBlock)
        return order === 'asc' ? blockNumber > anchorBlock : blockNumber < anchorBlock
      return order === 'asc' ? logIndex > anchorIndex : logIndex < anchorIndex
    }

    // Advance the internal cursor past a whole fetched group so the next batch
    // re-fetches that group and `past()` drops all of its rows. The row anchor
    // is direction-aware: `asc` drops rows `<=` the group's max fill logIndex,
    // `desc` drops rows `>=` its min (`first_idx`). `group.fills` is sorted
    // ascending by `parseGroupRow`.
    const cursorAfterWholeGroup = (group: parseGroupRow.Group) =>
      [
        group.blockNumber,
        group.firstIndex,
        order === 'asc' ? group.fills.at(-1)!.logIndex : group.firstIndex,
      ] as const

    // Resolve one fetched batch of groups into assembled, ordered candidate
    // rows. Only the state + quote lookups are needed to assemble; `mode`
    // (calldata) is deferred to the final page.
    const resolveBatch = async (
      groups: parseGroupRow.Group[],
      activeCursor: typeof cursor,
    ): Promise<Candidate[]> => {
      // Point-in-time order state (token/isBid/tick) for this batch's fills —
      // required to orient fills and assemble each swap's route ends.
      const blocks = [...new Set(groups.map((group) => group.blockNumber))]
      const states = await Timing.time(c, 'swap_states', () =>
        resolveOrderStates(c, {
          chainId,
          maxBlock: Math.max(...blocks),
          minBlock: Math.min(...blocks),
          orderIds: [
            ...new Set(groups.flatMap((group) => group.fills.map((fill) => fill.orderId))),
          ],
        }),
      )

      // Each fill's book quote comes back with its order state when indexed;
      // otherwise read the base token's on-chain `quoteToken()` — some books
      // predate the indexer's `PairCreated` decode (genesis system books).
      const bases = [
        ...new Set([...states.values()].flatMap((events) => events.map((event) => event.token))),
      ]
      const quoteByBase = new Map<string, z.output<typeof Schema.TokenAddress>>()
      for (const events of states.values())
        for (const event of events)
          if (event.quoteToken !== undefined) quoteByBase.set(event.token, event.quoteToken)
      const fallback = await Timing.time(c, 'swap_quotes', () =>
        resolveBookQuotes(c, {
          bases: bases.filter((bookBase) => !quoteByBase.has(bookBase)),
          chainId,
        }),
      )
      for (const [bookBase, quote] of fallback) quoteByBase.set(bookBase, quote)

      // Assemble each group into swap rows — a batched transaction expands one
      // group into several rows — then drop already-returned rows of the
      // re-fetched anchor group (cursor resume); every surviving row is kept.
      const candidates: Candidate[] = []
      for (const group of groups)
        for (const swap of buildSwapRows(group, { order, quoteByBase, states })) {
          if (!past(activeCursor, group.blockNumber, swap.fills[0]!.logIndex)) continue
          // The `maker`/`participant` HAVING filters are group-level
          // (`(block, tx, taker)`), but a batched transaction expands one group
          // into several swap rows — only the rows whose own fills involve the
          // address satisfy the per-swap contract, so refine to row granularity
          // here. The accumulation loop tops the page back up to `limit`.
          if (maker !== undefined && !swap.fills.some((fill) => fill.maker === maker)) continue
          if (
            participant !== undefined &&
            swap.taker !== participant &&
            !swap.fills.some((fill) => fill.maker === participant)
          )
            continue
          candidates.push({ group, swap })
        }
      candidates.sort((a, b) => {
        const delta =
          a.group.blockNumber - b.group.blockNumber ||
          a.swap.fills[0]!.logIndex - b.swap.fills[0]!.logIndex
        return order === 'asc' ? delta : -delta
      })
      return candidates
    }

    // A selective row predicate (`taker`/`maker`/`participant`/`transactionHash`)
    // makes matching fills sparse across a wide block range. Find the next
    // populated block before each window so the sweep skips empty gaps.
    //
    // The discovery predicate must be a *superset* of the grouped page query's
    // matches. The page query applies `maker`/`participant` as existential
    // HAVING aggregates over each `(block, tx, taker)` group, so a single group
    // can satisfy `maker` and `participant` via *different* fills. A row-level
    // conjunction would wrongly require one fill to satisfy both; OR the
    // group-level witnesses together instead — every block with a matching group
    // has at least one witness fill, and a group's fills share its block.
    const selective =
      taker !== undefined ||
      maker !== undefined ||
      participant !== undefined ||
      transactionHash !== undefined
    const blockFilters = [...baseFilters]
    {
      const witnesses: string[] = []
      if (maker !== undefined) witnesses.push(`f.maker = '${maker}'`)
      if (participant !== undefined) {
        witnesses.push(`f.taker = '${participant}'`)
        witnesses.push(`f.maker = '${participant}'`)
      }
      if (witnesses.length > 0) blockFilters.push(`(${witnesses.join(' OR ')})`)
    }

    const cursorBlock = cursor !== undefined ? Number(cursor[0]) : undefined

    // Keep discovery bounded so TIDX response limits cannot truncate the scan
    // head. The predicate stays a superset of the grouped query.
    const findMatchingBlock = async (anchor: number | undefined) => {
      const filters = [...blockFilters]
      if (anchor !== undefined)
        filters.push(`f.block_num ${order === 'asc' ? '>=' : '<='} ${anchor}`)
      const result = await tidx.fetch({
        chainId,
        engine: 'clickhouse',
        query:
          `SELECT f.block_num AS block_num FROM dex_fills f WHERE ${filters.join(' AND ')} ORDER BY block_num ${direction} LIMIT 1` as string,
      })
      const row = result.rows[0]
      return row ? Value.toNumber(row['block_num']) : undefined
    }

    type ScanBounds = {
      firstMatchingBlock?: number | undefined
      firstWindow?: 'indexedHead' | undefined
      hi: number
      lo: number
    }
    const bounds: ScanBounds | undefined = await Timing.time(c, 'swap_bounds', async () => {
      if (selective) {
        const firstMatchingBlock = await findMatchingBlock(cursorBlock)
        return firstMatchingBlock !== undefined
          ? {
              firstMatchingBlock,
              hi: toBlock ?? Number.MAX_SAFE_INTEGER,
              lo: fromBlock ?? 0,
            }
          : undefined
      }
      // The newest unfiltered page can derive its 100k-block lower bound from
      // `blocks` inside the grouped query, avoiding a separate TIDX request.
      if (
        order === 'desc' &&
        cursorBlock === undefined &&
        fromBlock === undefined &&
        fromTimestamp === undefined &&
        toBlock === undefined &&
        toTimestamp === undefined
      )
        return { firstWindow: 'indexedHead', hi: Number.MAX_SAFE_INTEGER, lo: 0 }
      const result = await tidx.fetch({
        chainId,
        engine: 'clickhouse',
        query:
          `SELECT min(f.block_num) AS lo, max(f.block_num) AS hi FROM dex_fills f WHERE ${blockFilters.join(' AND ')}` as string,
      })
      const row = result.rows[0]
      const lo = row ? Value.toNumber(row['lo']) : undefined
      const hi = row ? Value.toNumber(row['hi']) : undefined
      return lo !== undefined && hi !== undefined ? { hi, lo } : undefined
    })

    const accumulated: Candidate[] = []
    if (bounds) {
      const lowerBound = Math.max(fromBlock ?? bounds.lo, bounds.lo)
      const upperBound = Math.min(toBlock ?? bounds.hi, bounds.hi)

      // Window sweep: every grouped batch is bounded to one block window, since
      // an unbounded `dex_fills` GROUP BY is rejected by TIDX (HTTP 422) once the
      // table is large. `desc` walks windows downward from the cursor/head block;
      // `asc` upward. Within a window the keyset cursor advances by whole groups
      // (a full batch may leave more groups in the same window); a short batch
      // exhausts the window and slides the sweep to the next disjoint range.
      const batchLimit = limit + 1
      let activeCursor = cursor

      // Selective scans jump to the next populated block through a bounded
      // lookup. Unfiltered scans retain the fixed-stride min/max sweep after
      // the newest page's indexed-head window.
      const windowFrom = async (
        anchor: number | undefined,
        knownMatchingBlock?: number,
      ): Promise<BlockWindow | undefined> => {
        if (order === 'desc') {
          const value = Math.min(upperBound, anchor ?? upperBound)
          const hi = selective ? (knownMatchingBlock ?? (await findMatchingBlock(value))) : value
          return hi !== undefined && hi >= lowerBound
            ? { hi, lo: Math.max(lowerBound, hi - swapWindowBlocks + 1) }
            : undefined
        }
        const value = Math.max(lowerBound, anchor ?? lowerBound)
        const lo = selective ? (knownMatchingBlock ?? (await findMatchingBlock(value))) : value
        return lo !== undefined && lo <= upperBound
          ? { hi: Math.min(upperBound, lo + swapWindowBlocks - 1), lo }
          : undefined
      }

      let window: BlockWindow | undefined =
        bounds.firstWindow ?? (await windowFrom(cursorBlock, bounds.firstMatchingBlock))

      // Slide to the next disjoint window away from the newest block and reset
      // the cursor: every row in the next window is already past the cursor
      // anchor in the requested order, so no anchor refinement is needed.
      const slideWindow = async (indexedHeadAnchor?: number) => {
        activeCursor = undefined
        if (window === 'indexedHead') {
          const matchingBlock = await findMatchingBlock(indexedHeadAnchor)
          window =
            matchingBlock === undefined ? undefined : await windowFrom(matchingBlock, matchingBlock)
          return
        }
        window =
          window === undefined
            ? undefined
            : await windowFrom(order === 'desc' ? window.lo - 1 : window.hi + 1)
      }

      while (accumulated.length <= limit && window !== undefined) {
        const result = await fetchGroups(activeCursor, batchLimit, window)
        const fetchedFull = result.rows.length >= batchLimit

        const groups: parseGroupRow.Group[] = []
        for (const row of result.rows) {
          const group = parseGroupRow(row)
          if (group) groups.push(group)
        }

        for (const candidate of await resolveBatch(groups, activeCursor)) {
          accumulated.push(candidate)
          if (accumulated.length > limit) break
        }
        if (accumulated.length > limit) break

        // No parsable groups: a full batch cannot advance, so stop rather than
        // risk skipping lower groups in the same window; a short batch just
        // exhausts the window.
        const tail = groups.at(-1)
        if (tail === undefined) {
          if (fetchedFull) break
          await slideWindow()
          continue
        }

        // A short batch exhausts this window; a full one may leave more groups
        // in it, so advance the keyset cursor and re-query the same window.
        if (!fetchedFull) {
          await slideWindow(order === 'desc' ? tail.blockNumber - 1 : tail.blockNumber + 1)
          continue
        }
        const next = cursorAfterWholeGroup(tail)
        // Progress guard: if the cursor cannot advance, slide instead of looping.
        if (
          activeCursor !== undefined &&
          Number(activeCursor[0]) === next[0] &&
          Number(activeCursor[1]) === next[1] &&
          Number(activeCursor[2]) === next[2]
        ) {
          await slideWindow()
          continue
        }
        activeCursor = [next[0], next[1], next[2]]
      }
    }

    // Cut to `limit`; the external cursor anchors at the last INCLUDED row so
    // the next call resumes exactly after it (the anchor group is re-fetched
    // and dropped row-wise). `> limit` means at least one more matching row
    // exists, so the page continues; otherwise the feed is exhausted.
    const pageRows = accumulated.slice(0, limit)
    const last = pageRows.at(-1)
    const nextCursor =
      accumulated.length > limit && last !== undefined
        ? Cursor.encode([
            last.group.blockNumber,
            last.group.firstIndex,
            last.swap.fills[0]!.logIndex,
          ])
        : null

    return { data: pageRows.map((candidate) => candidate.swap), nextCursor }
  })
}

declare namespace getSwaps {
  /** Swap side before RPC metadata and structured amount enrichment. */
  type Side = {
    address: z.output<typeof Schema.TokenAddress>
    amount: z.output<typeof Schema.DecimalString>
  }

  /** Maker fill before RPC metadata and structured amount enrichment. */
  type Fill = Omit<
    z.output<typeof schema.getSwaps.Fill>,
    'destinationAmount' | 'destinationToken' | 'sourceAmount' | 'sourceToken'
  > & {
    /** Destination token and raw base-unit amount. */
    destinationToken: Side
    /** Source token and raw base-unit amount. */
    sourceToken: Side
  }

  /** Logical swap before RPC metadata and structured amount enrichment. */
  type Swap = Omit<
    z.output<typeof schema.getSwaps.Swap>,
    'destinationAmount' | 'destinationToken' | 'fills' | 'sourceAmount' | 'sourceToken'
  > & {
    /** Destination token and raw base-unit amount. */
    destinationToken: Side
    /** Maker fills that compose this swap. */
    fills: Fill[]
    /** Source token and raw base-unit amount. */
    sourceToken: Side
  }

  /** Options for the swap page query. */
  type Options = {
    chainId: z.output<typeof Schema.ChainId>
    /** Opaque keyset cursor anchoring the page; omit for the head page. */
    cursor?: string | undefined
    fromBlock?: number | undefined
    fromTimestamp?: string | undefined
    limit: number
    /** Any-fill maker filter. */
    maker?: z.output<typeof Schema.Address> | undefined
    /** Sort direction over `(block_num, first fill log_idx)`. */
    order: 'asc' | 'desc'
    /** Either-side filter — the taker or any fill's maker. */
    participant?: z.output<typeof Schema.Address> | undefined
    taker?: z.output<typeof Schema.Address> | undefined
    toBlock?: number | undefined
    toTimestamp?: string | undefined
    /** Only swaps contained in this transaction. */
    transactionHash?: z.output<typeof Schema.Hash> | undefined
  }
}

/**
 * Validates and shapes one grouped page row — the `(block, tx, taker)` key
 * plus its `groupArray` fill tuples — into a typed group. Fills are deduped
 * by `log_idx` (ReplacingMergeTree can surface unmerged duplicate rows) and
 * sorted into execution order. Returns undefined for malformed rows so the
 * caller can drop them without failing the page.
 */
function parseGroupRow(row: Record<string, unknown>): parseGroupRow.Group | undefined {
  const blockNumber = Value.toNumber(row['block_num'])
  const firstIndex = Value.toNumber(row['first_idx'])
  const taker = Schema.Address.safeParse(row['taker'])
  const timestamp = Value.toIsoDateTime(row['ts'])
  const transactionHash = Schema.Hash.safeParse(row['tx_hash'])
  const tuples = row['fills']
  if (
    blockNumber === undefined ||
    firstIndex === undefined ||
    !taker.success ||
    timestamp === undefined ||
    !transactionHash.success ||
    !Array.isArray(tuples)
  )
    return undefined

  const fills = new Map<number, parseGroupRow.Fill>()
  for (const tuple of tuples) {
    if (!Array.isArray(tuple)) continue
    const logIndex = Value.toNumber(tuple[0])
    const orderId = Value.toIntegerString(tuple[1])
    const maker = Schema.Address.safeParse(tuple[2])
    const amountFilled = Value.toIntegerString(tuple[3])
    if (
      logIndex === undefined ||
      orderId === undefined ||
      !maker.success ||
      amountFilled === undefined
    )
      continue
    fills.set(logIndex, {
      amountFilled,
      logIndex,
      maker: maker.data,
      orderId,
      // ClickHouse surfaces booleans as `0`/`1`.
      partialFill: tuple[4] === true || tuple[4] === 1,
    })
  }
  if (fills.size === 0) return undefined

  return {
    blockNumber,
    fills: [...fills.values()].sort((a, b) => a.logIndex - b.logIndex),
    firstIndex,
    taker: taker.data,
    timestamp,
    transactionHash: transactionHash.data,
  }
}

declare namespace parseGroupRow {
  /** One raw `OrderFilled` fill within a group, before order-state resolution. */
  type Fill = {
    /** Base-token amount filled, as a decimal integer string. */
    amountFilled: string
    logIndex: number
    maker: z.output<typeof Schema.Address>
    /** On-chain order identifier (decimal string). */
    orderId: string
    partialFill: boolean
  }

  /** One `(block, transaction, taker)` group of fills. */
  type Group = {
    blockNumber: number
    /** The group's fills, deduped and sorted by `logIndex`. */
    fills: Fill[]
    /** Lowest fill `logIndex` — the group's keyset cursor position. */
    firstIndex: number
    taker: z.output<typeof Schema.Address>
    /** Block timestamp (ISO). */
    timestamp: string
    transactionHash: z.output<typeof Schema.Hash>
  }
}

/**
 * Resolves the point-in-time order-state streams for a set of `orderId`s
 * from the raw `OrderPlaced`/`OrderFlipped` logs.
 *
 * A T5+ flip order keeps its `orderId` while `(isBid, tick)` swap sides on
 * every flip, so a fill's state is the latest event *before* the fill — not
 * the original `OrderPlaced`. One query returns, per order, its pre-page
 * history compressed to the latest event (`argMax`, sentinel position
 * `(-1, -1)` so it sorts first) plus every raw in-page event, so an order
 * that flips between two of its page fills resolves each fill correctly.
 * The same query attaches each book's indexed quote token, avoiding a second
 * TIDX request; callers retain an RPC fallback for books absent from the index.
 * `logs` is sorted by `(address, selector, block_num, log_idx)` with a
 * `topic1` bloom index, so both halves are pruned range reads. Returned
 * per-order event lists are position-sorted.
 */
export async function resolveOrderStates(
  c: Context<App.Environment>,
  options: resolveOrderStates.Options,
): Promise<Map<string, resolveOrderStates.State[]>> {
  const { chainId, maxBlock, minBlock, orderIds } = options
  const map = new Map<string, resolveOrderStates.State[]>()
  if (orderIds.length === 0) return map
  const tidx = c.get('getTidx')(chainId)

  const topics = orderIds.map((id) => `'0x${BigInt(id).toString(16).padStart(64, '0')}'`)
  const eventFilter = `address = '${stablecoinDex}'
          AND selector IN ('${orderPlacedTopic}', '${orderFlippedTopic}')
          AND topic1 IN (${topics.join(', ')})`
  // Aliases must not shadow the source columns (`t3`, not `topic3`) —
  // ClickHouse rejects self-shadowing aliases in queries that also filter on
  // the shadowed column.
  const result = await tidx.fetch({
    chainId,
    engine: 'clickhouse',
    query: `
      WITH states AS (
        SELECT topic1, argMax(topic3, (block_num, log_idx)) AS t3,
               argMax(data, (block_num, log_idx)) AS payload,
               toInt64(-1) AS bn, toInt32(-1) AS li
        FROM logs
        WHERE ${eventFilter} AND block_num < ${minBlock}
        GROUP BY topic1
        UNION ALL
        SELECT topic1, topic3 AS t3, data AS payload, block_num AS bn, log_idx AS li
        FROM logs
        WHERE ${eventFilter} AND block_num >= ${minBlock} AND block_num <= ${maxBlock}
      )
      SELECT states.*, pairs.quote AS quote_token
      FROM states
      LEFT JOIN (
        -- Preserve the oldest-pair rule without sorting the complete table.
        SELECT base, argMin(quote, (block_num, log_idx)) AS quote
        FROM dex_pairs
        WHERE address = '${stablecoinDex}'
        GROUP BY base
      ) AS pairs ON pairs.base = concat('0x', lower(substring(states.t3, 27)))
    ` as string,
  })

  for (const row of result.rows) {
    const parsed = parseStateRow(row)
    if (!parsed) continue
    const events = map.get(parsed.orderId)
    if (events) events.push(parsed.state)
    else map.set(parsed.orderId, [parsed.state])
  }
  for (const events of map.values())
    events.sort((a, b) => a.blockNumber - b.blockNumber || a.logIndex - b.logIndex)
  return map
}

declare namespace resolveOrderStates {
  /** Options for the order-state stream query. */
  type Options = {
    chainId: z.output<typeof Schema.ChainId>
    /** Highest block of the page (inclusive in-span bound). */
    maxBlock: number
    /** Lowest block of the page (baseline/in-span split point). */
    minBlock: number
    /** Decimal order ids to resolve. */
    orderIds: readonly string[]
  }

  /** Order state effective from `(blockNumber, logIndex)` onward. */
  type State = {
    /** Event block, or `-1` for the compressed pre-page baseline. */
    blockNumber: number
    /** Maker side at this state: `true` = bid (taker sells base into it). */
    isBid: boolean
    /** Event log index, or `-1` for the baseline. */
    logIndex: number
    /** Indexed quote token for the state's book, when available. */
    quoteToken?: z.output<typeof Schema.TokenAddress> | undefined
    /** Signed price tick at this state. */
    tick: number
    /** Book base token (lowercase address). */
    token: string
  }
}

/**
 * Decodes one order-state row (raw log topics/data) into a positioned
 * `(token, isBid, tick)` state. `OrderPlaced` and `OrderFlipped` share the
 * layout of the words read here: `data` word 1 is `isBid`, word 2 is `tick`
 * (int16, sign-extended); the book base token is `topic3`.
 */
function parseStateRow(
  row: Record<string, unknown>,
): { orderId: string; state: resolveOrderStates.State } | undefined {
  const topic1 = Value.toText(row['topic1'])
  const topic3 = Value.toText(row['t3'])
  const payload = Value.toText(row['payload'])
  const blockNumber = Value.toNumber(row['bn'])
  const logIndex = Value.toNumber(row['li'])
  const quoteToken = Schema.TokenAddress.safeParse(row['quote_token'])
  if (
    topic1 === undefined ||
    !/^0x[0-9a-fA-F]{64}$/.test(topic1) ||
    topic3 === undefined ||
    !/^0x[0-9a-fA-F]{64}$/.test(topic3) ||
    payload === undefined ||
    payload.length < 2 + 64 * 3 ||
    blockNumber === undefined ||
    logIndex === undefined
  )
    return undefined

  const isBid = BigInt(`0x${payload.slice(66, 130)}`) !== 0n
  // int16 sign-extended across the 32-byte word: values ≥ 2^255 are negative.
  const tickWord = BigInt(`0x${payload.slice(130, 194)}`)
  const tick = Number(tickWord >= 2n ** 255n ? tickWord - 2n ** 256n : tickWord)
  return {
    orderId: BigInt(topic1).toString(),
    state: {
      blockNumber,
      isBid,
      logIndex,
      ...(quoteToken.success ? { quoteToken: quoteToken.data } : {}),
      tick,
      token: `0x${topic3.slice(26).toLowerCase()}`,
    },
  }
}

/**
 * Recovers the decoded swap calls of a page's transactions from calldata,
 * keyed by transaction hash — the source of the swap-level `mode`. Direct
 * calls decode from `txs.input`; AA (type `0x76`) transactions use the
 * indexer's pre-decoded `calls` JSON column (`[{to, value, input}]`), so no
 * envelope parsing is needed.
 *
 * The `block_num IN (...)` bound is required: `txs` is sorted by
 * `(block_num, idx)` with no hash index, so a bare `hash IN` lookup
 * full-scans the table and times out upstream.
 */
async function resolveSwapCalls(
  c: Context<App.Environment>,
  options: resolveSwapCalls.Options,
): Promise<Map<string, resolveSwapCalls.Call[]>> {
  const { blocks, chainId, hashes } = options
  const map = new Map<string, resolveSwapCalls.Call[]>()
  if (hashes.length === 0) return map
  const tidx = c.get('getTidx')(chainId)

  const result = await tidx.fetch({
    chainId,
    engine: 'clickhouse',
    query: `
      SELECT hash, "type", "to", input, calls
      FROM txs
      WHERE block_num IN (${blocks.join(', ')})
        AND hash IN (${hashes.map((hash) => `'${hash}'`).join(', ')})
    ` as string,
  })

  for (const row of result.rows) {
    const hash = Value.toText(row['hash'])
    if (hash === undefined) continue
    // AA transactions carry their batched calls in the pre-decoded JSON
    // column; every other type is a single top-level call.
    type InnerCall = { data?: unknown; input?: unknown; to?: unknown }
    const inner: InnerCall[] = (() => {
      if (Value.toNumber(row['type']) !== 0x76) return [{ input: row['input'], to: row['to'] }]
      try {
        const raw = row['calls']
        const parsed: unknown = typeof raw === 'string' ? JSON.parse(raw) : raw
        return Array.isArray(parsed) ? (parsed as InnerCall[]) : []
      } catch {
        return []
      }
    })()

    const calls: resolveSwapCalls.Call[] = []
    for (const call of inner) {
      const to = Value.toText(call.to)
      const data = Value.toText(call.input) ?? Value.toText(call.data)
      if (to?.toLowerCase() !== stablecoinDex || data === undefined) continue
      try {
        const decoded = decodeFunctionData({ abi: swapAbi, data: data as Hex.Hex })
        calls.push({
          mode: decoded.functionName === 'swapExactAmountIn' ? 'exactSource' : 'exactDestination',
          tokenIn: decoded.args[0].toLowerCase(),
          tokenOut: decoded.args[1].toLowerCase(),
        })
      } catch {
        // Not a swap call (place/cancel/withdraw/...); skip.
      }
    }
    if (calls.length > 0) map.set(hash, calls)
  }
  return map
}

declare namespace resolveSwapCalls {
  /** One decoded swap call: the pinned-amount mode plus its route ends. */
  type Call = {
    /** `exactSource` = `swapExactAmountIn`; `exactDestination` = `swapExactAmountOut`. */
    mode: 'exactSource' | 'exactDestination'
    /** Call `tokenIn` (lowercase). */
    tokenIn: string
    /** Call `tokenOut` (lowercase). */
    tokenOut: string
  }

  /** Options for the page txs lookup. */
  type Options = {
    /** The page's block numbers (prunes the `txs` sort-key scan). */
    blocks: readonly number[]
    chainId: z.output<typeof Schema.ChainId>
    /** The page's transaction hashes. */
    hashes: readonly string[]
  }
}

/**
 * Attaches the swap-level `mode` (exact-source vs exact-destination) to each
 * row in place, recovered from the decoded transaction calldata
 * ({@link resolveSwapCalls}). A row's mode is pinned when one or more matching
 * calls (`tokenIn = route start`, `tokenOut = route end`) all agree; otherwise
 * it stays `null` (a resting order that crossed on placement, undecodable
 * calldata, or a failed lookup). Mutates `swaps`; runs only on the final page,
 * so the candidate scan never pays this `txs` lookup for filtered-out rows.
 */
async function resolveSwapModes(
  c: Context<App.Environment>,
  options: resolveSwapModes.Options,
): Promise<void> {
  const { chainId, swaps } = options
  if (swaps.length === 0) return
  const calls = await Timing.time(c, 'swap_calls', () =>
    resolveSwapCalls(c, {
      blocks: [...new Set(swaps.map((swap) => swap.blockNumber))],
      chainId,
      hashes: [...new Set(swaps.map((swap) => swap.transactionHash))],
    }),
  ).catch(() => new Map<string, resolveSwapCalls.Call[]>())

  for (const swap of swaps) {
    const matched = (calls.get(swap.transactionHash) ?? []).filter(
      (call) =>
        call.tokenIn === swap.sourceToken.address &&
        call.tokenOut === swap.destinationToken.address,
    )
    swap.mode =
      matched.length > 0 && matched.every((call) => call.mode === matched[0]!.mode)
        ? matched[0]!.mode
        : null
  }
}

declare namespace resolveSwapModes {
  /** Options for the deferred swap-mode resolution. */
  type Options = {
    chainId: z.output<typeof Schema.ChainId>
    /** The final page's swap rows; `mode` is mutated in place. */
    swaps: getSwaps.Swap[]
  }
}

/**
 * Resolves the quote token of order books missing from the `dex_pairs` index
 * (books created without a `PairCreated` event, e.g. genesis system books)
 * by reading the base token's on-chain `quoteToken()` — the quote is
 * intrinsic and immutable for a TIP-20, so results memoize long. `undefined`
 * from the RPC means the default quote (pathUSD). Best-effort: a base whose
 * quote cannot be resolved is absent from the returned map, and its fills
 * are dropped downstream.
 */
export async function resolveBookQuotes(
  c: Context<App.Environment>,
  options: resolveBookQuotes.Options,
): Promise<Map<string, z.output<typeof Schema.TokenAddress>>> {
  const { bases, chainId } = options
  const getClient = c.get('getClient')
  const store = c.get('store')
  const map = new Map<string, z.output<typeof Schema.TokenAddress>>()
  if (bases.length === 0) return map

  await Promise.all(
    bases.map(async (base) => {
      try {
        const quote = await Store.memoize(
          async () => {
            const metadata = await getClient(chainId).token.getMetadata({
              token: base as Address,
            })
            return metadata.quoteToken?.toLowerCase() ?? Addresses.pathUsd
          },
          { key: `exchange:v1:${chainId}:book-quote:${base}`, store, ttl: Ttl.minutes(60) },
        )
        const parsed = Schema.TokenAddress.safeParse(quote)
        if (parsed.success) map.set(base, parsed.data)
      } catch {
        // Best-effort: unresolvable books drop their fills downstream.
      }
    }),
  )
  return map
}

declare namespace resolveBookQuotes {
  /** Options for the book-quote fallback resolution. */
  type Options = {
    /** Book base addresses missing from the pair index (lowercase). */
    bases: readonly string[]
    chainId: z.output<typeof Schema.ChainId>
  }
}

/**
 * Assembles one group's fills into public swap rows.
 *
 * Each fill is oriented from the taker's perspective using its point-in-time
 * order state (maker bid → the taker sold the book base; maker ask → bought
 * it; the quote side is reconstructed from the tick: `1 base =
 * (priceScale + tick) / priceScale quote`). Oriented fills bucket into legs
 * (one book + direction), and legs chain source → destination into rows:
 *
 * - one (possibly multi-hop) swap → one chain → one row;
 * - unrelated swaps batched in one transaction → separate rows;
 * - a token-wise chaining batch is indistinguishable from one multi-hop swap
 *   and merges (documented trade-off);
 * - cyclic batches that fit no linear chain degrade to one row per leg.
 *
 * Fills with no resolvable state (historical `orderId = 0` ghost fills) or an
 * unknown book are dropped; a group may therefore produce no rows.
 */
function buildSwapRows(
  group: parseGroupRow.Group,
  options: buildSwapRows.Options,
): getSwaps.Swap[] {
  const { order, quoteByBase, states } = options
  const priceScale = BigInt(Tick.priceScale)

  type Oriented = {
    destination: { address: z.output<typeof Schema.TokenAddress>; amount: bigint }
    fill: getSwaps.Fill
    source: { address: z.output<typeof Schema.TokenAddress>; amount: bigint }
  }
  const oriented: Oriented[] = []
  for (const fill of group.fills) {
    // Latest state event strictly before the fill position. The flip a fill
    // itself triggers logs *after* the fill, so it never applies to its own
    // fill; the baseline (argMax) event carries position `(-1, -1)`.
    const events = states.get(fill.orderId)
    if (!events) continue
    let state: resolveOrderStates.State | undefined
    for (let i = events.length - 1; i >= 0; i--) {
      const event = events[i]!
      if (
        event.blockNumber < group.blockNumber ||
        (event.blockNumber === group.blockNumber && event.logIndex < fill.logIndex)
      ) {
        state = event
        break
      }
    }
    if (!state) continue
    const baseAddress = Schema.TokenAddress.safeParse(state.token)
    const quoteAddress = quoteByBase.get(state.token)
    if (!baseAddress.success || quoteAddress === undefined) continue

    // `amountFilled` is denominated in the book base; reconstruct the quote
    // side from the tick.
    const baseAmount = BigInt(fill.amountFilled)
    const quoteAmount = (baseAmount * (priceScale + BigInt(state.tick))) / priceScale
    const base = { address: baseAddress.data, amount: baseAmount }
    const quote = { address: quoteAddress, amount: quoteAmount }
    const source = state.isBid ? base : quote
    const destination = state.isBid ? quote : base
    oriented.push({
      destination,
      fill: {
        destinationToken: { address: destination.address, amount: destination.amount.toString() },
        logIndex: fill.logIndex,
        maker: fill.maker,
        orderId: fill.orderId,
        partialFill: fill.partialFill,
        price: Tick.toPrice(state.tick),
        sourceToken: { address: source.address, amount: source.amount.toString() },
      },
      source,
    })
  }
  if (oriented.length === 0) return []

  // Bucket fills into legs: one (source → destination) edge per book +
  // direction, amounts summed.
  const legs = new Map<string, buildSwapRows.Leg>()
  for (const item of oriented) {
    const key = `${item.source.address}>${item.destination.address}`
    const leg = legs.get(key)
    if (leg) {
      leg.destinationAmount += item.destination.amount
      leg.fills.push(item.fill)
      leg.sourceAmount += item.source.amount
    } else
      legs.set(key, {
        destination: item.destination.address,
        destinationAmount: item.destination.amount,
        fills: [item.fill],
        source: item.source.address,
        sourceAmount: item.source.amount,
      })
  }

  // Chain legs source → destination; each maximal chain is one swap row.
  // When no chain head exists (a cyclic batch), fall back to one row per leg.
  const remaining = new Set(legs.values())
  const chains: buildSwapRows.Leg[][] = []
  while (remaining.size > 0) {
    const pool = [...remaining]
    const head = pool.find(
      (leg) => !pool.some((other) => other !== leg && other.destination === leg.source),
    )
    if (!head) {
      for (const leg of pool) chains.push([leg])
      break
    }
    const chain = [head]
    remaining.delete(head)
    let current = head
    while (true) {
      const next = [...remaining].find((leg) => leg.source === current.destination)
      if (!next) break
      chain.push(next)
      remaining.delete(next)
      current = next
    }
    chains.push(chain)
  }

  const rows = chains.map((chain) => {
    const first = chain[0]!
    const last = chain.at(-1)!
    const fills = chain.flatMap((leg) => leg.fills).sort((a, b) => a.logIndex - b.logIndex)
    const logIndex = fills[0]!.logIndex
    // `mode` is recovered from transaction calldata and attached later
    // ({@link resolveSwapModes}), only for the rows that make the final page —
    // so the calldata lookup stays off the candidate scan's hot path.
    return {
      blockNumber: group.blockNumber,
      destinationToken: { address: last.destination, amount: last.destinationAmount.toString() },
      filledAt: group.timestamp,
      fills,
      id: `${group.transactionHash}-${logIndex}`,
      logIndex,
      mode: null as 'exactSource' | 'exactDestination' | null,
      rate: formatRate(last.destinationAmount, first.sourceAmount),
      route: [first.source, ...chain.map((leg) => leg.destination)],
      sourceToken: { address: first.source, amount: first.sourceAmount.toString() },
      taker: group.taker,
      transactionHash: group.transactionHash,
    }
  })

  // Order a multi-row group by each row's first fill, in page direction.
  const firstIndex = (row: (typeof rows)[number]) => row.fills[0]!.logIndex
  rows.sort((a, b) =>
    order === 'asc' ? firstIndex(a) - firstIndex(b) : firstIndex(b) - firstIndex(a),
  )
  return rows
}

declare namespace buildSwapRows {
  /** A directed swap leg: every fill on one book in one taker direction. */
  type Leg = {
    /** Token the taker received on this leg. */
    destination: z.output<typeof Schema.TokenAddress>
    /** Total received on this leg, in base units. */
    destinationAmount: bigint
    /** The leg's fills before metadata enrichment. */
    fills: getSwaps.Fill[]
    /** Token the taker sent on this leg. */
    source: z.output<typeof Schema.TokenAddress>
    /** Total sent on this leg, in base units. */
    sourceAmount: bigint
  }

  /** Per-group assembly inputs. */
  type Options = {
    /** Page sort direction (orders multi-row groups). */
    order: 'asc' | 'desc'
    /** Each book's quote token, keyed by its base address. */
    quoteByBase: Map<string, z.output<typeof Schema.TokenAddress>>
    /** Point-in-time order-state streams by orderId. */
    states: Map<string, resolveOrderStates.State[]>
  }
}

/**
 * Formats a destination/source amount ratio as a fixed-decimal string at the
 * DEX price scale (5 dp), via bigint math so large amounts don't lose float
 * precision.
 */
function formatRate(destination: bigint, source: bigint): string {
  const scale = BigInt(Tick.priceScale)
  const decimals = String(Tick.priceScale).length - 1
  if (source === 0n) return `0.${'0'.repeat(decimals)}`
  const scaled = (destination * scale) / source
  return `${scaled / scale}.${(scaled % scale).toString().padStart(decimals, '0')}`
}

/**
 * Formats the taker-perspective rate (destination/source) for a fill of an
 * order at `tick`. For maker bids that is the DEX price (quote per base); for
 * asks it is its inverse, formatted to the same 5 dp as the price scale.
 */
function ohlcRate(order: { isBid: boolean; tick: number }): string {
  if (order.isBid) return Tick.toPrice(order.tick)
  const priceScale = BigInt(Tick.priceScale)
  return (
    Number(priceScale * priceScale) /
    Number(priceScale + BigInt(order.tick)) /
    Number(priceScale)
  ).toFixed(5)
}

/**
 * Parses a `(tick, isBid)` tuple returned by a ClickHouse `argMin`/`argMax`
 * aggregate (JSON renders tuples as arrays; both members are small integers,
 * so no string forcing is needed).
 */
function parseTickSide(value: unknown): { isBid: boolean; tick: number } | undefined {
  if (!Array.isArray(value)) return undefined
  const tick = Value.toNumber(value[0])
  if (tick === undefined) return undefined
  return { isBid: value[1] === 1 || value[1] === true, tick }
}

/**
 * Resolves a candlestick (OHLC) aggregation for a single pair over a
 * rolling window.
 *
 * The candles are aggregated **in ClickHouse**: each bucket folds server-side
 * (open/close by fill position, high/low by executed rate, volume sums and
 * fill counts), so the response covers the entire requested window no matter
 * how many fills it contains — the previous implementation shipped the newest
 * raw fills to the app capped at 1,000, which truncated busy pairs' 7d/30d
 * windows to a few hours. The pair filter still resolves through the
 * OrderPlaced-only `dex_orders` table, so fills of flipped orders remain
 * invisible until order state lands at ingest (the `dex_ohlc_1m` plan).
 *
 * `OrderFilled` only carries the base-token `amountFilled`; the quote-side
 * amount is reconstructed per fill from the order's tick using the DEX's
 * fixed price scale (`quote = base * (priceScale + tick) / priceScale`),
 * which assumes base and quote share decimals — true for the stablecoin DEX.
 *
 * Buckets are anchored against epoch (`intDiv(ts, intervalMs)`), so a `1h`
 * bucket at `13:00–14:00` aligns to the wall clock regardless of when the
 * request was made. Empty buckets are dropped; charting UIs that need a
 * continuous axis can forward-fill from the previous close. `data` is
 * returned oldest → newest, which is the canonical chart input order.
 */
async function getPairOhlc(c: Context<App.Environment>, options: getPairOhlc.Options) {
  const { chainId, interval, pair, window } = options
  const tidx = c.get('getTidx')(chainId)
  // ClickHouse `DateTime64(3, 'UTC')` rejects ISO-Z; normalize to space form.
  const fromTimestamp = formatClickHouseDateTime(
    new Date(Date.now() - windowMs(window)).toISOString(),
  )
  const bucketMs = intervalMs(interval)

  return Timing.time(c, 'pair_ohlc', async () => {
    // Taker-perspective rate used only to *select* the high/low fill; the
    // formatted rate string is recomputed in JS from the winning `(tick,
    // isBid)` so it goes through the exact same formatting as open/close.
    const scale = Tick.priceScale
    const rate = `if(p."isBid" = 1, (${scale} + p.tick) / ${scale}, ${scale} / (${scale} + p.tick))`
    // UInt256 sums are forced through `toString` so values past 2^53 survive
    // JSON; the `(tick, isBid)` tuples are small integers and need no forcing.
    const result = await tidx.fetch({
      chainId,
      engine: 'clickhouse',
      query: `
        SELECT
          intDiv(toUnixTimestamp64Milli(f.block_timestamp), ${bucketMs}) AS bucket,
          count() AS fill_count,
          toString(sum(f."amountFilled")) AS base_volume,
          toString(sum(intDiv(f."amountFilled" * toUInt256(${scale} + p.tick), ${scale}))) AS quote_volume,
          argMin(tuple(p.tick, p."isBid"), tuple(f.block_num, f.log_idx)) AS open_t,
          argMax(tuple(p.tick, p."isBid"), tuple(f.block_num, f.log_idx)) AS close_t,
          argMax(tuple(p.tick, p."isBid"), ${rate}) AS high_t,
          argMin(tuple(p.tick, p."isBid"), ${rate}) AS low_t
        FROM dex_fills f
        JOIN dex_orders p ON p."orderId" = f."orderId"
        WHERE f.address = '${stablecoinDex}'
          AND f.block_timestamp >= '${fromTimestamp}'
          AND f."orderId" IN (
            SELECT p2."orderId" FROM dex_orders p2 WHERE p2.token = '${pair.base}'
          )
        GROUP BY bucket
        ORDER BY bucket ASC
      ` as string,
    })

    const data: z.output<typeof schema.getPairOhlc.Bucket>[] = []
    for (const row of result.rows) {
      const bucket = Value.toNumber(row['bucket'])
      const fillCount = Value.toNumber(row['fill_count'])
      const base = Value.toIntegerString(row['base_volume'])
      const quote = Value.toIntegerString(row['quote_volume'])
      const open = parseTickSide(row['open_t'])
      const close = parseTickSide(row['close_t'])
      const high = parseTickSide(row['high_t'])
      const low = parseTickSide(row['low_t'])
      if (
        bucket === undefined ||
        fillCount === undefined ||
        base === undefined ||
        quote === undefined ||
        !open ||
        !close ||
        !high ||
        !low
      )
        continue
      data.push({
        close: ohlcRate(close),
        fillCount,
        high: ohlcRate(high),
        id: `${pair.key}-${interval}-${bucket}`,
        low: ohlcRate(low),
        open: ohlcRate(open),
        timestamp: new Date(bucket * bucketMs).toISOString(),
        volume: { base, quote },
      })
    }

    // The full window is always aggregated now; the field stays for contract
    // stability until the next breaking version.
    return { data, truncated: false }
  })
}

declare namespace getPairOhlc {
  /** Options for the pair OHLC query. */
  type Options = {
    chainId: z.output<typeof Schema.ChainId>
    /** Bucket size. */
    interval: z.output<typeof schema.getPairOhlc.Interval>
    /** The resolved pair whose OHLC is being aggregated. */
    pair: getPairs.Row
    /** Rolling lookback. */
    window: z.output<typeof schema.getPairOhlc.Window>
  }
}

/**
 * Hard cap on the per-pair `OrderPlaced` scan that backs a resting-order
 * snapshot. Bounds the worst-case SQL `LIMIT` and the number of `orderId`s
 * batched into the cancellation/fill join. When the cap is hit the snapshot
 * reflects only the most-recent window of placements; the response carries
 * `truncated: true` so callers know older orders may be missing.
 */
const orderScanCap = 5000

/**
 * Scan cap for the global resting-order feed (all pairs). Lower than
 * {@link orderScanCap} on purpose: the global scan has no `token` predicate to
 * narrow it, so it both reads more `OrderPlaced` rows and fans out more
 * `OrderCancelled`/`OrderFilled` IN-batches than any per-pair scan. Each of
 * those queries is an independent transient-`db error` opportunity under CI
 * load, and a smaller window keeps the global request to a handful of round
 * trips. Deep pages still flip `truncated`.
 */
const globalOrderScanCap = 2000

/**
 * Batch size for `OrderCancelled` / `OrderFilled` `WHERE "orderId" IN (...)`
 * lookups. A 2000-ID batch uses indexed state paths while native fill
 * aggregation keeps the response below SQL and result limits.
 */
const orderInBatchSize = 2000

/**
 * Signed-int16 bias used to encode `tick` in the cursor as a non-negative
 * uint string. `Cursor.Field === 'int'` is non-negative-only, so signed ticks
 * (range `[-32768, 32767]`) are shifted into `[0, 65535]` before encoding.
 */
const orderCursorTickBias = 32768

/**
 * Resolves a page of resting maker orders, optionally scoped to one pair.
 *
 * Strategy: snapshot the resting set in one pass (memoized briefly to
 * amortize bursty UI polling), then sort, filter, and paginate the snapshot
 * in memory. The underlying scans target the indexer's ClickHouse OLAP path
 * (`engine=clickhouse`): `OrderPlaced` is filtered by the pair's base `token`
 * when scoped (string column with a bloom-filter index on `topic1`), and the
 * fill / cancellation joins use `WHERE "orderId" IN (...)` against the same
 * index. The decoded `tick` column has no ClickHouse index, so tick sort is
 * still applied app-side. ClickHouse aggregates fills per order before the
 * app computes each remaining amount.
 *
 * `remaining = OrderPlaced.amount - sum(OrderFilled.amountFilled)`; orders
 * with an `OrderCancelled` event are excluded outright. Each row is annotated
 * with its pair: the scoped feed reuses the resolved `pair`, the unscoped
 * feed resolves the page's base tokens with one targeted `dex_pairs` lookup
 * (rows whose pair is unknown are dropped).
 */
async function getOrders(c: Context<App.Environment>, options: getOrders.Options) {
  const { chainId, cursor, limit, maker, order, page, pair, side, sort } = options
  const store = c.get('store')

  return Timing.time(c, 'orders', async () => {
    // Snapshot is invariant in `(chainId, base?, maker?)`; sort/side/order are
    // applied app-side so they don't fragment the cache. Short TTL keeps the
    // resting set fresh against incoming fills/cancellations without making
    // each request pay the scan cost.
    const snapshot = await Store.memoize(
      () =>
        loadRestingOrders(
          c,
          pair
            ? { chainId, base: pair.base, maker }
            : { chainId, maker, scanCap: globalOrderScanCap },
        ),
      {
        key: `exchange:v1:${chainId}:orders:${pair ? pair.base : 'all'}:${maker ?? '*'}:v1`,
        store,
        ttl: Ttl.seconds(15),
      },
    )

    const { hasMore, nextCursor, pageRows, total } = pageRestingOrders(snapshot.orders, {
      cursor,
      limit,
      order,
      page,
      side,
      sort,
    })

    const bases = pair ? [] : [...new Set(pageRows.map((row) => row.token))]
    const pairsByBase = await getPairsByBase(c, { bases, chainId })

    const data: z.output<typeof schema.getOrders.Order>[] = []
    for (const row of pageRows) {
      const rowPair = pair ?? pairsByBase.get(row.token)
      if (!rowPair) continue
      const shaped = shapeRestingOrder(row)
      if (!shaped) continue
      data.push({
        ...shaped,
        pair: {
          base: { address: rowPair.base },
          key: rowPair.key,
          quote: { address: rowPair.quote },
        },
      })
    }

    return { data, hasMore, nextCursor, total, truncated: snapshot.truncated }
  })
}

declare namespace getOrders {
  /** Options for the resting-order page query. */
  type Options = {
    chainId: z.output<typeof Schema.ChainId>
    /** Opaque keyset cursor anchoring the page; omit for the head page. */
    cursor?: string | undefined
    limit: number
    /** Only include orders placed by this maker. */
    maker?: z.output<typeof Schema.Address> | undefined
    /** 1-indexed page number (positional pagination; exclusive with `cursor`). */
    page?: number | undefined
    /** Sort direction applied to whatever `sort` selects. */
    order: 'asc' | 'desc'
    /** When set, scope the snapshot to this resolved pair. */
    pair?: getPairsByBase.Pair | undefined
    /** Restrict to one side of the book. */
    side?: 'bid' | 'ask' | undefined
    /** Sort key. `tick` ranks by price; `time` orders by placement (block, log). */
    sort: 'tick' | 'time'
  }
}

/**
 * Applies the side filter, sort, and keyset pagination over an in-memory
 * resting-order snapshot, returning the page rows plus pagination state.
 * Shared by the per-pair ({@link getPairOrders}) and global ({@link getOrders})
 * feeds so the sort comparator and cursor walk never drift between them.
 *
 * Sort: `tick` ranks by price (canonical orderbook orientation when paired
 * with `order=asc` for asks / `order=desc` for bids), tie-breaking by newest
 * placement so callers can always derive a stable cursor. `time` orders by
 * placement `(block_num, log_idx)`. The cursor encodes the full sort key (not
 * just `(block, log)`) so pagination stays correct even if the snapshot
 * rotates between requests and the previous-page row is gone.
 */
function pageRestingOrders(
  orders: readonly RestingOrder[],
  options: pageRestingOrders.Options,
): { hasMore: boolean; nextCursor: string | null; pageRows: RestingOrder[]; total: number } {
  const { cursor: cursorToken, limit, order, side, sort } = options
  // Bounded positional lane (exclusive with `cursor` at the schema): the
  // snapshot is in memory, so a positional page is an exact slice of
  // `(page - 1) * limit` rows.
  const page = options.page !== undefined && options.page > 1 ? options.page : undefined
  const offset = page !== undefined ? (page - 1) * limit : undefined

  // Side filter is the cheapest in-memory cut, so apply it first.
  const sided = side === undefined ? orders : orders.filter((row) => row.isBid === (side === 'bid'))

  const dir = order === 'asc' ? 1 : -1
  const sorted = [...sided].sort((a, b) => {
    if (sort === 'tick') {
      if (a.tick !== b.tick) return (a.tick - b.tick) * dir
      if (a.blockNumber !== b.blockNumber) return b.blockNumber - a.blockNumber
      return b.logIndex - a.logIndex
    }
    if (a.blockNumber !== b.blockNumber) return (a.blockNumber - b.blockNumber) * dir
    return (a.logIndex - b.logIndex) * dir
  })

  const cursorFields: readonly Cursor.Field[] =
    sort === 'tick' ? ['uint', 'int', 'int'] : ['int', 'int']
  const cursor = cursorToken ? Cursor.decode(cursorToken, cursorFields) : undefined

  let startIndex = offset ?? 0
  if (cursor) {
    while (startIndex < sorted.length && !isPastCursor(sorted[startIndex]!, cursor, sort, dir))
      startIndex += 1
  }

  const window = sorted.slice(startIndex, startIndex + limit + 1)
  const hasMore = window.length > limit
  const pageRows = window.slice(0, limit)
  const last = hasMore ? pageRows.at(-1) : undefined
  const nextTuple = last ? cursorTupleFromOrder(last, sort) : undefined
  const nextCursor = nextTuple ? Cursor.encode(nextTuple) : null

  // `total` is the count of side-filtered resting orders before pagination —
  // the in-memory basis for the opt-in `totalCount` (a lower bound when the
  // underlying placement scan was truncated).
  return { hasMore, nextCursor, pageRows, total: sided.length }
}

declare namespace pageRestingOrders {
  /** Options for {@link pageRestingOrders}. */
  type Options = {
    /** Opaque keyset cursor anchoring the page; omit for the head page. */
    cursor?: string | undefined
    limit: number
    /** 1-indexed page number (positional pagination; exclusive with `cursor`). */
    page?: number | undefined
    /** Sort direction applied to whatever `sort` selects. */
    order: 'asc' | 'desc'
    /** Restrict to one side of the book. */
    side?: 'bid' | 'ask' | undefined
    /** Sort key. `tick` ranks by price; `time` orders by placement (block, log). */
    sort: 'tick' | 'time'
  }
}

/**
 * Number of grid ticks probed per side per multicall round while walking the
 * book outward from its best tick. Real books cluster near peg, so the
 * default `levels` resolves in one round for a typical book; a sparse book
 * keeps walking in rounds of this size until the tick range is exhausted
 * (worst case ~401 positions per side, the old blanket-scan cost).
 */
const depthScanChunk = 100

/**
 * Resolves the current orderbook depth for a single pair, denominated in the
 * base token. The stablecoin DEX precompile exposes per-tick aggregates via
 * `getTickLevel(base, tick, isBid)`. Rather than blanket-scanning every valid
 * tick (~401 per side — multiple seconds of serial EVM work per request), we
 * first read the book head (`books(pairKey)` → `bestBidTick`/`bestAskTick`)
 * and walk each side outward from its best tick in {@link depthScanChunk}
 * chunks, stopping as soon as `levels` non-empty levels are collected. Viem's
 * deployless multicall coalesces each chunk into one `eth_call`. The best
 * tick is by definition the side's outermost populated tick, so the outward
 * walk observes every non-empty level in best-first order. A side whose best
 * tick sits outside the valid range is empty (the precompile signals "no
 * orders" with the int16 min/max sentinels) and is returned without scanning.
 *
 * Memoized briefly via `Store.memoize` so concurrent dashboard requests for
 * the same pair share one RPC fanout; the route's `Cache.policies.feed`
 * response cache layers on top of that.
 */
async function getPairDepth(c: Context<App.Environment>, options: getPairDepth.Options) {
  const { chainId, levels, pair } = options
  const client = c.get('getClient')(chainId)
  const store = c.get('store')

  /**
   * Walks one side outward from its best tick, probing `depthScanChunk` grid
   * positions per round, until `levels` non-empty levels are collected or the
   * tick range is exhausted. Probed ticks stay in best-first order, so
   * `collectLevels` builds `cumulativeSize` and applies the cap directly.
   */
  const scanSide = async (isBid: boolean, bestTick: number) => {
    if (bestTick < Tick.minTick || bestTick > Tick.maxTick) return []
    const step = isBid ? -tickSpacing : tickSpacing
    const boundary = isBid ? Tick.minTick : Tick.maxTick
    const inRange = (tick: number) => (isBid ? tick >= boundary : tick <= boundary)

    const probedTicks: number[] = []
    const probedLevels: { totalLiquidity: bigint }[] = []
    let nonEmpty = 0
    let cursor = bestTick
    while (nonEmpty < levels && inRange(cursor)) {
      const chunk: number[] = []
      while (chunk.length < depthScanChunk && inRange(cursor)) {
        chunk.push(cursor)
        cursor += step
      }
      // Each chunk's per-tick reads share one deployless multicall round trip
      // via viem's auto-batching. Only `totalLiquidity` is needed.
      const results = await Promise.all(
        chunk.map((tick) => client.dex.getTickLevel({ base: pair.base, isBid, tick })),
      )
      for (let i = 0; i < chunk.length; i++) {
        probedTicks.push(chunk[i]!)
        probedLevels.push(results[i]!)
        if (results[i]!.totalLiquidity > 0n) nonEmpty += 1
      }
    }
    return collectLevels(probedTicks, probedLevels, {
      limit: levels,
      pairKey: pair.key,
      side: isBid ? 'bid' : 'ask',
    })
  }

  return Timing.time(c, 'pair_depth', () =>
    Store.memoize(
      async () => {
        const book = await Actions.dex.getOrderbook(client, {
          base: pair.base,
          quote: pair.quote,
        })
        const [asks, bids] = await Promise.all([
          scanSide(false, book.bestAskTick),
          scanSide(true, book.bestBidTick),
        ])
        return { asks, bids }
      },
      {
        key: `exchange:v1:${chainId}:depth:${pair.base}:${pair.quote}:levels:${levels}`,
        store,
        // The response cache (10s) already absorbs concurrent reads at the
        // edge; this in-process memo just dedupes overlapping in-flight
        // multicalls within a single instance.
        ttl: Ttl.seconds(5),
      },
    ),
  )
}

declare namespace getPairDepth {
  type Options = {
    chainId: z.output<typeof Schema.ChainId>
    /** Maximum number of non-empty levels to keep per side, in best-first order. */
    levels: number
    pair: getPairsByBase.Pair
  }
}

/**
 * Returns `true` when `row` falls strictly after the position the cursor was
 * issued at, under the active sort. Mirrors the comparator in `getPairOrders`
 * so the cursor walk and the sort agree on row order.
 */
function isPastCursor(
  row: RestingOrder,
  cursor: Cursor.Cursor,
  sort: 'tick' | 'time',
  dir: 1 | -1,
): boolean {
  if (sort === 'tick') {
    const cursorTick = Number(cursor[0]) - orderCursorTickBias
    const cursorBlock = cursor[1] as number
    const cursorLog = cursor[2] as number
    if (row.tick !== cursorTick) return (row.tick - cursorTick) * dir > 0
    if (row.blockNumber !== cursorBlock) return row.blockNumber < cursorBlock
    return row.logIndex < cursorLog
  }
  const cursorBlock = cursor[0] as number
  const cursorLog = cursor[1] as number
  if (row.blockNumber !== cursorBlock) return (row.blockNumber - cursorBlock) * dir > 0
  return (row.logIndex - cursorLog) * dir > 0
}

/** Builds the cursor tuple for the last row of a page, matching the active sort. */
function cursorTupleFromOrder(row: RestingOrder, sort: 'tick' | 'time'): Cursor.Cursor {
  if (sort === 'tick')
    return [String(row.tick + orderCursorTickBias), row.blockNumber, row.logIndex]
  return [row.blockNumber, row.logIndex]
}

/**
 * In-memory representation of a single resting order. Field types are already
 * narrowed/validated; the public response shape is built by `shapeRestingOrder`.
 *
 * `amount` and `remaining` are decimal-string integers (not `bigint`) so the
 * snapshot survives `JSON.stringify` inside `Store.memoize`. Bigint arithmetic
 * is kept inside `loadRestingOrders` where fills are subtracted from the raw
 * placement amount.
 */
type RestingOrder = {
  amount: string
  blockNumber: number
  isBid: boolean
  logIndex: number
  maker: z.output<typeof Schema.Address>
  orderId: string
  remaining: string
  tick: number
  timestamp: string
  /** The order's base token, used by the global feed to resolve its pair. */
  token: z.output<typeof Schema.TokenAddress>
  transactionHash: z.output<typeof Schema.Hash>
}

/**
 * Loads every resting maker order for a pair: scan `OrderPlaced` filtered by
 * the pair's base token (and optional `maker`), then subtract fills and drop
 * cancelled orders using batched `WHERE "orderId" IN (...)` joins.
 *
 * Returns the resting set along with a `truncated` flag set when the
 * `OrderPlaced` scan hit `orderScanCap`. Callers surface the flag so
 * pagination consumers can detect a partial view of the book.
 */
async function loadRestingOrders(
  c: Context<App.Environment>,
  options: {
    /** Restrict to one pair's base token; omit for the global feed (all pairs). */
    base?: z.output<typeof Schema.TokenAddress> | undefined
    chainId: z.output<typeof Schema.ChainId>
    maker?: z.output<typeof Schema.Address> | undefined
    /** Newest-placement scan window; defaults to {@link orderScanCap}. */
    scanCap?: number | undefined
  },
): Promise<{ orders: RestingOrder[]; truncated: boolean }> {
  const { base, chainId, maker, scanCap = orderScanCap } = options
  const tidx = c.get('getTidx')(chainId)

  // Scan window is intentionally newest-first so a `truncated` snapshot still
  // reflects current market activity rather than ancient placements.
  const filters = [`address = '${stablecoinDex}'`]
  if (base !== undefined) filters.push(`token = '${base}'`)
  if (maker !== undefined) filters.push(`maker = '${maker}'`)
  const where = `WHERE ${filters.join(' AND ')}`

  // ClickHouse OLAP path: bloom-filter index on `topic1` (and the `token`
  // string column) handles `WHERE token = '<base>'` and the downstream
  // `"orderId" IN (...)` joins natively, without the per-row decode cost of
  // the Postgres `abi_uint(topic1)` fallback.
  //
  // `"isBid"` is aliased to a name that is *not* in the OrderPlaced event
  // signature so tidx.ts's signature-driven result decoder skips it. ClickHouse
  // serializes booleans as `1`/`0`, but tidx.ts validates `bool` columns with a
  // strict `z.boolean()` derived from the signature and rejects integer
  // representations. Aliasing dodges that decoder and lets us coerce
  // numerically in `parsePlacedRow`.
  const result = await tidx.fetch({
    chainId,
    engine: 'clickhouse',
    query: `
      SELECT "orderId", maker, amount, token, "isBid" AS is_bid_int, tick, tx_hash, block_num, log_idx, block_timestamp
      FROM dex_orders
      ${where}
      ORDER BY block_num DESC, log_idx DESC
      LIMIT ${scanCap + 1}
    ` as string,
  })

  const truncated = result.rows.length > scanCap
  const placedRows = result.rows.slice(0, scanCap)
  // `dex_orders` can carry duplicate rows per placement; keep the first
  // (newest) copy so pages emit each order once and a duplicate straddling an
  // IN-batch boundary can't double-count fills.
  const placed: RestingOrder[] = []
  const seen = new Set<string>()
  for (const row of placedRows) {
    const parsed = parsePlacedRow(row)
    if (!parsed || seen.has(parsed.orderId)) continue
    seen.add(parsed.orderId)
    placed.push(parsed)
  }
  if (placed.length === 0) return { orders: [], truncated }

  const orderIds = placed.map((row) => row.orderId)
  const [cancelled, fillSums] = await Promise.all([
    fetchCancelledByIds(c, { chainId, orderIds }),
    fetchFillSumsByIds(c, { chainId, orderIds }),
  ])

  const orders: RestingOrder[] = []
  for (const row of placed) {
    if (cancelled.has(row.orderId)) continue
    const filled = fillSums.get(row.orderId) ?? 0n
    const remaining = BigInt(row.amount) - filled
    if (remaining <= 0n) continue
    orders.push({ ...row, remaining: remaining.toString() })
  }

  return { orders, truncated }
}

/**
 * Base-token chunk size for the multi-book `WHERE token IN (...)` scan in
 * {@link resolveBookReserves}. `tidx.fetch` transmits queries as GET, so a
 * single `IN (...)` of every base in a wide page overflows the request URI;
 * 200 lowercase 20-byte literals stay comfortably under the indexer's ~60 KB
 * SQL ceiling. Chunks resolve serially and merge.
 */
const reserveBaseInBatchSize = 200

/**
 * Computes per-book resting-order reserves for a set of base tokens, mirroring
 * the legacy `/gecko` adapter's pool-liquidity semantics: scan `OrderPlaced`
 * scoped to the given bases, subtract fills, drop cancelled orders, then
 * aggregate each order's `remaining` by side — asks into the base reserve, bids
 * into the quote reserve.
 *
 * The bid leg is added in its placed (base-denominated) units **without** tick
 * conversion, matching the legacy adapter exactly so GeckoTerminal's per-pool
 * `reserve_in_usd` stays consistent with the figures it has indexed historically.
 * (The on-chain `balanceOf` of the DEX settlement account — what the adapter
 * used previously — is a single shared pool across every book sharing a quote
 * token, so it cannot express per-book liquidity.)
 *
 * Reserves are keyed by **base** token (a book's identity), so two books sharing
 * a quote token report distinct reserves. The `OrderPlaced` scan uses a fair
 * per-token cap (`LIMIT ... BY token`) so one active book can't starve the
 * others' newest placements under a shared budget; `truncated` is set when any
 * book hits the cap. Like {@link loadRestingOrders}, this reads only
 * `OrderPlaced` (`dex_orders`), so flip orders (which mutate side/tick via
 * `OrderFlipped` without a new placement) are not reflected — acceptable for
 * legacy parity, but not an exact economic TVL.
 */
export async function resolveBookReserves(
  c: Context<App.Environment>,
  options: resolveBookReserves.Options,
): Promise<resolveBookReserves.Result> {
  const { bases, chainId, scanCap = orderScanCap } = options
  const reserves = new Map<string, resolveBookReserves.Reserve>()
  const unique = [...new Set(bases.map((base) => base.toLowerCase()))]
  if (unique.length === 0) return { reserves, truncated: false }

  const tidx = c.get('getTidx')(chainId)

  // Scan the newest `scanCap` `OrderPlaced` rows per base. `LIMIT ... BY token`
  // is the ClickHouse fair cap over raw rows; `rawPerToken` flips `truncated`
  // when a token's window overflows, since unfetched older rows may hold unique
  // placements even when duplicates keep the unique count under the cap.
  let truncated = false
  const placed: RestingOrder[] = []
  const perToken = new Map<string, number>()
  const rawPerToken = new Map<string, number>()
  const seen = new Set<string>()
  for (let i = 0; i < unique.length; i += reserveBaseInBatchSize) {
    const group = unique.slice(i, i + reserveBaseInBatchSize)
    const list = group.map((base) => `'${base}'`).join(', ')
    const result = await tidx.fetch({
      chainId,
      engine: 'clickhouse',
      query: `
        SELECT "orderId", maker, amount, token, "isBid" AS is_bid_int, tick, tx_hash, block_num, log_idx, block_timestamp
        FROM dex_orders
        WHERE address = '${stablecoinDex}' AND token IN (${list})
        ORDER BY block_num DESC, log_idx DESC
        LIMIT ${scanCap + 1} BY token
      ` as string,
    })
    for (const row of result.rows) {
      const parsed = parsePlacedRow(row)
      if (!parsed) continue
      const raw = (rawPerToken.get(parsed.token) ?? 0) + 1
      rawPerToken.set(parsed.token, raw)
      if (raw > scanCap) truncated = true
      // Duplicate placement rows (see `loadRestingOrders`) can't add the same
      // order's remaining liquidity twice; the raw probe above still counts them.
      if (seen.has(parsed.orderId)) continue
      const count = perToken.get(parsed.token) ?? 0
      if (count >= scanCap) continue
      seen.add(parsed.orderId)
      perToken.set(parsed.token, count + 1)
      placed.push(parsed)
    }
  }
  if (placed.length === 0) return { reserves, truncated }

  const orderIds = placed.map((row) => row.orderId)
  const [cancelled, fillSums] = await Promise.all([
    fetchCancelledByIds(c, { chainId, orderIds }),
    fetchFillSumsByIds(c, { chainId, orderIds }),
  ])

  for (const row of placed) {
    if (cancelled.has(row.orderId)) continue
    const filled = fillSums.get(row.orderId) ?? 0n
    const remaining = BigInt(row.amount) - filled
    if (remaining <= 0n) continue
    const reserve = reserves.get(row.token) ?? { base: 0n, quote: 0n }
    if (row.isBid) reserve.quote += remaining
    else reserve.base += remaining
    reserves.set(row.token, reserve)
  }

  return { reserves, truncated }
}

export declare namespace resolveBookReserves {
  type Options = {
    /** Book base tokens to compute reserves for (case-insensitive). */
    bases: readonly string[]
    chainId: z.output<typeof Schema.ChainId>
    /** Newest-placement scan window per base; defaults to {@link orderScanCap}. */
    scanCap?: number | undefined
  }

  /** Per-book resting liquidity, in each token's smallest unit. */
  type Reserve = {
    /** Ask-side resting liquidity (base token units). */
    base: bigint
    /** Bid-side resting liquidity (base-denominated, legacy-compatible). */
    quote: bigint
  }

  type Result = {
    /** Reserves keyed by lowercase base token. */
    reserves: Map<string, Reserve>
    /** Set when any book hit `scanCap` (older placements may be missing). */
    truncated: boolean
  }
}

/**
 * Parses one `OrderPlaced` row into the in-memory `RestingOrder` shape, with
 * `remaining` provisionally set to `amount` (the loader fills it in once fills
 * are joined). Returns `undefined` for malformed rows so the caller can drop
 * them without failing the whole snapshot.
 */
function parsePlacedRow(row: Record<string, unknown>): RestingOrder | undefined {
  const orderId = Value.toIntegerString(row['orderId'])
  const maker = Schema.Address.safeParse(row['maker'])
  const amount = Value.toIntegerString(row['amount'])
  const tick = Value.toNumber(row['tick'])
  const transactionHash = Schema.Hash.safeParse(row['tx_hash'])
  const blockNumber = Value.toNumber(row['block_num'])
  const logIndex = Value.toNumber(row['log_idx'])
  const timestamp = Value.toIsoDateTime(row['block_timestamp'])
  const token = Schema.TokenAddress.safeParse(row['token'])
  if (
    orderId === undefined ||
    !maker.success ||
    amount === undefined ||
    tick === undefined ||
    !transactionHash.success ||
    blockNumber === undefined ||
    logIndex === undefined ||
    timestamp === undefined ||
    !token.success
  )
    return undefined
  // `is_bid_int` is the aliased `OrderPlaced."isBid"` column; ClickHouse
  // serializes booleans as `1`/`0`, and Postgres returns native booleans, so
  // coerce both representations.
  const isBidRaw = row['is_bid_int']
  const isBid = isBidRaw === true || isBidRaw === 1 || isBidRaw === '1'
  // `amount` is already a decimal string (from `Value.toIntegerString`);
  // `remaining` is provisionally the same until `loadRestingOrders` subtracts
  // fills against it.
  return {
    amount,
    blockNumber,
    isBid,
    logIndex,
    maker: maker.data,
    orderId,
    remaining: amount,
    tick,
    timestamp,
    token: token.data,
    transactionHash: transactionHash.data,
  }
}

/**
 * Resolves the set of cancelled `orderId`s in the snapshot window via batched
 * `OrderCancelled WHERE "orderId" IN (...)` queries. Empty input short-circuits
 * without touching the indexer.
 */
async function fetchCancelledByIds(
  c: Context<App.Environment>,
  options: { chainId: z.output<typeof Schema.ChainId>; orderIds: readonly string[] },
): Promise<Set<string>> {
  const { chainId, orderIds } = options
  const cancelled = new Set<string>()
  if (orderIds.length === 0) return cancelled
  const tidx = c.get('getTidx')(chainId)
  // Chunks resolve serially: the managed indexer rejects bursts of concurrent
  // chunk queries (~2-3 in flight), and the cancel/fill helpers already
  // overlap via `Promise.all`.
  for (let i = 0; i < orderIds.length; i += orderInBatchSize) {
    const batch = orderIds.slice(i, i + orderInBatchSize)
    const result = await tidx.fetch({
      chainId,
      engine: 'clickhouse',
      query: `
        SELECT "orderId"
        FROM OrderCancelled
        WHERE address = '${stablecoinDex}' AND "orderId" IN (${batch.join(', ')})
      `,
      signatures: [orderCancelledSignature],
    })
    for (const row of result.rows) {
      const id = Value.toIntegerString(row['orderId'])
      if (id !== undefined) cancelled.add(id)
    }
  }
  return cancelled
}

/**
 * Sums `amountFilled` per `orderId` across all matching native `dex_fills`
 * rows in the snapshot window via batched `WHERE "orderId" IN (...)` queries.
 * Native aggregation bounds results, and stringification preserves UInt256 precision through JSON parsing.
 */
async function fetchFillSumsByIds(
  c: Context<App.Environment>,
  options: { chainId: z.output<typeof Schema.ChainId>; orderIds: readonly string[] },
): Promise<Map<string, bigint>> {
  const { chainId, orderIds } = options
  const sums = new Map<string, bigint>()
  if (orderIds.length === 0) return sums
  const tidx = c.get('getTidx')(chainId)
  // Chunks resolve serially: the managed indexer rejects bursts of concurrent
  // chunk queries (~2-3 in flight), and the cancel/fill helpers already
  // overlap via `Promise.all`.
  for (let i = 0; i < orderIds.length; i += orderInBatchSize) {
    const batch = orderIds.slice(i, i + orderInBatchSize)
    const result = await tidx.fetch({
      chainId,
      engine: 'clickhouse',
      // Dynamic order IDs prevent signature-backed row inference; each aggregate is validated below before use.
      query: `
        SELECT "orderId", toString(sum("amountFilled")) AS amount_filled
        FROM dex_fills
        WHERE address = '${stablecoinDex}' AND "orderId" IN (${batch.join(', ')})
        GROUP BY "orderId"
      ` as string,
    })
    for (const row of result.rows) {
      const id = Value.toIntegerString(row['orderId'])
      const filled = Value.toIntegerString(row['amount_filled'])
      if (id === undefined || filled === undefined) continue
      sums.set(id, BigInt(filled))
    }
  }
  return sums
}

/**
 * Validates and shapes one resting order into the public response shape. Rate
 * is destination/source ratio relative to a taker: `price` for bids,
 * `1/price` for asks, formatted to the same 5-dp precision as `Tick.toPrice`
 * uses for `Tick.priceScale`.
 */
function shapeRestingOrder(
  row: RestingOrder,
): z.output<typeof schema.getOrders.RestingOrder> | undefined {
  const priceScale = BigInt(Tick.priceScale)
  const tickOffset = BigInt(row.tick)
  const price = Tick.toPrice(row.tick)
  const rate = row.isBid
    ? price
    : (
        Number(priceScale * priceScale) /
        Number(priceScale + tickOffset) /
        Number(priceScale)
      ).toFixed(5)

  return {
    amount: row.amount,
    blockNumber: row.blockNumber,
    id: row.orderId,
    logIndex: row.logIndex,
    maker: row.maker,
    orderId: row.orderId,
    placedAt: row.timestamp,
    price,
    rate,
    remaining: row.remaining,
    side: row.isBid ? 'bid' : 'ask',
    tick: row.tick,
    transactionHash: row.transactionHash,
  }
}

/**
 * Builds the per-side levels from a best-first list of `getTickLevel` results.
 * Filters out empty ticks, accumulates `size` into `cumulativeSize` as we walk
 * outward, and truncates to `limit` non-empty levels.
 */
function collectLevels(
  ticks: readonly number[],
  results: readonly { totalLiquidity: bigint }[],
  options: collectLevels.Options,
): z.output<typeof schema.getPairDepth.Level>[] {
  const { limit, pairKey, side } = options
  const out: z.output<typeof schema.getPairDepth.Level>[] = []
  let running = 0n
  for (let i = 0; i < ticks.length && out.length < limit; i++) {
    const size = results[i]?.totalLiquidity
    if (size === undefined || size === 0n) continue
    running += size
    out.push({
      cumulativeSize: running.toString(),
      id: `${pairKey}-${side}-${ticks[i]!}`,
      price: Tick.toPrice(ticks[i]!),
      size: size.toString(),
      tick: ticks[i]!,
    })
  }
  return out
}

declare namespace collectLevels {
  type Options = {
    limit: number
    pairKey: z.output<typeof Schema.Hash>
    side: 'ask' | 'bid'
  }
}

/**
 * Resolves a single DEX order's live placement + current state via the
 * on-chain `dex.getOrder` view. The contract returns `orderId === 0n` when
 * an order is unknown, which the caller surfaces as a 404. The pair the
 * order belongs to is resolved from `bookKey` against the cached pair index.
 *
 * Memoized briefly so a burst of `/orders/:orderId` requests for the same id
 * coalesces into a single RPC; the TTL stays short because `remaining`
 * decreases as fills land.
 */
async function getOrder(c: Context<App.Environment>, options: getOrder.Options) {
  const { chainId, orderId } = options
  const store = c.get('store')
  const client = c.get('getClient')(chainId)

  return Timing.time(c, 'order', () =>
    Store.memoize(
      async () => {
        // The on-chain DEX exposes the full order struct (placement params
        // plus current `remaining`) in one view call.
        const chainOrder = await Timing.time(c, 'order_dex_get', () =>
          // The DEX reverts with `OrderDoesNotExist()` for ids it no longer
          // holds (never placed, or filled/cancelled and cleared), rather
          // than returning the zero struct. Map that revert to not-found so
          // the route answers 404 instead of bubbling up as a 502.
          client.dex.getOrder({ orderId: BigInt(orderId) }).catch((cause) => {
            if (isOrderDoesNotExist(cause)) return undefined
            throw cause
          }),
        )

        // The DEX returns the zero struct (or reverts, handled above) when an
        // order id is unknown.
        if (!chainOrder || chainOrder.orderId === 0n) return undefined

        // Targeted lookup off the returned `bookKey`; the bounded pair index
        // misses books outside its newest window, 404ing live orders.
        const pair = await getPairByKey(c, {
          chainId,
          key: chainOrder.bookKey.toLowerCase(),
        })
        if (!pair) return undefined

        const tick = chainOrder.tick
        const isBid = chainOrder.isBid
        const tickOffset = BigInt(tick)
        const priceScaleBig = BigInt(Tick.priceScale)
        const price = Tick.toPrice(tick)
        const mode = isBid ? 'exactSource' : 'exactDestination'
        // Destination/source rate. Maker bids: taker sells base, receives
        // quote; rate is `price` (quote per base). Maker asks: taker buys
        // base; rate is `1/price`, formatted to the same fixed-decimal
        // precision the DEX price scale supplies.
        const rate = isBid
          ? price
          : (
              Number(priceScaleBig * priceScaleBig) /
              Number(priceScaleBig + tickOffset) /
              Number(priceScaleBig)
            ).toFixed(5)

        return {
          amount: chainOrder.amount.toString(),
          flipTick: chainOrder.flipTick,
          id: chainOrder.orderId.toString(),
          isBid,
          isFlipOrder: chainOrder.isFlip,
          maker: chainOrder.maker.toLowerCase() as z.output<typeof Schema.Address>,
          mode,
          orderId: chainOrder.orderId.toString(),
          pair: {
            base: { address: pair.base },
            key: pair.key,
            quote: { address: pair.quote },
          },
          price,
          rate,
          remaining: chainOrder.remaining.toString(),
          tick,
        }
      },
      {
        key: `exchange:v1:${chainId}:order:${orderId}`,
        store,
        ttl: Ttl.seconds(15),
      },
    ),
  )
}

declare namespace getOrder {
  type Options = {
    chainId: z.output<typeof Schema.ChainId>
    orderId: string
  }
}

/**
 * Detects the DEX `OrderDoesNotExist()` custom-error revert thrown by
 * `dex.getOrder` for an unknown order id. Walks viem's wrapped error chain to
 * the decoded `ContractFunctionRevertedError` and matches its `errorName`, so a
 * different revert (or a transport failure) still propagates as a 502.
 */
function isOrderDoesNotExist(error: unknown): boolean {
  if (!(error instanceof BaseError)) return false
  const revert = error.walk((e) => e instanceof ContractFunctionRevertedError)
  return (
    revert instanceof ContractFunctionRevertedError &&
    revert.data?.errorName === 'OrderDoesNotExist'
  )
}

async function providerQuoteResponse(
  c: Context<App.Environment>,
  options: {
    chainId: number
    gasPrice: bigint
    gasUnits: bigint
    include: readonly z.output<typeof schema.createQuote.Include>[]
    oracle: FxOracle.Oracle
    provider: ExchangeProvider.Provider
    query: z.output<typeof schema.createQuote.Query>
    quote: ExchangeProvider.QuoteResult
    request: z.output<typeof schema.createQuote.Request>
  },
) {
  const { chainId, gasPrice, gasUnits, oracle, provider, query, quote, request } = options
  const tokensByAddress = await Tokens.resolveTokens(c, {
    addresses: [request.destinationToken, request.sourceToken],
    chainId,
    include: options.include,
  })
  const destinationToken = tokensByAddress.get(request.destinationToken)
  const sourceToken = tokensByAddress.get(request.sourceToken)
  if (!destinationToken || !sourceToken)
    throw new Error('Unable to resolve exchange quote token metadata')

  const denomination = query['valuation.currency']
  const snapshot = denomination
    ? await VerifiedTokens.snapshot(c, chainId).catch(() => undefined)
    : undefined
  const rates = denomination
    ? await Valuation.ratesFor(c, {
        currencies: snapshot
          ? [request.destinationToken, request.sourceToken].flatMap((address) => {
              const held = snapshot.byAddress.get(address.toLowerCase())?.currency
              return held === undefined ? [] : [held]
            })
          : [],
        denomination,
        oracle,
      })
    : undefined
  const valued = (
    baseUnits: bigint | string,
    address: string,
    token: { currency: string; decimals: number },
  ) => ({
    ...Value.tokenAmount({ baseUnits, currency: token.currency, decimals: token.decimals }),
    ...(denomination
      ? {
          valuation: Valuation.valuationFor({
            amount: BigInt(baseUnits),
            denomination,
            rates,
            token: snapshot?.byAddress.get(address.toLowerCase()),
          }),
        }
      : {}),
  })
  const execution =
    quote.status === 'ready'
      ? {
          status: quote.status,
          transaction: { calls: [...quote.transaction.calls], chainId },
        }
      : quote.status === 'approvalRequired'
        ? { approval: { calls: [...quote.approval.calls] }, status: quote.status }
        : {
            continuation: quote.continuation,
            status: quote.status,
            typedData: quote.typedData,
          }
  const common = {
    destinationToken: { ...destinationToken, address: request.destinationToken },
    ...execution,
    ...(rates ? { meta: { valuation: Valuation.pricing(rates, oracle) } } : {}),
    gasEstimate: gasCost({ gasPrice, gasUnits }),
    provider: provider.id,
    sourceToken: { ...sourceToken, address: request.sourceToken },
  }

  if (request.mode === 'exactSource') {
    if (!quote.minimumDestinationAmount) throw new ExchangeProvider.ProviderPayloadError()
    return {
      ...common,
      destinationAmount: valued(
        quote.destinationAmount,
        request.destinationToken,
        destinationToken,
      ),
      destinationAmountMin: valued(
        quote.minimumDestinationAmount,
        request.destinationToken,
        destinationToken,
      ),
      mode: request.mode,
      sourceAmount: valued(quote.sourceAmount, request.sourceToken, sourceToken),
    }
  }

  if (!quote.maximumSourceAmount) throw new ExchangeProvider.ProviderPayloadError()
  return {
    ...common,
    destinationAmount: valued(quote.destinationAmount, request.destinationToken, destinationToken),
    mode: request.mode,
    sourceAmount: valued(quote.sourceAmount, request.sourceToken, sourceToken),
    sourceAmountMax: valued(quote.maximumSourceAmount, request.sourceToken, sourceToken),
  }
}

function providerExecution(options: providerExecution.Options): ExchangeProvider.Execution {
  const { quote, request } = options
  return {
    destinationAmount: quote.destinationAmount,
    destinationToken: request.destinationToken,
    gasUnits: quote.gasUnits,
    ...(quote.maximumSourceAmount ? { maximumSourceAmount: quote.maximumSourceAmount } : {}),
    ...(quote.minimumDestinationAmount
      ? { minimumDestinationAmount: quote.minimumDestinationAmount }
      : {}),
    mode: request.mode,
    sourceAmount: quote.sourceAmount,
    sourceToken: request.sourceToken,
  }
}

declare namespace providerExecution {
  type Options = {
    quote: ExchangeProvider.QuoteResult
    request: z.output<typeof schema.createQuote.Request>
  }
}

async function validateProviderCalls(
  client: Viem.getClient.ReturnType,
  options: validateProviderCalls.Options,
) {
  try {
    const { results } = await Actions.simulate.simulateCalls(client, {
      account: options.account,
      calls: options.calls.map((call) => ({
        data: call.data,
        to: call.to,
        value: BigInt(call.value),
      })),
      traceTransfers: true,
    })
    if (results.some((result) => result.status !== 'success'))
      throw new ExchangeProvider.ProviderPayloadError()

    const logs = results.flatMap((result) => result.logs ?? [])
    const transfers = parseEventLogs({
      abi: [AbiEvent.fromAbi(Abis.tip20, 'Transfer')],
      eventName: 'Transfer',
      logs,
    }).map((log) => ({
      amount: log.args.amount,
      from: log.args.from,
      to: log.args.to,
      token: log.address,
    }))
    ExchangeProvider.validateExecution({
      account: options.account,
      execution: options.execution,
      transfers,
    })
  } catch (cause) {
    if (cause instanceof ExchangeProvider.ProviderPayloadError) throw cause
    throw new ExchangeProvider.ProviderPayloadError()
  }
}

declare namespace validateProviderCalls {
  type Options = {
    account: `0x${string}`
    calls: readonly ExchangeProvider.Call[]
    execution: ExchangeProvider.Execution
  }
}

/** Converts estimated gas units to a native-token base-unit fee. */
function gasCost(options: { gasPrice: bigint; gasUnits: bigint }) {
  return (options.gasPrice * options.gasUnits).toString()
}

/** Returns the 404 used when no executable quote is available. */
function quoteNotAvailable(c: Context<App.Environment>) {
  return Response.error(c, {
    code: 'quote_not_available',
    message: 'No configured exchange provider has an executable quote',
    status: 404,
  })
}

/** Distinguishes native DEX route failures from RPC transport failures. */
function isQuoteUnavailable(error: unknown): boolean {
  if (!(error instanceof BaseError)) return false
  const revert = error.walk((e) => e instanceof ContractFunctionRevertedError)
  if (!(revert instanceof ContractFunctionRevertedError)) return false
  return (
    revert.raw === contractPausedSelector ||
    ['InsufficientLiquidity', 'InvalidBaseToken', 'InvalidToken', 'PairDoesNotExist'].includes(
      revert.data?.errorName ?? '',
    )
  )
}

/** Detects uint128 arithmetic overflow from exact-source DEX quote calls. */
function isQuoteAmountOutOfRange(error: unknown): boolean {
  if (!(error instanceof BaseError)) return false
  const revert = error.walk((e) => e instanceof ContractFunctionRevertedError)
  if (!(revert instanceof ContractFunctionRevertedError)) return false
  const data = revert.data
  return data?.errorName === 'Panic' && data.args?.[0] === 0x11n
}

/**
 * Resolves a single page of `OrderFilled` events for one order. Pagination is
 * keyset on `(block_num, log_idx)`, which is unique and append-only for the
 * stream of fills against any one order — so cursors stay stable across new
 * fills landing.
 */
async function getOrderFills(c: Context<App.Environment>, options: getOrderFills.Options) {
  const { chainId, cursor: cursorToken, limit, order, orderId } = options
  const store = c.get('store')
  const tidx = c.get('getTidx')(chainId)
  const direction = order === 'asc' ? 'ASC' : 'DESC'
  const cursor = cursorToken ? Cursor.decode(cursorToken, ['int', 'int']) : undefined
  // Bounded positional lane (exclusive with `cursor` at the schema): page 1 is
  // the head page, so it shares the head's SQL and cache entries; deeper pages
  // translate to a SQL OFFSET of `(page - 1) * limit` rows.
  const offset =
    options.page !== undefined && options.page > 1 ? (options.page - 1) * limit : undefined

  return Timing.time(c, 'order_fills', () =>
    Store.memoize(
      async () => {
        const filters: string[] = [`address = '${stablecoinDex}'`, `"orderId" = ${orderId}`]
        if (cursor !== undefined)
          filters.push(
            Cursor.keyset([
              { literal: Cursor.literal(cursor[0]!, 'int'), name: 'block_num', order },
              { literal: Cursor.literal(cursor[1]!, 'int'), name: 'log_idx', order },
            ]),
          )
        const where = `WHERE ${filters.join(' AND ')}`

        const result = await tidx.fetch({
          chainId,
          engine: 'clickhouse',
          query: `
            SELECT "orderId", taker, "amountFilled", "partialFill", tx_hash, block_num, log_idx, block_timestamp
            FROM dex_fills
            ${where}
            ORDER BY block_num ${direction}, log_idx ${direction}
            LIMIT ${limit + 1}${offset !== undefined ? ` OFFSET ${offset}` : ''}
          ` as string,
        })

        const page = Cursor.paginate({
          rows: result.rows,
          limit,
          key: (row) => {
            const block = Value.toNumber(row['block_num'])
            const index = Value.toNumber(row['log_idx'])
            return block !== undefined && index !== undefined ? [block, index] : undefined
          },
        })

        const data: z.output<typeof schema.getOrderFills.Fill>[] = []
        for (const row of page.rows) {
          const fill = parseOrderFillRow(row)
          if (fill) data.push(fill)
        }

        return { data, nextCursor: page.nextCursor }
      },
      {
        key: `exchange:v1:${chainId}:order:${orderId}:fills:${order}:${cursor ? `cursor:${cursor[0]}:${cursor[1]}` : offset !== undefined ? `page:${options.page}` : 'head'}:${limit}`,
        store,
        ttl: Ttl.seconds(10),
      },
    ),
  )
}

declare namespace getOrderFills {
  type Options = {
    chainId: z.output<typeof Schema.ChainId>
    cursor?: string | undefined
    limit: number
    /** 1-indexed page number (positional pagination; exclusive with `cursor`). */
    page?: number | undefined
    order: 'asc' | 'desc'
    orderId: string
  }
}

/**
 * Exact count of an order's fills, sharing {@link getOrderFills}'s
 * `address`/`orderId` filter (minus pagination) so the count matches the page
 * it annotates. Fills are 1:1 with `OrderFilled` events for the order; the
 * `address`/`"orderId"` predicate aligns with the `dex_fills` sort key, so
 * ClickHouse counts via the sparse primary index and the result is exact and
 * cheap with no cap (`totalCountCapped` is always `false`). Feeds the opt-in
 * `meta.totalCount`.
 */
async function countOrderFills(
  c: Context<App.Environment>,
  options: countOrderFills.Options,
): Promise<{ totalCountCapped: boolean; totalCount: number }> {
  const { chainId, orderId } = options
  const store = c.get('store')
  const tidx = c.get('getTidx')(chainId)

  return Timing.time(c, 'order_fills_count', () =>
    Store.memoize(
      async () => {
        const result = await tidx.fetch({
          chainId,
          engine: 'clickhouse',
          query:
            `SELECT count(*) AS total FROM dex_fills WHERE address = '${stablecoinDex}' AND "orderId" = ${orderId}` as string,
        })
        const totalCount = Value.toNumber(result.rows[0]?.['total']) ?? 0
        return { totalCountCapped: false, totalCount }
      },
      {
        key: `exchange:v1:${chainId}:order:${orderId}:fills:count`,
        store,
        ttl: Ttl.seconds(10),
      },
    ),
  )
}

declare namespace countOrderFills {
  type Options = {
    chainId: z.output<typeof Schema.ChainId>
    orderId: string
  }
}

/**
 * Validates and shapes one `OrderFilled` row into the public `Fill` shape.
 * Drops the row when any required column is malformed so the caller can omit
 * it from the page without failing the whole response.
 */
function parseOrderFillRow(
  row: Record<string, unknown>,
): z.output<typeof schema.getOrderFills.Fill> | undefined {
  const amountFilled = Value.toIntegerString(row['amountFilled'])
  const orderId = Value.toIntegerString(row['orderId'])
  const taker = Schema.Address.safeParse(row['taker'])
  const transactionHash = Schema.Hash.safeParse(row['tx_hash'])
  const blockNumber = Value.toNumber(row['block_num'])
  const logIndex = Value.toNumber(row['log_idx'])
  const timestamp = Value.toIsoDateTime(row['block_timestamp'])
  if (
    amountFilled === undefined ||
    orderId === undefined ||
    !taker.success ||
    !transactionHash.success ||
    blockNumber === undefined ||
    logIndex === undefined ||
    timestamp === undefined
  )
    return undefined
  return {
    amountFilled,
    blockNumber,
    filledAt: timestamp,
    id: `${transactionHash.data}-${logIndex}`,
    logIndex,
    orderId,
    partialFill: row['partialFill'] === true,
    taker: taker.data,
    transactionHash: transactionHash.data,
  }
}

class NativeDexAmountOutOfRangeError extends Error {
  override name = 'Exchanges.NativeDexAmountOutOfRangeError'
}