import { type Context, Hono } from 'hono'
import { createMiddleware } from 'hono/factory'
import { Address, Hash } from 'ox'
import { type Address as ViemAddress, formatUnits } from 'viem'
import { 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 OpenApi from '../../../internal/OpenApi.js'
import * as Response from '../../../internal/Response.js'
import * as Schema from '../../../internal/Schema.js'
import * as Timing from '../../../internal/Timing.js'
import * as Value from '../../../internal/Value.js'
import { resolveBookQuotes, resolveBookReserves, resolveOrderStates } from './exchanges.js'
import { getTokenMetadata, isTokenNotFound } from './tokens.js'

// Stablecoin DEX precompile. Pairs are announced by `PairCreated` (pre-decoded
// into `dex_pairs`); fills land in `dex_fills`. The lowercase form is for inline
// SQL; `dexAccount` keeps the checksummed address for on-chain reads.
const stablecoinDex = Addresses.stablecoinDex.toLowerCase()
const dexAccount = Addresses.stablecoinDex as ViemAddress

// `OrderFilled` event signature, used to read the latest decoded DEX fill block
// from the indexer's signature-backed event stream (mirrors the legacy
// `tempoxyz/api` adapter's `/latest-block` query).
const orderFilledSignature =
  'event OrderFilled(uint128 indexed orderId, address indexed maker, address indexed taker, uint128 amountFilled, bool partialFill)'

// Default and maximum page size backing the `/pairs` list endpoint's `?limit`
// search param. GeckoTerminal itself never enumerates pairs — it discovers them
// from `/events` and fetches `/pairs/:pairId` — so listing the full set (820k+
// pairs on Moderato) is neither required nor feasible. Callers tune the page
// size via `?limit`; the max stays a server-side cap so the list can't be made
// unbounded. Returns the newest pairs, mirroring `/v1/exchange/pairs`' index.
const defaultPairsLimit = 50
const maxPairsLimit = 500

// Total `OrderPlaced` budget for the `/pairs` list's per-book reserve scan. The
// per-book cap is this budget divided by the page size (floored at a small
// minimum), so the aggregate scan — and the downstream fill/cancel `IN`-joins —
// stays bounded regardless of `?limit`, and one very active book can't starve
// the others under a shared cap. `/pairs/:pairId` instead uses the full per-book
// default cap for an accurate single-pair reserve.
const listReserveOrderBudget = 10_000
const minListReserveScanCap = 20

// Stable identifier for the single venue this adapter exposes to GeckoTerminal.
const dexKey = 'tempo-stablecoin-dex'

/**
 * Builds a fully-qualified cache-key URL for `Cache.response`. The Workers Cache
 * API (used in production) rejects bare-string keys — "Cache API keys must be
 * fully-qualified, valid URLs" — so every coingecko cache key is namespaced
 * under a synthetic origin whose path/query identify the entry. (The in-memory
 * test store accepts any string, which is why this only surfaces in workerd.)
 */
function cacheKey(path: string): string {
  return new URL(path, 'https://cache.tempo-api.internal').toString()
}

// Largest `toBlock - fromBlock` span the `/events` endpoint will scan in one
// request. GeckoTerminal polls in small windows; a hard cap keeps a single
// request from sweeping the entire (10⁹-row) fill history and timing out
// upstream. Mirrors the swap feed's `swapWindowBlocks`.
const maxEventBlockRange = 100_000

// Default `/events` window when `fromBlock`/`toBlock` are omitted: `toBlock`
// defaults to the latest indexed block and `fromBlock` to `toBlock` minus this
// span. Kept small so the convenience default stays fast (the per-fill state
// scan grows with the window); callers wanting more pass an explicit range.
const defaultEventBlockRange = 1_000

// 36-dp fixed-point precision for `priceNative`, matching the GeckoTerminal DEX
// integration contract the legacy `/gecko` adapter served.
const pricePrecision = 36n

// `tidx.fetch` transmits queries as GET, so an `IN (...)` list of every id in a
// wide `/events` window overflows the request URI (the indexer drops the
// request near ~60 KB of SQL). These cap how many ids/addresses go into a single
// query; the rest are split across queries that run sequentially and merge. 256
// padded order-id literals produce ~37 KB of SQL — comfortably under the limit
// (verified: 384 ids/~55 KB still OK, 448 ids/~64 KB dropped).
const orderIdChunkSize = 256
const addressChunkSize = 200

// How many token reads (RPC `getMetadata`/`getBalance`) to fan out per batch.
// The client auto-batches concurrent `eth_call`s in the same macrotask into one
// deployless Multicall3 call; a whole page of pairs (~2N tokens) overflows the
// Tempo RPC's per-call limit ("Request exceeds defined limit"). Reading in small
// chunks — with an `await` between groups so each chunk is its own batch — keeps
// every multicall comfortably under the limit while still aggregating per chunk.
const tokenReadChunkSize = 20

// How many metadata chunks to keep in flight. Each `getMetadata` is a
// standalone deployless aggregate3 call (never batch-merged), so concurrency
// cuts wall time without growing any single RPC call.
const tokenReadConcurrency = 4

/** Zod schemas owned by the CoinGecko/GeckoTerminal compatibility adapter. */
export namespace schema {
  /**
   * A 20-byte hex address that preserves case. Unlike `Schema.Address` (which
   * lowercases on output), the GeckoTerminal contract returns EIP-55
   * checksummed addresses, so response bodies keep the checksum produced by
   * `Address.checksum`.
   */
  export const ChecksummedAddress = z
    .string()
    .check(
      z.regex(/^0x[0-9a-fA-F]{40}$/),
      z.describe('Checksummed 0x-prefixed 20-byte account or contract address.'),
    )

  /**
   * Path `chainId`. Unlike the rest of Tempo API (where `chainId` is an optional
   * query parameter defaulting from app composition), GeckoTerminal templates a
   * per-network base URL, so the chain is addressed positionally here.
   */
  export const ChainIdParam = z
    .object({
      chainId: Schema.ChainId.check(z.meta({ examples: [4217] })),
    })
    .check(
      z.describe(
        'Path values that select the Tempo network for this GeckoTerminal-compatible request.',
      ),
    )

  /** Indexed-block tip exposed to GeckoTerminal. */
  export const Block = Schema.describe(
    z.object({
      blockNumber: z
        .number()
        .check(
          z.int(),
          z.nonnegative(),
          z.describe('Block number on Tempo.'),
          z.meta({ examples: [23456789] }),
        ),
      blockTimestamp: z
        .number()
        .check(
          z.int(),
          z.nonnegative(),
          z.describe('Block time as Unix seconds.'),
          z.meta({ examples: [1718668800] }),
        ),
    }),
    'A Tempo block reference with its number and timestamp.',
  )

  /** `GET /:chainId/latest-block` response. */
  export const LatestBlockResponse = Schema.describe(
    z.object({ block: Block }),
    'The newest Tempo block that has been indexed for CoinGecko-compatible data.',
  )

  /** Path parameters addressing a single asset. */
  export const AssetParams = z
    .object({
      chainId: Schema.ChainId.check(z.meta({ examples: [4217] })),
      address: Schema.TokenAddress.check(
        z.describe('TIP-20 token contract address for the asset.'),
      ),
    })
    .check(z.describe('Path values that select a TIP-20 asset on a Tempo network.'))

  /** A TIP-20 asset in GeckoTerminal shape. */
  export const Asset = Schema.describe(
    z.object({
      id: schema.ChecksummedAddress.check(
        z.describe('Checksummed TIP-20 token contract address.'),
        z.meta({ examples: ['0x20c0000000000000000000008f5425160ebe5525'] }),
      ),
      name: z
        .string()
        .check(z.describe('Human-readable token name.'), z.meta({ examples: ['Tempo USD'] })),
      symbol: z
        .string()
        .check(z.describe('Short token ticker symbol.'), z.meta({ examples: ['USDT'] })),
      decimals: z
        .number()
        .check(
          z.int(),
          z.gte(0),
          z.lte(255),
          z.describe('Number of decimal places the token uses.'),
          z.meta({ examples: [6] }),
        ),
      totalSupply: z
        .string()
        .check(
          z.describe(
            'Total token supply as a decimal string, already adjusted for token decimals.',
          ),
          z.meta({ examples: ['1000000'] }),
        ),
    }),
    'A TIP-20 token in CoinGecko’s GeckoTerminal asset format.',
  )

  /** `GET /:chainId/assets/:address` response. */
  export const AssetResponse = Schema.describe(
    z.object({ asset: Asset }),
    'One TIP-20 token in CoinGecko’s GeckoTerminal asset format.',
  )

  /** A pair-side token (asset plus checksummed address). */
  export const PairToken = Schema.describe(
    z.object({
      address: schema.ChecksummedAddress.check(
        z.describe('Checksummed TIP-20 token contract address.'),
        z.meta({ examples: ['0x20c0000000000000000000008f5425160ebe5525'] }),
      ),
      name: z
        .string()
        .check(z.describe('Human-readable token name.'), z.meta({ examples: ['Tempo USD'] })),
      symbol: z
        .string()
        .check(z.describe('Short token ticker symbol.'), z.meta({ examples: ['USDT'] })),
      decimals: z
        .number()
        .check(
          z.int(),
          z.gte(0),
          z.lte(255),
          z.describe('Number of decimal places the token uses.'),
          z.meta({ examples: [6] }),
        ),
      totalSupply: z
        .string()
        .check(
          z.describe(
            'Total token supply as a decimal string, already adjusted for token decimals.',
          ),
          z.meta({ examples: ['1000000'] }),
        ),
    }),
    'One token side of a trading pair.',
  )

  /** A GeckoTerminal trading pair. */
  export const Pair = Schema.describe(
    z.object({
      id: Schema.Hash.check(z.describe('32-byte onchain order-book key for this trading pair.')),
      dexKey: z
        .string()
        .check(
          z.describe('Stable identifier for Tempo’s DEX venue.'),
          z.meta({ examples: [dexKey] }),
        ),
      asset0Id: schema.ChecksummedAddress.check(
        z.describe('Checksummed address of the base token (`asset0`).'),
        z.meta({ examples: ['0x20c0000000000000000000008f5425160ebe5525'] }),
      ),
      asset1Id: schema.ChecksummedAddress.check(
        z.describe('Checksummed address of the quote token (`asset1`).'),
        z.meta({ examples: ['0x20C0000000000000000000000000000000000000'] }),
      ),
      token0: PairToken,
      token1: PairToken,
      reserve0: z
        .string()
        .check(
          z.describe('Best-effort current base-side reserve as a decimal string.'),
          z.meta({ examples: ['1000000'] }),
        ),
      reserve1: z
        .string()
        .check(
          z.describe('Best-effort current quote-side reserve as a decimal string.'),
          z.meta({ examples: ['1500000'] }),
        ),
      createdAtBlockNumber: z
        .optional(
          z
            .number()
            .check(
              z.int(),
              z.nonnegative(),
              z.describe('Block number where the pair was created, when indexed.'),
            ),
        )
        .check(z.meta({ examples: [23456789] })),
      createdAtBlockTimestamp: z
        .optional(
          z
            .number()
            .check(
              z.int(),
              z.nonnegative(),
              z.describe('Unix timestamp for when the pair was created, when indexed.'),
            ),
        )
        .check(z.meta({ examples: [1718668800] })),
      createdAtTxnId: z
        .optional(
          z.string().check(z.describe('Transaction hash that created the pair, when indexed.')),
        )
        .check(
          z.meta({
            examples: ['0x515801d7f9a5ac705e793e85904c9c69b3f1694b465cc8fb6ba3f0298dc82665'],
          }),
        ),
    }),
    'A Tempo stablecoin DEX trading pair in GeckoTerminal format.',
  )

  /** `GET /:chainId/pairs` response. */
  export const PairsResponse = Schema.describe(
    z.object({
      pairs: z.array(Pair).check(z.describe('Trading pairs ordered by creation block.')),
    }),
    'A page of Tempo stablecoin DEX trading pairs.',
  )

  /** `GET /:chainId/pairs/:pairId` response. */
  export const PairResponse = Schema.describe(
    z.object({ pair: Pair }),
    'One Tempo stablecoin DEX trading pair.',
  )

  /** Path parameters addressing a single pair. */
  export const PairParams = z
    .object({
      chainId: Schema.ChainId.check(z.meta({ examples: [4217] })),
      pairId: z.string().check(
        z.describe('32-byte onchain order-book key for this trading pair.'),
        z.meta({
          examples: ['0x44f7b8011db3e3647a530b4ff635726de5fafc8fa8ad10f0f31c0eb9dd52fc65'],
        }),
      ),
    })
    .check(z.describe('Path values that select one Tempo stablecoin DEX trading pair.'))

  /** `GET /:chainId/pairs` query parameters. */
  export const PairsQuery = z
    .object({
      limit: z
        ._default(
          z.coerce.number().check(z.int(), z.gte(1), z.lte(maxPairsLimit)),
          defaultPairsLimit,
        )
        .check(
          z.describe(
            `Maximum number of pairs to return. Defaults to ${defaultPairsLimit} and is capped at ${maxPairsLimit}.`,
          ),
          z.meta({ examples: [50] }),
        ),
    })
    .check(z.describe('Pagination controls for the trading-pair list.'))

  /** `GET /:chainId/events` query parameters. */
  export const EventsQuery = z
    .object({
      fromBlock: z
        .optional(z.coerce.number().check(z.int(), z.nonnegative()))
        .check(
          z.describe('First block to include. Defaults to `toBlock` minus 1000.'),
          z.meta({ examples: [23456789] }),
        ),
      toBlock: z
        .optional(z.coerce.number().check(z.int(), z.nonnegative()))
        .check(
          z.describe('Last block to include. Defaults to the latest indexed block.'),
          z.meta({ examples: [23456999] }),
        ),
    })
    .check(z.describe('Block range to scan for swap events, including both endpoints.'))

  /** A GeckoTerminal swap event (one per `OrderFilled`). */
  export const SwapEvent = Schema.describe(
    z.object({
      block: Block,
      eventType: z
        .literal('swap')
        .check(
          z.describe('Event type; always `swap` for this feed.'),
          z.meta({ examples: ['swap'] }),
        ),
      txnId: z.string().check(
        z.describe('Hash of the transaction that emitted the swap event.'),
        z.meta({
          examples: ['0x515801d7f9a5ac705e793e85904c9c69b3f1694b465cc8fb6ba3f0298dc82665'],
        }),
      ),
      txnIndex: z
        .number()
        .check(
          z.int(),
          z.nonnegative(),
          z.describe('Zero-based position of the transaction within its block.'),
          z.meta({ examples: [1] }),
        ),
      eventIndex: z
        .number()
        .check(
          z.int(),
          z.nonnegative(),
          z.describe('Zero-based position of this swap within its transaction.'),
          z.meta({ examples: [0] }),
        ),
      maker: schema.ChecksummedAddress.check(
        z.describe('Checksummed address of the taker, usually the transaction sender.'),
        z.meta({ examples: ['0xbe058e1c4df8a4366a387bf595b284246a93039e'] }),
      ),
      pairId: Schema.Hash.check(z.describe('On-chain order-book key for the trading pair.')),
      asset0In: z
        .optional(z.string())
        .check(
          z.describe('Base token (`asset0`) amount sent into the swap, when applicable.'),
          z.meta({ examples: ['1000'] }),
        ),
      asset1In: z
        .optional(z.string())
        .check(
          z.describe('Quote token (`asset1`) amount sent into the swap, when applicable.'),
          z.meta({ examples: ['1500'] }),
        ),
      asset0Out: z
        .optional(z.string())
        .check(
          z.describe('Base token (`asset0`) amount received from the swap, when applicable.'),
          z.meta({ examples: ['1000'] }),
        ),
      asset1Out: z
        .optional(z.string())
        .check(
          z.describe('Quote token (`asset1`) amount received from the swap, when applicable.'),
          z.meta({ examples: ['1500'] }),
        ),
      priceNative: z
        .string()
        .check(
          z.describe('Quote-per-base price as a 36-decimal-place string.'),
          z.meta({ examples: ['1.5'] }),
        ),
      reserves: z
        .object({
          asset0: z
            .string()
            .check(
              z.describe('Best-effort current base-token reserve.'),
              z.meta({ examples: ['1000000'] }),
            ),
          asset1: z
            .string()
            .check(
              z.describe('Best-effort current quote-token reserve.'),
              z.meta({ examples: ['1500000'] }),
            ),
        })
        .check(z.describe('Best-effort current reserves for both sides of the pair.')),
    }),
    'A Tempo DEX swap event in GeckoTerminal format.',
  )

  /** `GET /:chainId/events` response. */
  export const EventsResponse = Schema.describe(
    z.object({
      events: z
        .array(SwapEvent)
        .check(
          z.describe('Swap events in the requested block range, ordered by block and log index.'),
        ),
    }),
    'Tempo DEX swap events for the requested block range.',
  )
}

/**
 * Keeps a local supported-chain check when this route group is composed
 * independently of the shared data app.
 */
const guardChainId = createMiddleware<App.Environment>(async (c, next) => {
  const parsed = Schema.ChainId.safeParse(c.req.param('chainId'))
  const supported = c.get('supportedChainIds')
  if (!parsed.success || supported.has(parsed.data)) return next()
  return Response.unsupportedChainId(c, parsed.data, supported)
})

/** Composes the GeckoTerminal/CoinGecko DEX-integration compatibility adapter. */
export function coingecko() {
  return new Hono<App.Environment>()
    .use('/gecko/:chainId/*', guardChainId)
    .get(
      '/gecko/:chainId/latest-block',
      Auth.policy({ apiKey: { scopes: ['data:read'] }, mpp: true, public: true }),
      OpenApi.validate('param', schema.ChainIdParam, {
        code: 'chain_id_invalid',
        message: 'Invalid chain id',
      }),
      OpenApi.describeRoute({
        description:
          'Returns the latest indexed Tempo block in GeckoTerminal format so DEX data tools know where to start polling.',
        operationId: 'coingeckoLatestBlock',
        responses: OpenApi.responses({
          success: {
            description: 'Latest indexed Tempo block.',
            schema: schema.LatestBlockResponse,
          },
        }),
        summary: 'Get latest indexed block',
        tags: ['CoinGecko'],
      }),
      // The latest block is a point-in-time read of the indexed tip (capped to
      // the latest decoded DEX fill), so it is never cached and carries no
      // `Cache-Control` header (DEX tools poll this to learn where to start).
      async (c) => {
        if (Auth.narrowAccess) return Auth.paidAccessError(c)
        if (OpenApi.narrowValidation)
          return OpenApi.validationError(c, {
            code: 'chain_id_invalid',
            message: 'Invalid chain id',
          })

        const { chainId } = c.req.valid('param')
        try {
          const block = await getLatestBlock(c, chainId)
          if (!block) return notFound(c, 'No indexed block found')
          return c.json(Response.validated(schema.LatestBlockResponse, { block }), 200)
        } catch (cause) {
          return Response.upstream(c, cause)
        }
      },
    )
    .get(
      '/gecko/:chainId/assets/:address',
      Auth.policy({ apiKey: { scopes: ['data:read'] }, mpp: true, public: true }),
      OpenApi.validate('param', schema.AssetParams, {
        code: 'address_invalid',
        message: 'Invalid chain id or token address',
      }),
      OpenApi.describeRoute({
        description: 'Returns one TIP-20 asset by contract address in GeckoTerminal format.',
        operationId: 'coingeckoAsset',
        responses: OpenApi.responses({
          errors: {
            404: {
              description: 'No TIP-20 asset was found for that address.',
              codes: ['not_found'],
            },
          },
          success: {
            description: 'A TIP-20 token in CoinGecko’s GeckoTerminal asset format.',
            schema: schema.AssetResponse,
          },
        }),
        summary: 'Get asset',
        tags: ['CoinGecko'],
      }),
      Cache.response<App.Environment>({
        cacheControl: Cache.policies.metadata,
        name: 'tempo-api:coingecko:v1',
        key: (c) =>
          cacheKey(
            `/gecko/${c.req.param('chainId')}/assets/${c.req.param('address')?.toLowerCase()}`,
          ),
      }),
      async (c) => {
        if (Auth.narrowAccess) return Auth.paidAccessError(c)
        if (OpenApi.narrowValidation)
          return OpenApi.validationError(c, {
            code: 'address_invalid',
            message: 'Invalid chain id or token address',
          })

        const { address, chainId } = c.req.valid('param')
        try {
          const asset = await resolveAsset(c, chainId, address)
          return c.json(Response.validated(schema.AssetResponse, { asset }), 200)
        } catch (cause) {
          // Non-TIP-20 addresses (LP pairs, EOAs) fail the metadata read; that
          // is the documented 404, not an upstream failure.
          if (isTokenNotFound(cause)) return notFound(c, 'Asset not found')
          return Response.upstream(c, cause)
        }
      },
    )
    .get(
      '/gecko/:chainId/pairs',
      Auth.policy({ apiKey: { scopes: ['data:read'] }, mpp: true, public: true }),
      OpenApi.validate('param', schema.ChainIdParam, {
        code: 'chain_id_invalid',
        message: 'Invalid chain id',
      }),
      OpenApi.validate('query', schema.PairsQuery, {
        code: 'query_invalid',
        message: 'Invalid limit',
      }),
      OpenApi.describeRoute({
        description: 'Lists Tempo stablecoin DEX trading pairs in GeckoTerminal format.',
        operationId: 'coingeckoPairs',
        responses: OpenApi.responses({
          errors: { 400: 'The `limit` query parameter is invalid.' },
          success: {
            description: 'Trading pairs in GeckoTerminal format.',
            schema: schema.PairsResponse,
          },
        }),
        summary: 'List trading pairs',
        tags: ['CoinGecko'],
      }),
      Cache.response<App.Environment>({
        cacheControl: Cache.policies.state,
        name: 'tempo-api:coingecko:v1',
        key: (c) => {
          const limit = new URL(c.req.url).searchParams.get('limit') ?? String(defaultPairsLimit)
          return cacheKey(`/gecko/${c.req.param('chainId')}/pairs?limit=${limit}`)
        },
      }),
      async (c) => {
        if (Auth.narrowAccess) return Auth.paidAccessError(c)
        if (OpenApi.narrowValidation)
          return OpenApi.validationError(c, {
            code: 'query_invalid',
            message: 'Invalid limit',
          })

        const { chainId } = c.req.valid('param')
        const { limit } = c.req.valid('query')
        try {
          const rows = await listPairRows(c, chainId, limit)
          const tokens = rows.flatMap((row) => [row.base, row.quote])
          const metadata = await loadTokenMetadata(c, chainId, tokens)
          // Per-book resting-order reserves, with a fair per-book cap bounded by
          // the page size. Best-effort: a failed scan degrades to zero reserves
          // (uncached) rather than failing the whole page.
          const scanCap = Math.max(
            minListReserveScanCap,
            Math.floor(listReserveOrderBudget / Math.max(rows.length, 1)),
          )
          const reserves = await loadBookReserves(
            c,
            chainId,
            rows.map((row) => row.base),
            scanCap,
          )
          // A degraded best-effort page (some reads failed) must not be cached as
          // if authoritative; serve it once and let the next request retry.
          if (metadata.failed || reserves.failed) Cache.setPolicy(c, Cache.policies.noStore)
          const pairs = rows.map((row) =>
            resolvePairDetails(row, reserves.reserves, metadata.metadata),
          )
          return c.json(Response.validated(schema.PairsResponse, { pairs }), 200)
        } catch (cause) {
          return Response.upstream(c, cause)
        }
      },
    )
    .get(
      '/gecko/:chainId/pairs/:pairId',
      Auth.policy({ apiKey: { scopes: ['data:read'] }, mpp: true, public: true }),
      OpenApi.validate('param', schema.PairParams, {
        code: 'pair_id_invalid',
        message: 'Invalid pair id',
      }),
      OpenApi.describeRoute({
        description: 'Returns one Tempo stablecoin DEX trading pair in GeckoTerminal format.',
        operationId: 'coingeckoPair',
        responses: OpenApi.responses({
          errors: {
            404: { description: 'No trading pair was found for that id.', codes: ['not_found'] },
          },
          success: {
            description: 'One trading pair in GeckoTerminal format.',
            schema: schema.PairResponse,
          },
        }),
        summary: 'Get trading pair',
        tags: ['CoinGecko'],
      }),
      Cache.response<App.Environment>({
        cacheControl: Cache.policies.state,
        name: 'tempo-api:coingecko:v1',
        key: (c) =>
          cacheKey(
            `/gecko/${c.req.param('chainId')}/pairs/${c.req.param('pairId')?.toLowerCase()}`,
          ),
      }),
      async (c) => {
        if (Auth.narrowAccess) return Auth.paidAccessError(c)
        if (OpenApi.narrowValidation)
          return OpenApi.validationError(c, {
            code: 'pair_id_invalid',
            message: 'Invalid pair id',
          })

        const { chainId, pairId } = c.req.valid('param')
        if (!Hash.validate(pairId))
          return Response.error(c, {
            code: 'pair_id_invalid',
            message: 'Invalid pair id',
            status: 400,
          })

        try {
          const row = await getPairRow(c, chainId, pairId as `0x${string}`)
          if (!row) return notFound(c, 'Pair not found')
          const tokens = [row.base, row.quote]
          const metadata = await loadTokenMetadata(c, chainId, tokens)
          // Full per-book scan cap for the single book — an accurate reserve for
          // the pair GeckoTerminal fetches on demand.
          const reserves = await loadBookReserves(c, chainId, [row.base])
          if (metadata.failed || reserves.failed) Cache.setPolicy(c, Cache.policies.noStore)
          const pair = resolvePairDetails(row, reserves.reserves, metadata.metadata)
          return c.json(Response.validated(schema.PairResponse, { pair }), 200)
        } catch (cause) {
          return Response.upstream(c, cause)
        }
      },
    )
    .get(
      '/gecko/:chainId/events',
      Auth.policy({ apiKey: { scopes: ['data:read'] }, mpp: true, public: true }),
      OpenApi.validate('param', schema.ChainIdParam, {
        code: 'chain_id_invalid',
        message: 'Invalid chain id',
      }),
      OpenApi.validate('query', schema.EventsQuery, {
        code: 'query_invalid',
        message: 'Invalid block range',
      }),
      OpenApi.describeRoute({
        description: 'Returns Tempo DEX swap events over a block range in GeckoTerminal format.',
        operationId: 'coingeckoEvents',
        responses: OpenApi.responses({
          errors: { 400: 'The requested block range is invalid.' },
          success: {
            description: 'Swap events in the requested block range.',
            schema: schema.EventsResponse,
          },
        }),
        summary: 'List swap events',
        tags: ['CoinGecko'],
      }),
      Cache.response<App.Environment>({
        cacheControl: Cache.policies.state,
        name: 'tempo-api:coingecko:v1',
        key: (c) => {
          const url = new URL(c.req.url)
          const fromBlock = url.searchParams.get('fromBlock') ?? ''
          const toBlock = url.searchParams.get('toBlock') ?? ''
          return cacheKey(
            `/gecko/${c.req.param('chainId')}/events?fromBlock=${fromBlock}&toBlock=${toBlock}`,
          )
        },
      }),
      async (c) => {
        if (Auth.narrowAccess) return Auth.paidAccessError(c)
        if (OpenApi.narrowValidation)
          return OpenApi.validationError(c, {
            code: 'query_invalid',
            message: 'Invalid block range',
          })

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

        // `toBlock` defaults to the latest indexed block; `fromBlock` to
        // `toBlock` minus the default window. Both may still be passed
        // explicitly (GeckoTerminal always does).
        let toBlock = query.toBlock
        if (toBlock === undefined) {
          try {
            const block = await getLatestBlock(c, chainId)
            if (!block) return notFound(c, 'No indexed block found')
            toBlock = block.blockNumber
          } catch (cause) {
            return Response.upstream(c, cause)
          }
        }
        const fromBlock = query.fromBlock ?? Math.max(0, toBlock - defaultEventBlockRange)

        if (toBlock < fromBlock)
          return Response.error(c, {
            code: 'query_invalid',
            message: '`toBlock` must be greater than or equal to `fromBlock`',
            status: 400,
          })
        if (toBlock - fromBlock > maxEventBlockRange)
          return Response.error(c, {
            code: 'query_invalid',
            message: `Block range exceeds the maximum of ${maxEventBlockRange} blocks`,
            status: 400,
          })

        try {
          const events = await listEvents(c, chainId, fromBlock, toBlock)
          return c.json(Response.validated(schema.EventsResponse, { events }), 200)
        } catch (cause) {
          return Response.upstream(c, cause)
        }
      },
    )
}

/** Standard 404 envelope for the adapter. */
function notFound(c: Context<App.Environment>, message: string) {
  return Response.error(c, { code: 'not_found', message, status: 404 })
}

/**
 * Resolves the block tip GeckoTerminal should treat as fully indexed: the lower
 * of the latest indexed block and the latest indexed DEX `OrderFilled` event.
 * Capping to the fill tip prevents GeckoTerminal from polling `/events` past the
 * point swaps are decoded (it never re-polls a range), at the cost of a slightly
 * stale tip during swap-quiet periods. Mirrors the legacy `tempoxyz/api`
 * adapter's query and `min(blockTip, fillTip)` logic. It is never cached, so DEX
 * tools polling this endpoint always see the current tip.
 */
async function getLatestBlock(
  c: Context<App.Environment>,
  chainId: z.output<typeof Schema.ChainId>,
): Promise<z.output<typeof schema.Block> | undefined> {
  const tidx = c.get('getTidx')(chainId)
  const [blocksTip, fillsTip] = await Promise.all([
    tidx.fetch({
      chainId,
      query: 'SELECT num, timestamp FROM blocks ORDER BY num DESC LIMIT 1' as string,
    }),
    tidx.fetch({
      chainId,
      query: `SELECT block_num AS num, block_timestamp AS ts FROM OrderFilled WHERE address = '${stablecoinDex}' ORDER BY block_num DESC LIMIT 1`,
      signatures: [orderFilledSignature],
    }),
  ])

  const blocksRow = blocksTip.rows[0]
  const blocksBlock = blocksRow ? Value.toNumber(blocksRow['num']) : undefined
  if (blocksBlock === undefined) return undefined

  const fillsRow = fillsTip.rows[0]
  const fillsBlock = (fillsRow ? Value.toNumber(fillsRow['num']) : undefined) ?? 0
  const blockNumber = fillsBlock > 0 ? Math.min(blocksBlock, fillsBlock) : blocksBlock
  const rawTimestamp =
    blockNumber === fillsBlock && fillsRow ? fillsRow['ts'] : blocksRow!['timestamp']
  return { blockNumber, blockTimestamp: toUnixSeconds(rawTimestamp) }
}

/** Reads TIP-20 metadata and shapes it into the GeckoTerminal asset object. */
async function resolveAsset(
  c: Context<App.Environment>,
  chainId: z.output<typeof Schema.ChainId>,
  address: z.output<typeof Schema.TokenAddress>,
): Promise<z.output<typeof schema.Asset>> {
  const metadata = await Timing.time(c, 'coingecko_metadata', () =>
    getTokenMetadata(c, { address, chainId }),
  )
  return {
    id: Address.checksum(address),
    name: metadata.name,
    symbol: metadata.symbol,
    decimals: metadata.decimals,
    totalSupply: formatUnits(BigInt(metadata.totalSupply), metadata.decimals),
  }
}

/** A decoded `dex_pairs` row. */
type PairRow = {
  base: z.output<typeof Schema.TokenAddress>
  blockNumber?: number | undefined
  key: z.output<typeof Schema.Hash>
  quote: z.output<typeof Schema.TokenAddress>
  timestamp?: number | undefined
  transactionHash?: string | undefined
}

/**
 * Lists the newest stablecoin-DEX pairs from the pre-decoded `dex_pairs` table,
 * bounded by `limit`. Ordered newest-first (not the legacy oldest-first): the
 * full set is far too large to enumerate at Tempo API's scale (820k+ pairs), and
 * GeckoTerminal discovers pairs from `/events` rather than this list.
 */
async function listPairRows(
  c: Context<App.Environment>,
  chainId: z.output<typeof Schema.ChainId>,
  limit: number,
): Promise<PairRow[]> {
  const tidx = c.get('getTidx')(chainId)
  const result = await tidx.fetch({
    chainId,
    engine: 'clickhouse',
    query:
      `SELECT key, base, quote, tx_hash, block_num, block_timestamp FROM dex_pairs WHERE address = '${stablecoinDex}' ORDER BY block_num DESC LIMIT ${limit}` as string,
  })
  const rows: PairRow[] = []
  for (const row of result.rows) {
    const parsed = parsePairRow(row)
    if (parsed) rows.push(parsed)
  }
  return rows
}

/** Resolves a single pair's `dex_pairs` row by its order-book key. */
async function getPairRow(
  c: Context<App.Environment>,
  chainId: z.output<typeof Schema.ChainId>,
  pairId: `0x${string}`,
): Promise<PairRow | undefined> {
  const tidx = c.get('getTidx')(chainId)
  const result = await tidx.fetch({
    chainId,
    engine: 'clickhouse',
    query:
      `SELECT key, base, quote, tx_hash, block_num, block_timestamp FROM dex_pairs WHERE address = '${stablecoinDex}' AND key = '${pairId.toLowerCase()}' ORDER BY block_num ASC LIMIT 1` as string,
  })
  const row = result.rows[0]
  return row ? parsePairRow(row) : undefined
}

/** Validates and shapes one `dex_pairs` row. */
function parsePairRow(row: Record<string, unknown>): PairRow | undefined {
  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) return undefined
  const transactionHash = Schema.Hash.safeParse(row['tx_hash'])
  const blockNumber = Value.toNumber(row['block_num'])
  const timestamp = Value.toIsoDateTime(row['block_timestamp'])
  return {
    base: base.data,
    blockNumber,
    key: key.data,
    quote: quote.data,
    timestamp: timestamp ? Math.floor(new Date(timestamp).getTime() / 1_000) : undefined,
    transactionHash: transactionHash.success ? transactionHash.data : undefined,
  }
}

