import { Hono, type Context } from 'hono'
import { AbiEvent, AbiParameters, Hash, type Hex } from 'ox'
import { Abis, Addresses } 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 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 * as Tokens from './tokens.js'

// The protocol Fee AMM lives at the reserved FeeManager address. Pools are
// `(userToken, validatorToken)` pairs discovered from the `Mint` events the
// FeeManager emits; there is no on-chain pool registry to enumerate.
const feeManager = Addresses.feeManager.toLowerCase()

// pathUSD is the protocol's default fee token: a transaction whose fee token
// matches the validator's token (or falls back to pathUSD) pays fees without
// needing a Fee AMM pool, so pathUSD is always fee-usable even when it has no
// user-side pool. Lowercased to match the indexer's stored address casing.
const pathUsd = Addresses.pathUsd.toLowerCase()

// Topic0 hashes of the FeeManager `Mint` event, current and legacy. Both
// signatures index `userToken` as topic2 and `validatorToken` as topic3, but
// their other fields differ, so rows are decoded per selector: the current
// event carries data words `[sender, amountValidatorToken, liquidity]` with
// topic1 = `to`; the legacy event carries `[amountUserToken,
// amountValidatorToken, liquidity]` with topic1 = `sender`. Queries hit the
// raw `logs` table rather than a signature CTE: the event-CTE planner rejects
// filters on event params, and the two signatures need different decoders
// anyway.
const mintTopic = AbiEvent.getSelector(
  'event Mint(address sender, address indexed to, address indexed userToken, address indexed validatorToken, uint256 amountValidatorToken, uint256 liquidity)',
)
const legacyMintTopic = AbiEvent.getSelector(
  'event Mint(address indexed sender, address indexed userToken, address indexed validatorToken, uint256 amountUserToken, uint256 amountValidatorToken, uint256 liquidity)',
)

// Block windows for the mint feed's page sweep, widened geometrically until
// the page fills. An unbounded top-K over the FeeManager's mint history sorts
// every matching row (tens of millions on busy chains) and crosses TIDX's 5s
// execution cap, so the page query is bounded to a block window anchored at
// the cursor (or the feed's head/tail block); dense filters fill the first
// window, and only sparse filters, whose top-K is inherently cheap, fall
// through to the unbounded query.
const mintWindowBlocks = [100_000, 1_000_000, 10_000_000]

/** Zod schemas owned by the fee AMM handlers. */
export namespace schema {
  /**
   * A token referenced by a fee pool: its contract address plus best-effort
   * metadata. Fee AMM mints use this shape; pools require core metadata through
   * `getFeeAmmPools.Token`.
   */
  export const PoolToken = Schema.describe(
    z.object({
      address: Schema.tokenAddress(Tokens.tokenExample.address).check(
        z.describe('TIP-20 token contract address used by this fee pool.'),
      ),
      currency: z
        .optional(z.string())
        .check(
          z.describe('Display currency for the token, when available.'),
          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.'),
          z.meta({ examples: [Tokens.tokenExample.decimals] }),
        ),
      logoUri: z.optional(z.string()).check(
        z.describe('URL for the token logo image, when available.'),
        z.meta({
          examples: [Tokens.tokenExample.logoUri],
        }),
      ),
      name: z
        .optional(z.string())
        .check(
          z.describe('Human-readable token name.'),
          z.meta({ examples: [Tokens.tokenExample.name] }),
        ),
      symbol: z
        .optional(z.string())
        .check(
          z.describe('Short token ticker symbol.'),
          z.meta({ examples: [Tokens.tokenExample.symbol] }),
        ),
      verified: z
        .optional(z.boolean())
        .check(
          z.describe('Whether Tempo has verified this token’s metadata.'),
          z.meta({ examples: [true] }),
        ),
    }),
    'A TIP-20 token used by a Fee AMM pool, with metadata when available.',
  )

  /** Schemas for the getFeeAmmPools operation. */
  export namespace getFeeAmmPools {
    /** Optional curated token fields that callers opt into via `include`. */
    export const Include = z
      .enum(['token.logoUri', 'token.verified'])
      .check(z.describe('Additional token fields to include in the pool response.'))

    /**
     * Parses comma-separated curated token fields. Core token metadata is
     * always present; callers opt into logo and verification fields.
     */
    export const includeQuery = Schema.includeQuery(
      Include,
      'Comma-separated token fields to include, such as `token.logoUri,token.verified`.',
    )

    /** Query parameters for the fee pool list request. */
    export const Query = z
      .strictObject({
        chainId: Schema.ChainIdQuery,
        cursor: Schema.Cursor,
        include: includeQuery,
        limit: Schema.Limit,
        page: Schema.Page,
      })
      .check(...Schema.pageChecks(), z.describe('Query parameters for listing Fee AMM pools.'))

    /** A fee-pool token with required RPC metadata and optional curated fields. */
    export const Token = Schema.describe(
      Tokens.schema.TokenReference,
      'A token referenced by one side of a Fee AMM pool.',
    )