/**
 * Builds the GeckoTerminal pair object from pre-loaded token metadata and
 * per-book resting-order reserves. Pure (no I/O): callers load the `metadata`
 * map via {@link loadTokenMetadata} and the base-keyed `reserves` map via
 * {@link resolveBookReserves}. When a token's metadata is missing (a degraded
 * best-effort read), its reserve is reported as `'0'` rather than formatting a
 * raw amount with an unknown (defaulted-to-zero) decimals, which would be off by
 * orders of magnitude.
 */
function resolvePairDetails(
  row: PairRow,
  reserves: Map<string, resolveBookReserves.Reserve>,
  metadata: Map<string, TokenMetadata | undefined>,
): z.output<typeof schema.Pair> {
  const base = metadata.get(row.base)
  const quote = metadata.get(row.quote)

  const baseDecimals = base?.decimals ?? 0
  const quoteDecimals = quote?.decimals ?? 0

  // Reserves are keyed by lowercase base token (a book's identity); `row.base`
  // is already lowercase from the indexer.
  const reserve = reserves.get(row.base)

  return {
    id: row.key,
    dexKey,
    asset0Id: Address.checksum(row.base),
    asset1Id: Address.checksum(row.quote),
    token0: {
      address: Address.checksum(row.base),
      name: base?.name ?? '',
      symbol: base?.symbol ?? '',
      decimals: baseDecimals,
      totalSupply: base ? formatUnits(BigInt(base.totalSupply), baseDecimals) : '0',
    },
    token1: {
      address: Address.checksum(row.quote),
      name: quote?.name ?? '',
      symbol: quote?.symbol ?? '',
      decimals: quoteDecimals,
      totalSupply: quote ? formatUnits(BigInt(quote.totalSupply), quoteDecimals) : '0',
    },
    reserve0: base ? formatUnits(reserve?.base ?? 0n, baseDecimals) : '0',
    reserve1: quote ? formatUnits(reserve?.quote ?? 0n, quoteDecimals) : '0',
    ...(row.blockNumber !== undefined && { createdAtBlockNumber: row.blockNumber }),
    ...(row.timestamp !== undefined && { createdAtBlockTimestamp: row.timestamp }),
    ...(row.transactionHash !== undefined && { createdAtTxnId: row.transactionHash }),
  }
}