    /** A fee AMM pool, aggregated from its `Mint` events. */
    export const Pool = z
      .object({
        createdAt: z.iso
          .datetime()
          .check(
            z.describe('Timestamp when liquidity was first minted into this pool.'),
            z.meta({ examples: ['2024-01-01T00:00:00Z'] }),
          ),
        id: Schema.Hash.check(
          z.describe('Stable resource id for the pool; this is the same value as `poolId`.'),
        ),
        lastMintAt: z.iso
          .datetime()
          .check(
            z.describe('Timestamp when liquidity was most recently minted into this pool.'),
            z.meta({ examples: ['2024-01-01T00:00:00Z'] }),
          ),
        mintCount: z
          .number()
          .check(
            z.int(),
            z.nonnegative(),
            z.describe('Number of liquidity mints into this pool.'),
            z.meta({ examples: [25] }),
          ),
        poolId: Schema.Hash.check(
          z.describe(
            'Pool id computed as `keccak256(abi.encode(userToken, validatorToken))`, matching `FeeManager.getPoolId`.',
          ),
        ),
        userAmount: z
          .optional(Schema.TokenAmount)
          .check(z.describe('Current user-token reserve. Omitted when the onchain read fails.')),
        userToken: Token.check(z.describe('Token users pay fees with in this pool.')),
        validatorAmount: z
          .optional(Schema.TokenAmount)
          .check(
            z.describe('Current validator-token reserve. Omitted when the onchain read fails.'),
          ),
        validatorToken: Token.check(z.describe('Token validators receive from this pool.')),
      })
      .check(z.describe('A Fee AMM pool discovered from its onchain `Mint` events.'))

    /** Page of fee pools, ordered by mint count (most active first). */
    export const Response = z
      .object({
        data: z
          .array(Pool)
          .check(z.describe('Fee AMM pools ordered by mint count, most active first.')),
        nextCursor: Schema.NextCursor,
      })
      .check(z.describe('A page of Fee AMM pools ordered by mint count, most active first.'))
  }

  /** Schemas for the getFeeAmmMints operation. */
  export namespace getFeeAmmMints {
    /** Query parameters for the pool mint list request. */
    export const Query = z
      .strictObject({
        chainId: Schema.ChainIdQuery,
        cursor: Schema.Cursor,
        include: Schema.totalCountInclude,
        limit: Schema.Limit,
        order: Schema.Order,
        page: Schema.Page,
        userToken: z
          .optional(Schema.TokenAddress)
          .check(z.describe('Only include mints for pools where users paid fees with this token.')),
        validatorToken: z
          .optional(Schema.TokenAddress)
          .check(z.describe('Only include mints for pools where validators receive this token.')),
      })
      .check(
        ...Schema.pageChecks(),
        z.describe('Query parameters for listing Fee AMM liquidity mints.'),
      )

    /** A single liquidity mint into a fee pool. */
    export const Mint = z
      .object({
        amountUserToken: z
          .optional(Schema.DecimalString)
          .check(
            z.describe(
              'Amount of the user-side token deposited, in base units. Only present on legacy-signature mints.',
            ),
          ),
        amountValidatorToken: Schema.DecimalString.check(
          z.describe('Amount of the validator-side token deposited, in base units.'),
        ),
        blockNumber: z
          .number()
          .check(
            z.int(),
            z.nonnegative(),
            z.describe('Block number where the mint occurred.'),
            z.meta({ examples: [23456789] }),
          ),
        id: z.string().check(
          z.describe(
            'Stable mint id built from the transaction hash and log index (`${transactionHash}-${logIndex}`).',
          ),
          z.meta({
            examples: ['0x515801d7f9a5ac705e793e85904c9c69b3f1694b465cc8fb6ba3f0298dc82665-0'],
          }),
        ),
        liquidity: Schema.DecimalString.check(
          z.describe('Amount of pool liquidity minted, as an integer string.'),
        ),
        logIndex: z
          .number()
          .check(
            z.int(),
            z.nonnegative(),
            z.describe('Log index of this mint within the block.'),
            z.meta({ examples: [0] }),
          ),
        minter: Schema.Address.check(z.describe('Address that deposited tokens into the pool.')),
        recipient: z
          .optional(Schema.Address)
          .check(
            z.describe(
              'Address that received the minted liquidity. Only present on current-signature mints.',
            ),
          ),
        timestamp: z.iso
          .datetime()
          .check(
            z.describe('Block timestamp when the mint occurred.'),
            z.meta({ examples: ['2024-01-01T00:00:00Z'] }),
          ),
        transactionHash: Schema.Hash.check(
          z.describe('Hash of the transaction that emitted this mint.'),
        ),
        userToken: PoolToken.check(z.describe('Token users pay fees with in this pool.')),
        validatorToken: PoolToken.check(z.describe('Token validators receive from this pool.')),
      })
      .check(z.describe('One liquidity mint into a Fee AMM pool.'))