/** Per-token RPC metadata, keyed by lowercase address. */
type TokenMetadata = Awaited<ReturnType<typeof getTokenMetadata>>

/**
 * Reads RPC metadata for a set of tokens, returning an `address -> metadata`
 * map. Deduplicates so a quote shared across many books (e.g. pathUSD) is read
 * once per request, and reads in {@link tokenReadChunkSize} chunks with at most
 * {@link tokenReadConcurrency} chunks in flight, bounding the RPC burst. A read
 * failure maps the token to `undefined` and flips `failed` rather than failing
 * the whole response (metadata is best-effort); callers must not cache a
 * degraded page.
 */
async function loadTokenMetadata(
  c: Context<App.Environment>,
  chainId: z.output<typeof Schema.ChainId>,
  tokens: readonly string[],
): Promise<{ failed: boolean; metadata: Map<string, TokenMetadata | undefined> }> {
  const metadata = new Map<string, TokenMetadata | undefined>()
  const unique = [...new Set(tokens)]
  let failed = false
  await Timing.time(c, 'coingecko_metadata', () =>
    Batch.mapConcurrent(Batch.chunk(unique, tokenReadChunkSize), tokenReadConcurrency, (group) =>
      Promise.all(
        group.map(async (address) => {
          try {
            metadata.set(
              address,
              await getTokenMetadata(c, {
                address: address as z.output<typeof Schema.Address>,
                chainId,
              }),
            )
          } catch {
            failed = true
            metadata.set(address, undefined)
          }
        }),
      ),
    ),
  )
  return { failed, metadata }
}