    /** Page of fee pool mints, ordered by block then log index. */
    export const Response = z
      .object({
        data: z.array(Mint).check(z.describe('Liquidity mints returned on this page.')),
        meta: z
          .optional(Schema.CountMeta)
          .check(
            z.describe('Response-wide metadata requested with `include`, such as `totalCount`.'),
          ),
        nextCursor: Schema.NextCursor,
      })
      .check(z.describe('A page of Fee AMM liquidity mints.'))
  }
}

/** Creates fee AMM handlers. */
export function feeAmm() {
  return new Hono<App.Environment>()
    .get(
      '/v1/fee-amm/pools',
      Auth.policy({ apiKey: { scopes: ['data:read'] }, mpp: true, public: true }),
      OpenApi.validate('query', schema.getFeeAmmPools.Query, {
        code: 'query_invalid',
        message: 'Invalid query parameters',
      }),
      OpenApi.describeRoute({
        description:
          'Lists Fee AMM pools, most active first, so you can see which stablecoin fee conversions have the most liquidity activity.',
        operationId: 'getFeeAmmPools',
        responses: OpenApi.responses({
          errors: { 502: 'Tempo RPC or the upstream indexer could not complete the request.' },
          success: {
            description: 'List of Fee AMM pools.',
            schema: schema.getFeeAmmPools.Response,
          },
        }),
        summary: 'List pools',
        tags: ['Fee AMM'],
      }),
      Cache.response<App.Environment>({
        cacheControl: Cache.policies.state,
        name: 'tempo-api:fee-amm-pools:v2',
        key: (c) => Cache.urlKey(c, schema.getFeeAmmPools.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 query = c.req.valid('query')
        const chainId = query.chainId ?? c.get('chainId')

        try {
          const page = await listPools(c, {
            chainId,
            cursor: query.cursor,
            limit: query.limit,
            page: query.page,
          })
          const pools = page.data

          // Reserves and token metadata are independent batched RPC reads.
          const [reserves, tokensByAddress] = await Promise.all([
            Timing.time(c, 'fee_amm_reserves', () =>
              Promise.all(
                pools.map((pool) =>
                  getPoolReserves(c, {
                    chainId,
                    userToken: pool.userToken,
                    validatorToken: pool.validatorToken,
                  }),
                ),
              ),
            ),
            resolveTokenMap(c, {
              addresses: Array.from(
                new Set(pools.flatMap((pool) => [pool.userToken, pool.validatorToken])),
              ),
              chainId,
              include: query.include,
            }),
          ])

          const token = (address: z.output<typeof Schema.Address>) => {
            const metadata = tokensByAddress.get(address)
            if (!metadata) throw new Error(`Token metadata unavailable for ${address}`)
            return {
              address,
              currency: metadata.currency,
              decimals: metadata.decimals,
              ...(query.include.includes('token.logoUri') && metadata.logoUri !== undefined
                ? { logoUri: metadata.logoUri }
                : {}),
              name: metadata.name,
              symbol: metadata.symbol,
              ...(query.include.includes('token.verified') && metadata.verified !== undefined
                ? { verified: metadata.verified }
                : {}),
            }
          }

          const data = pools.map((pool, index) => {
            const id = poolId(pool.userToken, pool.validatorToken)
            const userToken = token(pool.userToken)
            const validatorToken = token(pool.validatorToken)
            return {
              createdAt: pool.createdAt,
              id,
              lastMintAt: pool.lastMintAt,
              mintCount: pool.mintCount,
              poolId: id,
              userAmount: reserve({ amount: reserves[index]?.userToken, token: userToken }),
              userToken,
              validatorAmount: reserve({
                amount: reserves[index]?.validatorToken,
                token: validatorToken,
              }),
              validatorToken,
            }
          })

          return c.json(
            Response.validated(schema.getFeeAmmPools.Response, {
              data,
              nextCursor: page.nextCursor,
            }),
            200,
          )
        } catch (cause) {
          return Response.upstream(c, cause, { page: query.page })
        }
      },
    )
    .get(
      '/v1/fee-amm/mints',
      Auth.policy({ apiKey: { scopes: ['data:read'] }, mpp: true, public: true }),
      OpenApi.validate('query', schema.getFeeAmmMints.Query, {
        code: 'query_invalid',
        message: 'Invalid query parameters',
      }),
      OpenApi.describeRoute({
        description: 'Lists liquidity mints into Fee AMM pools.',
        operationId: 'getFeeAmmMints',
        responses: OpenApi.responses({
          errors: { 502: 'The upstream Tempo indexer could not complete the request.' },
          success: {
            description: 'Page of Fee AMM liquidity mints.',
            schema: schema.getFeeAmmMints.Response,
          },
        }),
        summary: 'List mints',
        tags: ['Fee AMM'],
      }),
      Cache.response<App.Environment>({
        cacheControl: Cache.policies.feed,
        name: 'tempo-api:fee-amm:v1',
        key: (c) => Cache.urlKey(c, schema.getFeeAmmMints.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 query = c.req.valid('query')
        const chainId = query.chainId ?? c.get('chainId')

        // A cursor page is anchored below the head, so its contents are
        // append-only and effectively immutable; cache it aggressively. The
        // head page (no cursor) keeps the route's short `feed` default.
        if (query.cursor !== undefined) Cache.setPolicy(c, Cache.policies.immutable)

        try {
          // `totalCount` is opt-in and shares the page's filters; run it
          // concurrently with the page. Best-effort: a count failure omits
          // `meta` rather than failing the page.
          const countPromise = query.include.includes('totalCount')
            ? countMints(c, {
                chainId,
                userToken: query.userToken,
                validatorToken: query.validatorToken,
              }).catch(() => undefined)
            : undefined

          const page = await listMints(c, {
            chainId,
            cursor: query.cursor,
            limit: query.limit,
            order: query.order,
            page: query.page,
            userToken: query.userToken,
            validatorToken: query.validatorToken,
          })

          // Token references take the same enriched shape as the pools
          // listing. The page's unique tokens are few (often one pool), and
          // each resolution is memoized, so this stays one bounded fan-out.
          const tokensByAddress = await resolveTokenMap(c, {
            addresses: Array.from(
              new Set(page.data.flatMap((mint) => [mint.userToken, mint.validatorToken])),
            ),
            chainId,
          })
          const data = page.data.map((mint) => ({
            ...mint,
            userToken: { address: mint.userToken, ...tokensByAddress.get(mint.userToken) },
            validatorToken: {
              address: mint.validatorToken,
              ...tokensByAddress.get(mint.validatorToken),
            },
          }))

          const meta = countPromise ? await countPromise : undefined
          return c.json(
            Response.validated(schema.getFeeAmmMints.Response, {
              data,
              ...(meta ? { meta } : {}),
              nextCursor: page.nextCursor,
            }),
            200,
          )
        } catch (cause) {
          return Response.upstream(c, cause, { page: query.page })
        }
      },
    )
}

/**
 * Discovers fee pools by aggregating the FeeManager's `Mint` logs into one
 * row per `(userToken, validatorToken)` pair, paginated and ordered by mint
 * count (most active first), so the real fee pools (thousands to millions of
 * mints) lead the head pages and the spam pairs (mint count 1) sink to the
 * tail.
 *
 * Keyset pagination on `(mint_count, topic2, topic3)`: `mint_count` ranks the
 * page (mutable — it grows as pools mint — so this is best-effort across mint
 * activity, the inherent limit of a ranking feed), and the `(topic2, topic3)`
 * group key is a stable, unique tiebreaker. A malformed cursor falls back to
 * the head page.
 */
function listPools(c: Context<App.Environment>, options: listPools.Options) {
  const { chainId, limit } = options
  const store = c.get('store')
  const tidx = c.get('getTidx')(chainId)
  const cursor = options.cursor
    ? Cursor.decode(options.cursor, ['int', 'address', '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

  return Timing.time(c, 'fee_amm_pools', () =>
    Store.memoize(
      async () => {
        // The keyset compares the aggregate `mint_count`, so it lives in the
        // outer WHERE over the GROUP BY subquery. Topics are padded 32-byte
        // log topics, so render their literals via `padTopic` (lowercased to
        // match ClickHouse's stored topic casing) rather than `Cursor.literal`.
        const keyset =
          cursor !== undefined
            ? ` WHERE ${Cursor.keyset([
                { literal: Cursor.literal(cursor[0]!, 'int'), name: 'mint_count', order: 'desc' },
                {
                  literal: `'${padTopic(cursor[1] as z.output<typeof Schema.Address>).toLowerCase()}'`,
                  name: 'topic2',
                  order: 'desc',
                },
                {
                  literal: `'${padTopic(cursor[2] as z.output<typeof Schema.Address>).toLowerCase()}'`,
                  name: 'topic3',
                  order: 'desc',
                },
              ])}`
            : ''

        // Raw-`logs` ClickHouse aggregate (dynamic query, hence the cast):
        // ordering is by mint count so the head pages surface the real fee
        // pools and the single-mint spam pairs sink to the tail. Fetch one extra
        // row to detect `hasMore` without a separate count query.
        const result = await tidx.fetch({
          chainId,
          engine: 'clickhouse',
          query: `
            SELECT topic2, topic3, mint_count, first_at, last_at
            FROM (
              SELECT topic2, topic3, count() AS mint_count, min(block_timestamp) AS first_at, max(block_timestamp) AS last_at
              FROM logs
              WHERE address = '${feeManager}' AND selector IN ('${mintTopic}', '${legacyMintTopic}')
              GROUP BY topic2, topic3
            )${keyset}
            ORDER BY mint_count DESC, topic2 DESC, topic3 DESC
            LIMIT ${limit + 1}${offset !== undefined ? ` OFFSET ${offset}` : ''}
          ` as string,
        })

        // The next page anchors below the last row's `(mint_count, topic2,
        // topic3)`; topics use the lowercase unpadded address so they round
        // trip through the keyset literals above.
        const page = Cursor.paginate({
          rows: result.rows,
          limit,
          key: (row) => {
            const mintCount = Value.toNumber(row['mint_count'])
            const userToken = unpadTopic(Value.toText(row['topic2']))?.toLowerCase()
            const validatorToken = unpadTopic(Value.toText(row['topic3']))?.toLowerCase()
            return mintCount !== undefined &&
              userToken !== undefined &&
              validatorToken !== undefined
              ? [mintCount, userToken, validatorToken]
              : undefined
          },
        })

        const data: listPools.Pool[] = []
        for (const row of page.rows) {
          const userToken = Schema.Address.safeParse(unpadTopic(Value.toText(row['topic2'])))
          const validatorToken = Schema.Address.safeParse(unpadTopic(Value.toText(row['topic3'])))
          const mintCount = Value.toNumber(row['mint_count'])
          const createdAt = Value.toIsoDateTime(row['first_at'])
          const lastMintAt = Value.toIsoDateTime(row['last_at'])
          if (
            !userToken.success ||
            !validatorToken.success ||
            mintCount === undefined ||
            createdAt === undefined ||
            lastMintAt === undefined
          )
            continue
          data.push({
            createdAt,
            lastMintAt,
            mintCount,
            userToken: userToken.data,
            validatorToken: validatorToken.data,
          })
        }

        return { data, nextCursor: page.nextCursor }
      },
      {
        key: `feeamm:v1:${chainId}:pools:${cursor ? `cursor:${cursor[0]}:${cursor[1]}:${cursor[2]}` : offset !== undefined ? `page:${options.page}` : 'head'}:${limit}`,
        store,
        ttl: Ttl.minutes(1),
      },
    ),
  )
}

declare namespace listPools {
  /** Options for {@link listPools}. */
  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
  }
  /** A discovered pool, aggregated from its `Mint` logs. */
  type Pool = {
    /** ISO timestamp of the pool's first mint. */
    createdAt: string
    /** ISO timestamp of the pool's most recent mint. */
    lastMintAt: string
    mintCount: number
    userToken: z.output<typeof Schema.Address>
    validatorToken: z.output<typeof Schema.Address>
  }
}

/**
 * The set of token addresses usable to pay transaction fees on a chain: the
 * curated verified tokens that have at least one Fee AMM pool (i.e. appear as a
 * `userToken` / topic2 in the FeeManager's `Mint` logs), plus the protocol
 * default fee token (`pathUSD`), which pays fees without needing a pool.
 * Addresses are lowercased to match the indexer's stored casing.
 *
 * The set is intentionally gated to the verified list rather than every pooled
 * token: anyone can mint a single-mint spam pool, so the raw set of pool
 * `userToken`s is unbounded (tens of thousands on busy chains) and those
 * liquidity-less pools can't actually settle fees. Restricting to verified
 * tokens keeps the set small (a handful of USD stablecoins), bounds the scan
 * (the `topic2 IN (…)` candidate list is the verified list), and matches the
 * tokens fees are realistically paid in.
 *
 * Memoized per chain and verified-snapshot version (the only inputs), so
 * callers — e.g. the address balances endpoint — can flag which holdings are
 * fee-usable without a second `/fee-amm/pools` query.
 */
export async function feeTokenSet(
  c: Context<App.Environment>,
  chainId: z.output<typeof Schema.ChainId>,
): Promise<Set<string>> {
  const store = c.get('store')
  const tidx = c.get('getTidx')(chainId)
  const snapshot = await VerifiedTokens.snapshot(c, chainId)
  // pathUSD is always fee-usable; the verified tokens are the only other
  // candidates we probe for a pool. With no verified candidates there is
  // nothing to query — pathUSD stands alone.
  const candidates = snapshot.list.map((token) => token.address.toLowerCase())
  if (candidates.length === 0) return new Set([pathUsd])

  // `memoize` JSON-encodes its value, so cache a plain `string[]` and rebuild
  // the `Set` per call (cheap; the list is a handful of stablecoins).
  const addresses = await Timing.time(c, 'fee_token_set', () =>
    Store.memoize(
      async () => {
        const tokens = new Set<string>([pathUsd])
        // PostgreSQL's `topic2` index proves negative lookups without scanning
        // FeeManager history. Probe serially so an uncapped verified list
        // cannot fan out indexer load.
        for (const address of candidates) {
          const topic = padTopic(address as `0x${string}`)
          const result = await tidx.fetch({
            chainId,
            engine: 'postgres',
            query: `
              SELECT 1 AS present
              FROM logs
              WHERE address = '${feeManager}' AND selector IN ('${mintTopic}', '${legacyMintTopic}')
                AND topic2 = '${topic}'
              LIMIT 1
            ` as string,
          })
          if (result.rows.length > 0) tokens.add(address)
        }
        return Array.from(tokens)
      },
      {
        key: `feeamm:v2:${chainId}:fee-token-set:v:${snapshot.version}`,
        store,
        ttl: Ttl.minutes(5),
      },
    ),
  )
  return new Set(addresses)
}

/**
 * Reads a pool's current on-chain reserves via `FeeManager.getPool`, memoized
 * briefly per pool. Reserves are best-effort decoration on the pool listing:
 * a failing RPC read resolves to `null` (degrading that pool's reserve
 * fields) rather than failing the whole response. Failures are not memoized,
 * so the next request retries.
 */
function getPoolReserves(
  c: Context<App.Environment>,
  options: getPoolReserves.Options,
): Promise<getPoolReserves.Reserves | null> {
  const { chainId, userToken, validatorToken } = options
  const client = c.get('getClient')(chainId)
  const store = c.get('store')

  return Store.memoize(
    async () => {
      const pool = await client.readContract({
        abi: Abis.feeAmm,
        address: Addresses.feeManager,
        args: [userToken, validatorToken],
        functionName: 'getPool',
      })
      return {
        userToken: pool.reserveUserToken.toString(),
        validatorToken: pool.reserveValidatorToken.toString(),
      }
    },
    {
      key: `feeamm:v1:${chainId}:pool:${userToken}:${validatorToken}:reserves`,
      store,
      ttl: Ttl.seconds(30),
    },
  ).catch(() => null)
}

declare namespace getPoolReserves {
  /** Options for {@link getPoolReserves}. */
  type Options = {
    chainId: z.output<typeof Schema.ChainId>
    userToken: z.output<typeof Schema.Address>
    validatorToken: z.output<typeof Schema.Address>
  }
  /** Raw base-unit reserves of both pool sides. */
  type Reserves = {
    /** User-side token reserve in its base unit. */
    userToken: string
    /** Validator-side token reserve in its base unit. */
    validatorToken: string
  }
}

/**
 * Lists `Mint` logs as a keyset-paginated page on `(block_num, log_idx)`,
 * optionally filtered by the pool's tokens. Runs on ClickHouse: the postgres
 * `logs` table has no topic index, so topic-filtered pages exceed its
 * statement timeout, while ClickHouse serves every filter shape. The page
 * query sweeps a widening block window (see `mintWindowBlocks`) so large
 * chains never top-K sort the full mint history.
 */
function listMints(c: Context<App.Environment>, options: listMints.Options) {
  const { chainId, limit, order, userToken, validatorToken } = options
  const store = c.get('store')
  const tidx = c.get('getTidx')(chainId)
  const direction = order === 'asc' ? 'ASC' : 'DESC'
  // Keyset pagination: anchor the page below the previous row's
  // `(block, log_idx)` position instead of a numeric offset, so rows arriving
  // at the head can't shift items across pages. 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, 'fee_amm_mints', () =>
    Store.memoize(
      async () => {
        const filters = [
          `address = '${feeManager}'`,
          `selector IN ('${mintTopic}', '${legacyMintTopic}')`,
        ]
        if (userToken !== undefined) filters.push(`topic2 = '${padTopic(userToken)}'`)
        if (validatorToken !== undefined) filters.push(`topic3 = '${padTopic(validatorToken)}'`)
        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 },
            ]),
          )

        // Fetch one extra row to detect `hasMore` without a separate count
        // query. Raw-`logs` dynamic query, hence the cast. `block_timestamp`
        // is aliased to a non-standard name on purpose: the TIDX client
        // coerces standard `logs` columns, and its coercion parses
        // ClickHouse's timezone-less `DateTime` text as *local* time. The
        // alias passes the raw text through so `Value.toIsoDateTime` can
        // apply its UTC normalization instead.
        const fetchPage = (where: readonly string[]) =>
          tidx.fetch({
            chainId,
            engine: 'clickhouse',
            query: `
              SELECT tx_hash, block_num, log_idx, block_timestamp AS block_time, selector, topic1, topic2, topic3, data
              FROM logs
              WHERE ${where.join(' AND ')}
              ORDER BY block_num ${direction}, log_idx ${direction}
              LIMIT ${limit + 1}${offset !== undefined ? ` OFFSET ${offset}` : ''}
            ` as string,
          })

        // Window anchor: the cursor block when paginating, else the fee
        // manager's newest (desc) or oldest (asc) log block. The bounds are
        // shared by every filter shape and staleness only shifts the first
        // window, so memoize them briefly per chain.
        const anchor = await (async () => {
          if (cursor !== undefined) return Number(cursor[0])
          const bounds = await Store.memoize(
            async () => {
              // Dynamic raw-`logs` query, hence the cast. `address` aligns
              // with the `logs` sort key, so the probe is index-pruned.
              const result = await tidx.fetch({
                chainId,
                engine: 'clickhouse',
                query:
                  `SELECT min(block_num) AS lo, max(block_num) AS hi FROM logs WHERE address = '${feeManager}'` as string,
              })
              const row = result.rows[0]
              return {
                hi: row ? Value.toNumber(row['hi']) : undefined,
                lo: row ? Value.toNumber(row['lo']) : undefined,
              }
            },
            { key: `feeamm:v1:${chainId}:mints-bounds`, store, ttl: Ttl.minutes(1) },
          )
          return order === 'asc' ? bounds.lo : bounds.hi
        })()

        // Sweep the windows from the anchor: window rows are a prefix of the
        // requested ordering, so a filled window returns exactly the rows the
        // unbounded query would.
        const result = await (async () => {
          if (anchor !== undefined)
            for (const window of mintWindowBlocks) {
              // A window reaching past block 0 no longer bounds the scan.
              if (order === 'desc' && window >= anchor) break
              const bound =
                order === 'asc'
                  ? `block_num < ${anchor + window}`
                  : `block_num > ${anchor - window}`
              const result = await fetchPage([...filters, bound])
              if (result.rows.length >= limit + 1) return result
            }
          return fetchPage(filters)
        })()

        // 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: decodeMint.Mint[] = []
        for (const row of page.rows) {
          const mint = decodeMint(row)
          if (mint) data.push(mint)
        }
        return { data, nextCursor: page.nextCursor }
      },
      {
        key: `feeamm:v1:${chainId}:mints:${order}:${userToken ?? ''}:${validatorToken ?? ''}:${cursor ? `cursor:${cursor[0]}:${cursor[1]}` : offset !== undefined ? `page:${options.page}` : 'head'}:${limit}`,
        store,
        ttl: Ttl.seconds(15),
      },
    ),
  )
}

declare namespace listMints {
  /** Options for {@link listMints}. */
  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'
    /** Restrict to pools with this user-side token. */
    userToken?: z.output<typeof Schema.Address> | undefined
    /** Restrict to pools with this validator-side token. */
    validatorToken?: z.output<typeof Schema.Address> | undefined
  }
}

/**
 * Exact total count of pool mints matching the same filters as the page. Runs
 * on ClickHouse over `logs`, pruning on the FeeManager `address` (the page's
 * own engine/key). Because `address` aligns with the `logs` sort key,
 * ClickHouse counts via the sparse primary index without a row scan, so the
 * count is exact and cheap with no cap (`totalCountCapped` is always `false`) —
 * verified at ~20M matching mints in ~1s.
 */
function countMints(
  c: Context<App.Environment>,
  options: countMints.Options,
): Promise<z.output<typeof Schema.CountMeta>> {
  const { chainId, userToken, validatorToken } = options
  const store = c.get('store')
  const tidx = c.get('getTidx')(chainId)
  return Timing.time(c, 'fee_amm_mints_count', () =>
    Store.memoize(
      async () => {
        const filters = [
          `address = '${feeManager}'`,
          `selector IN ('${mintTopic}', '${legacyMintTopic}')`,
        ]
        if (userToken !== undefined) filters.push(`topic2 = '${padTopic(userToken)}'`)
        if (validatorToken !== undefined) filters.push(`topic3 = '${padTopic(validatorToken)}'`)
        const result = await tidx.fetch({
          chainId,
          engine: 'clickhouse',
          query: `SELECT count(*) AS total FROM logs WHERE ${filters.join(' AND ')}` as string,
        })
        const totalCount = Value.toNumber(result.rows[0]?.['total']) ?? 0
        return { totalCountCapped: false, totalCount }
      },
      {
        key: `feeamm:v1:${chainId}:mints-count:${userToken ?? ''}:${validatorToken ?? ''}`,
        store,
        ttl: Ttl.seconds(15),
      },
    ),
  )
}