/**
 * Reads the DEX's current on-chain `balanceOf` for a set of tokens, returning a
 * `token -> balance` map. Deduplicates so a quote shared across many books (e.g.
 * pathUSD) is read once per request, and reads in {@link tokenReadChunkSize}
 * chunks so the client's deployless multicall never overflows the RPC's per-call
 * limit. A read failure maps the token to `0n` and flips `failed` rather than
 * failing the whole response (reserves are best-effort); callers must not cache
 * a degraded page. See the module docstring for the `reserve1` lower-bound
 * caveat.
 */
async function loadDexBalances(
  c: Context<App.Environment>,
  chainId: z.output<typeof Schema.ChainId>,
  tokens: readonly string[],
): Promise<{ balances: Map<string, bigint>; failed: boolean }> {
  const balances = new Map<string, bigint>()
  const unique = [...new Set(tokens)]
  let failed = false
  if (unique.length === 0) return { balances, failed }
  const client = c.get('getClient')(chainId)
  await Timing.time(c, 'coingecko_reserves', () =>
    Batch.mapSeries(Batch.chunk(unique, tokenReadChunkSize), (group) =>
      Promise.all(
        group.map(async (token) => {
          try {
            const balance = await client.token.getBalance({
              account: dexAccount,
              token: token as ViemAddress,
            })
            balances.set(token, balance.amount)
          } catch {
            failed = true
            balances.set(token, 0n)
          }
        }),
      ),
    ),
  )
  return { balances, failed }
}

/**
 * Loads per-book resting-order reserves for a set of base tokens via
 * {@link resolveBookReserves}, wrapping it as a best-effort read: a scan failure
 * degrades to an empty map (zero reserves) and flips `failed` so callers don't
 * cache a degraded page, rather than failing the whole response. `scanCap`
 * bounds the newest-placement window per book; omit for the per-pair default.
 */
async function loadBookReserves(
  c: Context<App.Environment>,
  chainId: z.output<typeof Schema.ChainId>,
  bases: readonly string[],
  scanCap?: number,
): Promise<{ failed: boolean; reserves: Map<string, resolveBookReserves.Reserve> }> {
  if (bases.length === 0) return { failed: false, reserves: new Map() }
  try {
    const { reserves } = await Timing.time(c, 'coingecko_reserves', () =>
      resolveBookReserves(c, { bases, chainId, scanCap }),
    )
    return { failed: false, reserves }
  } catch {
    return { failed: true, reserves: new Map() }
  }
}

/** A raw fill row from `dex_fills`. */
type FillRow = {
  amountFilled: bigint
  blockNumber: number
  logIndex: number
  orderId: string
  taker: z.output<typeof Schema.Address>
  timestamp: number
  transactionHash: z.output<typeof Schema.Hash>
  txnIndex: number
}