declare namespace countMints {
  /** Filters for {@link countMints}: the page filters minus pagination. */
  type Options = Pick<listMints.Options, 'chainId' | 'userToken' | 'validatorToken'>
}

/**
 * Decodes one raw `Mint` log row, dispatching on its selector (current vs
 * legacy signature — see the topic constants for the field layouts). Returns
 * undefined for rows that fail to decode so callers can skip them.
 */
function decodeMint(row: Record<string, unknown>): decodeMint.Mint | undefined {
  const selector = Value.toText(row['selector'])
  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_time'])
  const userToken = Schema.Address.safeParse(unpadTopic(Value.toText(row['topic2'])))
  const validatorToken = Schema.Address.safeParse(unpadTopic(Value.toText(row['topic3'])))
  const data = Value.toText(row['data'])
  if (
    !transactionHash.success ||
    blockNumber === undefined ||
    logIndex === undefined ||
    timestamp === undefined ||
    !userToken.success ||
    !validatorToken.success ||
    data === undefined
  )
    return undefined

  const word = (index: number) => data.slice(2 + 64 * index, 2 + 64 * (index + 1))
  const uint = (value: string) =>
    /^[0-9a-fA-F]{64}$/.test(value) ? BigInt(`0x${value}`).toString() : undefined

  if (selector === mintTopic) {
    const minter = Schema.Address.safeParse(`0x${word(0).slice(24)}`)
    const recipient = Schema.Address.safeParse(unpadTopic(Value.toText(row['topic1'])))
    const amountValidatorToken = uint(word(1))
    const liquidity = uint(word(2))
    if (
      !minter.success ||
      !recipient.success ||
      amountValidatorToken === undefined ||
      liquidity === undefined
    )
      return undefined
    return {
      amountValidatorToken,
      blockNumber,
      id: `${transactionHash.data}-${logIndex}`,
      liquidity,
      logIndex,
      minter: minter.data,
      recipient: recipient.data,
      timestamp,
      transactionHash: transactionHash.data,
      userToken: userToken.data,
      validatorToken: validatorToken.data,
    }
  }

  if (selector === legacyMintTopic) {
    const minter = Schema.Address.safeParse(unpadTopic(Value.toText(row['topic1'])))
    const amountUserToken = uint(word(0))
    const amountValidatorToken = uint(word(1))
    const liquidity = uint(word(2))
    if (
      !minter.success ||
      amountUserToken === undefined ||
      amountValidatorToken === undefined ||
      liquidity === undefined
    )
      return undefined
    return {
      amountUserToken,
      amountValidatorToken,
      blockNumber,
      id: `${transactionHash.data}-${logIndex}`,
      liquidity,
      logIndex,
      minter: minter.data,
      timestamp,
      transactionHash: transactionHash.data,
      userToken: userToken.data,
      validatorToken: validatorToken.data,
    }
  }

  return undefined
}