/**
 * Builds the GeckoTerminal swap-event feed for a bounded block range.
 *
 * One event per `OrderFilled`. Each fill is oriented and priced from its
 * point-in-time `(token, isBid, tick)` state ({@link resolveOrderStates}, which
 * — unlike the legacy adapter's `OrderPlaced`-only lookup — also tracks flip
 * orders), with amounts base-denominated and the quote leg reconstructed from
 * the tick.
 */
async function listEvents(
  c: Context<App.Environment>,
  chainId: z.output<typeof Schema.ChainId>,
  fromBlock: number,
  toBlock: number,
): Promise<z.output<typeof schema.SwapEvent>[]> {
  const tidx = c.get('getTidx')(chainId)
  const priceScale = BigInt(Tick.priceScale)

  const fills = await Timing.time(c, 'coingecko_fills', async () => {
    const result = await tidx.fetch({
      chainId,
      engine: 'clickhouse',
      query: `
        SELECT toString("orderId") AS orderId, taker, toString("amountFilled") AS amountFilled,
               block_num, block_timestamp, tx_hash, tx_idx, log_idx
        FROM dex_fills
        WHERE address = '${stablecoinDex}' AND block_num >= ${fromBlock} AND block_num <= ${toBlock}
        ORDER BY block_num ASC, log_idx ASC
      ` as string,
    })
    const rows: FillRow[] = []
    for (const row of result.rows) {
      const parsed = parseFillRow(row)
      if (parsed) rows.push(parsed)
    }
    return rows
  })

  if (fills.length === 0) return []

  // Point-in-time order state for every filled order id. The order ids are
  // chunked: `tidx.fetch` sends queries as GET, so a single `topic1 IN (...)`
  // of every id in a wide window overflows the request URI (HTTP 414). Each
  // chunk resolves independently and the (disjoint) maps merge.
  const orderIds = [...new Set(fills.map((fill) => fill.orderId))]
  const states: Awaited<ReturnType<typeof resolveOrderStates>> = new Map()
  await Timing.time(c, 'coingecko_states', async () => {
    const partials = await Batch.mapSeries(Batch.chunk(orderIds, orderIdChunkSize), (ids) =>
      resolveOrderStates(c, {
        chainId,
        maxBlock: toBlock,
        minBlock: fromBlock,
        orderIds: ids,
      }),
    )
    for (const partial of partials) for (const [id, events] of partial) states.set(id, events)
  })

  // Resolve each book's quote token, plus its pair id (the `dex_pairs` key) so
  // event `pairId`s line up with `/pairs`. Genesis books absent from
  // `dex_pairs` fall back to the on-chain `quoteToken()` and a derived key.
  const bases = [
    ...new Set([...states.values()].flatMap((events) => events.map((event) => event.token))),
  ]
  const { keyByBase, quoteByBase } = await Timing.time(c, 'coingecko_pair_index', () =>
    resolvePairIndex(c, chainId, bases),
  )

  // Token decimals (RPC metadata) and best-effort current reserves (RPC
  // `balanceOf`) for every base/quote involved. Loaded in bounded chunks and
  // sequentially (not `Promise.all`) so the two read kinds don't merge into the
  // same multicall window and overflow the RPC's per-call limit. A degraded read
  // drops the affected events below and must not be cached as authoritative.
  const tokens = [...new Set([...bases, ...quoteByBase.values()])]
  const metadata = await loadTokenMetadata(c, chainId, tokens)
  const reserves = await loadDexBalances(c, chainId, tokens)
  if (metadata.failed || reserves.failed) Cache.setPolicy(c, Cache.policies.noStore)
  const balances = reserves.balances
  const decimals = new Map<string, number>()
  for (const [token, info] of metadata.metadata) if (info) decimals.set(token, info.decimals)

  const events: z.output<typeof schema.SwapEvent>[] = []
  const eventIndexByTx = new Map<string, number>()

  for (const fill of fills) {
    const state = latestStateBefore(states.get(fill.orderId), fill.blockNumber, fill.logIndex)
    if (!state) continue

    const quote = quoteByBase.get(state.token)
    if (!quote) continue

    const baseDecimals = decimals.get(state.token)
    const quoteDecimals = decimals.get(quote)
    if (baseDecimals === undefined || quoteDecimals === undefined) continue

    const pairId = keyByBase.get(state.token) ?? deriveBookKey(state.token, quote)

    const tickPrice = priceScale + BigInt(state.tick)
    const baseAmount = fill.amountFilled
    const quoteAmount = (baseAmount * tickPrice) / priceScale

    const baseDec = formatUnits(baseAmount, baseDecimals)
    const quoteDec = formatUnits(quoteAmount, quoteDecimals)

    // Maker bid: taker sells base into the book (base in, quote out).
    // Maker ask: taker buys base out of the book (quote in, base out).
    const sides = state.isBid
      ? { asset0In: baseDec, asset1Out: quoteDec }
      : { asset0Out: baseDec, asset1In: quoteDec }

    const baseReserve = balances.get(state.token) ?? 0n
    const quoteReserve = balances.get(quote) ?? 0n

    const eventIndex = eventIndexByTx.get(fill.transactionHash) ?? 0
    eventIndexByTx.set(fill.transactionHash, eventIndex + 1)

    events.push({
      block: { blockNumber: fill.blockNumber, blockTimestamp: fill.timestamp },
      eventType: 'swap',
      txnId: fill.transactionHash,
      txnIndex: fill.txnIndex,
      eventIndex,
      maker: Address.checksum(fill.taker),
      pairId,
      ...sides,
      priceNative: computePriceNative(tickPrice, baseDecimals, quoteDecimals, priceScale),
      reserves: {
        asset0: formatUnits(baseReserve, baseDecimals),
        asset1: formatUnits(quoteReserve, quoteDecimals),
      },
    })
  }

  return events
}

/** Parses one `dex_fills` row, dropping malformed rows. */
function parseFillRow(row: Record<string, unknown>): FillRow | undefined {
  const orderId = Value.toIntegerString(row['orderId'])
  const taker = Schema.Address.safeParse(row['taker'])
  const amountFilled = Value.toIntegerString(row['amountFilled'])
  const blockNumber = Value.toNumber(row['block_num'])
  const logIndex = Value.toNumber(row['log_idx'])
  const txnIndex = Value.toNumber(row['tx_idx'])
  const transactionHash = Schema.Hash.safeParse(row['tx_hash'])
  const timestamp = Value.toIsoDateTime(row['block_timestamp'])
  if (
    orderId === undefined ||
    !taker.success ||
    amountFilled === undefined ||
    blockNumber === undefined ||
    logIndex === undefined ||
    txnIndex === undefined ||
    !transactionHash.success ||
    timestamp === undefined
  )
    return undefined
  return {
    amountFilled: BigInt(amountFilled),
    blockNumber,
    logIndex,
    orderId,
    taker: taker.data,
    timestamp: Math.floor(new Date(timestamp).getTime() / 1_000),
    transactionHash: transactionHash.data,
    txnIndex,
  }
}

/**
 * Resolves, for a set of base tokens, each book's quote token and `dex_pairs`
 * key. Indexed pairs come from `dex_pairs`; the rest fall back to the on-chain
 * `quoteToken()` ({@link resolveBookQuotes}) with a key derived from
 * `keccak256(base ++ quote)`.
 */