declare namespace decodeMint {
  /**
   * A decoded `Mint` row, carrying bare token addresses. The route shapes
   * them into the enriched `PoolToken` references before responding.
   */
  type Mint = Omit<z.output<typeof schema.getFeeAmmMints.Mint>, 'userToken' | 'validatorToken'> & {
    /** The pool's user-side (fee) token address. */
    userToken: z.output<typeof Schema.Address>
    /** The pool's validator-side token address. */
    validatorToken: z.output<typeof Schema.Address>
  }
}

/**
 * Resolves RPC metadata and selected curated fields. Logo enrichment uses cached verification data or onchain metadata, avoiding one R2 read per token.
 */
async function resolveTokenMap(c: Context<App.Environment>, options: resolveTokenMap.Options) {
  const { addresses, chainId } = options
  const includeLogoUri = options.include?.includes('token.logoUri') ?? true
  const includeVerified = options.include?.includes('token.verified') ?? true
  const snapshot =
    includeLogoUri || includeVerified
      ? await VerifiedTokens.snapshot(c, chainId).catch(() => undefined)
      : undefined
  const resolved = await Timing.time(c, 'tokens', () =>
    Promise.all(
      addresses.map((token) =>
        Tokens.getTokenMetadata(c, { address: token, chainId })
          .then((metadata) => {
            const verified = snapshot?.byAddress.get(token)
            return [
              token,
              {
                currency: metadata.currency,
                decimals: metadata.decimals,
                logoUri: includeLogoUri ? (verified?.logoUri ?? metadata.logoUri) : undefined,
                name: metadata.name,
                symbol: metadata.symbol,
                verified: includeVerified ? snapshot?.byAddress.has(token) : undefined,
              },
            ] as const
          })
          .catch(() => [token, undefined] as const),
      ),
    ),
  )
  return new Map(resolved)
}

declare namespace resolveTokenMap {
  /** Options for {@link resolveTokenMap}. */
  type Options = {
    /** Unique token addresses to resolve. */
    addresses: readonly z.output<typeof Schema.Address>[]
    /** Tempo chain id. */
    chainId: z.output<typeof Schema.ChainId>
    /** Curated token fields to return; omitted to preserve mint enrichment. */
    include?: readonly z.output<typeof schema.getFeeAmmPools.Include>[] | undefined
  }
}

/**
 * Computes a pool id locally: `keccak256(abi.encode(userToken,
 * validatorToken))`, matching the on-chain `FeeManager.getPoolId`.
 */
function poolId(
  userToken: z.output<typeof Schema.Address>,
  validatorToken: z.output<typeof Schema.Address>,
): Hex.Hex {
  return Hash.keccak256(
    AbiParameters.encode(AbiParameters.from(['address', 'address']), [userToken, validatorToken]),
  )
}

/** Builds a self-contained reserve amount after a successful onchain read. */
function reserve(options: reserve.Options) {
  if (options.amount === undefined) return undefined
  return Value.tokenAmount({
    baseUnits: options.amount,
    currency: options.token.currency,
    decimals: options.token.decimals,
  })
}

declare namespace reserve {
  /** Inputs for formatting one side of a Fee AMM pool reserve. */
  type Options = {
    /** Base-unit reserve returned by `FeeManager.getPool`. */
    amount: string | undefined
    /** Token metadata used to describe and format the reserve. */
    token: Pick<z.output<typeof Tokens.schema.TokenReference>, 'currency' | 'decimals'>
  }
}

/** Left-pads a 20-byte address to its 32-byte log-topic form. */
function padTopic(address: z.output<typeof Schema.Address>): string {
  return `0x${'0'.repeat(24)}${address.slice(2)}`
}

/** Unpads a 32-byte log topic to its trailing 20-byte address. */
function unpadTopic(topic: string | undefined): string | undefined {
  return topic === undefined ? undefined : `0x${topic.slice(26)}`
}