async function resolvePairIndex(
  c: Context<App.Environment>,
  chainId: z.output<typeof Schema.ChainId>,
  bases: readonly string[],
): Promise<{
  keyByBase: Map<string, z.output<typeof Schema.Hash>>
  quoteByBase: Map<string, z.output<typeof Schema.TokenAddress>>
}> {
  const keyByBase = new Map<string, z.output<typeof Schema.Hash>>()
  const quoteByBase = new Map<string, z.output<typeof Schema.TokenAddress>>()
  if (bases.length === 0) return { keyByBase, quoteByBase }

  // Chunk the `base IN (...)` lookup for the same GET-URI reason as the order
  // states above; chunks resolve serially and merge.
  const tidx = c.get('getTidx')(chainId)
  const results = await Batch.mapSeries(Batch.chunk(bases, addressChunkSize), (group) => {
    const list = group.map((base) => `'${base}'`).join(', ')
    return tidx.fetch({
      chainId,
      engine: 'clickhouse',
      query:
        `SELECT key, base, quote FROM dex_pairs WHERE address = '${stablecoinDex}' AND base IN (${list})` as string,
    })
  })
  for (const result of results)
    for (const row of result.rows) {
      const parsed = parsePairRow(row)
      if (!parsed) continue
      keyByBase.set(parsed.base, parsed.key)
      quoteByBase.set(parsed.base, parsed.quote)
    }

  const missing = bases.filter((base) => !quoteByBase.has(base))
  const fallback = await resolveBookQuotes(c, { bases: missing, chainId })
  for (const [base, quote] of fallback) quoteByBase.set(base, quote)

  return { keyByBase, quoteByBase }
}

/** Latest order state strictly before a fill's `(blockNumber, logIndex)`. */
function latestStateBefore(
  events:
    | { blockNumber: number; isBid: boolean; logIndex: number; tick: number; token: string }[]
    | undefined,
  blockNumber: number,
  logIndex: number,
) {
  if (!events) return undefined
  for (let i = events.length - 1; i >= 0; i--) {
    const event = events[i]!
    if (
      event.blockNumber < blockNumber ||
      (event.blockNumber === blockNumber && event.logIndex < logIndex)
    )
      return event
  }
  return undefined
}

namespace Batch {
  /** Splits `items` into consecutive groups of at most `size`. */
  export function chunk<item>(items: readonly item[], size: number): item[][] {
    const groups: item[][] = []
    for (let i = 0; i < items.length; i += size) groups.push(items.slice(i, i + size))
    return groups
  }

  /**
   * Maps `fn` over `items` strictly one at a time. ClickHouse chunk queries are
   * issued serially on purpose: the managed indexer rejects bursts of concurrent
   * queries with a fast HTTP 422 (verified: heavy baseline scans 422 at 2-3 in
   * flight), and each chunk is individually fast (~2 s), so serial execution
   * trades a little latency for reliability. Real GeckoTerminal polling windows
   * are small and cached, so the latency is rarely on the hot path.
   */
  export async function mapSeries<item, result>(
    items: readonly item[],
    fn: (item: item) => Promise<result>,
  ): Promise<result[]> {
    const results: result[] = []
    for (const item of items) results.push(await fn(item))
    return results
  }

  /**
   * Maps `fn` over `items` with at most `concurrency` calls in flight, preserving
   * result order. For RPC chunk fan-outs only: ClickHouse chunk queries stay
   * serial through {@link mapSeries}.
   */
  export async function mapConcurrent<item, result>(
    items: readonly item[],
    concurrency: number,
    fn: (item: item) => Promise<result>,
  ): Promise<result[]> {
    const results: result[] = Array.from({ length: items.length })
    let next = 0
    const workers = Array.from({ length: Math.min(concurrency, items.length) }, async () => {
      while (next < items.length) {
        const index = next++
        results[index] = await fn(items[index]!)
      }
    })
    await Promise.all(workers)
    return results
  }
}

/** `keccak256(base ++ quote)` — the on-chain order-book key. */
function deriveBookKey(base: string, quote: string): z.output<typeof Schema.Hash> {
  return Hash.keccak256(`0x${base.slice(2)}${quote.slice(2)}` as `0x${string}`)
}

/**
 * Formats `priceNative` as quote-per-base at 36-dp precision, matching the
 * legacy adapter: `tickPrice * 10^(baseDecimals + 36) / (scale * 10^quoteDecimals)`.
 */
function computePriceNative(
  tickPrice: bigint,
  baseDecimals: number,
  quoteDecimals: number,
  scale: bigint,
): string {
  const priceBig =
    (tickPrice * 10n ** (BigInt(baseDecimals) + pricePrecision)) /
    (scale * 10n ** BigInt(quoteDecimals))
  return formatUnits(priceBig, Number(pricePrecision))
}

/** Coerces a TIDX timestamp (Unix seconds/ms or ISO/CH text) to Unix seconds. */
function toUnixSeconds(value: unknown): number {
  const iso = Value.toIsoDateTime(value)
  if (iso) return Math.floor(new Date(iso).getTime() / 1_000)
  const num = Value.toNumber(value)
  if (num === undefined) return 0
  return num > 1e12 ? Math.floor(num / 1_000) : num
}