import { Hono, type Context } from 'hono'
import { Address, Hex } from 'ox'
import { 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 Mpp from '../../../internal/Mpp.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 Tidx from '../../../internal/Tidx.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 FxOracle from '../FxOracle.js'
import * as Tokens from './tokens.js'
import * as Valuation from './valuation.js'

// TIP-20 tokens emit the standard ERC-20 Transfer event. Supplying the
// signature exposes a decoded `Transfer` CTE (with `from`/`to`/`value`) to
// the indexer SQL.
export const signature = 'event Transfer(address indexed from, address indexed to, uint256 value)'

/**
 * Half-open `[start, end)` bounds of the deterministic TIP-20 contract range.
 * Full-width so the values compare correctly as PostgreSQL bytea, ClickHouse
 * hex strings, and lowercased JavaScript strings.
 */
export const tip20AddressRange = [
  '0x20c0000000000000000000000000000000000000',
  '0x20c1000000000000000000000000000000000000',
] as const

// The global feed spans every TIP-20 token, so its newest-first query cannot
// follow `token_transfers`' token-first sort key. Block windows bound the sort.
const transferWindowBlocks = 100_000

// Sparse global scans retain progress without monopolizing one request.
const transferWindowScanCap = 8

// Recipient-first decoded scans usually fill a page immediately. A larger raw
// batch leaves room to discard non-TIP-20 `Transfer` events without extra I/O.
const decodedTransferBatchSize = 256

// Bound decoded recipient traversal while preserving the largest positional
// page and its extra row for `nextCursor` detection.
const decodedTransferRowCap = Schema.maxPageWindow + 1

// A memo'd TIP-20 transfer emits `TransferWithMemo` alongside the plain
// `Transfer`; the two rows match on `(tx_hash, from, to, amount, token)`. A
// text memo is decoded into `memo`; an MPP attribution fingerprint memo is
// resolved into `attribution` instead.
const memoSignature =
  'event TransferWithMemo(address indexed from, address indexed to, uint256 amount, bytes32 indexed memo)'

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

// Concurrent metadata reads collapse into one deployless multicall. Small
// chunks keep each call under the Tempo RPC's per-call size limit.
const tokenReadChunkSize = 10

/** Zod schemas owned by the transfer handlers. */
export namespace schema {
  /** A transfer token reference with RPC metadata. Inlined to avoid the `transfers`/`tokens` module cycle. */
  export const SourceToken = z
    .object({
      address: Schema.tokenAddress(Tokens.tokenExample.address).check(
        z.describe('The TIP-20 token contract address on Tempo.'),
      ),
      currency: z
        .string()
        .check(
          z.describe(
            'The currency label for this token, such as `USD` for USD-denominated stablecoins.',
          ),
          z.meta({ examples: [Tokens.tokenExample.currency] }),
        ),
      decimals: z
        .number()
        .check(z.int(), z.nonnegative())
        .check(
          z.describe(
            'The number of decimal places the token uses; Tempo stablecoins typically use 6.',
          ),
          z.meta({ examples: [Tokens.tokenExample.decimals] }),
        ),
      logoUri: z.optional(z.string()).check(
        z.describe('A URL for the token’s logo image, returned with `include=token.logoUri`.'),
        z.meta({
          examples: [Tokens.tokenExample.logoUri],
        }),
      ),
      name: z
        .string()
        .check(
          z.describe('The token’s human-readable name.'),
          z.meta({ examples: [Tokens.tokenExample.name] }),
        ),
      symbol: z
        .string()
        .check(
          z.describe('The short ticker symbol wallets and apps show for this token.'),
          z.meta({ examples: [Tokens.tokenExample.symbol] }),
        ),
      verified: z
        .optional(z.boolean())
        .check(
          z.describe(
            'Whether this token is in Tempo’s curated verified list, returned with `include=token.verified`.',
          ),
          z.meta({ examples: [true] }),
        ),
    })
    .check(
      z.describe(
        'The token that moved in this transfer, with RPC metadata and optional curated fields.',
      ),
    )

  const transferToken = OpenApi.component(SourceToken, 'TokenTransferToken')

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

  const valuedTokenAmount = OpenApi.component(ValuedAmount, 'ValuedTokenAmount')

  /**
   * Filter predicates shared verbatim by the `GET /transfers` list query
   * ({@link getTransfers.Query}, which spreads `Predicates.shape`) and `transfer`
   * webhook subscriptions (`Webhooks.schema.TransferFilters`), so the two never
   * drift. Read-only concerns (pagination, windowing, ordering, `include`,
   * `chainId`) are *not* here — they live only on the list query and have no
   * meaning for an open-ended subscription. Consumers that need different
   * per-field docs (e.g. the list query's mutually-exclusive `address`) override
   * that one field.
   */
  export const Predicates = z.object({
    address: z
      .optional(Schema.Address)
      .check(
        z.describe('Match transfers where this account is either the sender or the recipient.'),
      ),
    recipient: z
      .optional(Schema.Address)
      .check(z.describe('Match transfers sent to this recipient address (`to`).')),
    sender: z
      .optional(Schema.Address)
      .check(z.describe('Match transfers sent from this sender address (`from`).')),
    token: z
      .optional(Schema.TokenAddress)
      .check(z.describe('Match transfers for this TIP-20 token contract address.')),
  })

  /** Schemas for the getTransfers (list) operation. */
  export namespace getTransfers {
    /** Optional related resources that callers opt into via `include`. */
    export const Include = z
      .enum(['crossToken', 'memo', ...tokenFields, 'totalCount'])
      .check(z.describe('Related resources to include only when you request them with `include`.'))

    /**
     * Parses a comma-separated `include` query value into a list of optional
     * resources to embed. Extra lookups run only when their resource is
     * requested, keeping the base transfer page fast.
     */
    export const includeQuery = Schema.includeQuery(
      Include,
      'Comma-separated related resources to include, such as `token.logoUri,token.verified,memo,crossToken,totalCount`.',
    )

    /**
     * Query parameters for top-level transfer list requests. `address` is a
     * convenience for "either side" — passing it alongside an explicit
     * `sender` or `recipient` is rejected as ambiguous (the helper would have
     * to AND/OR them in a non-obvious way).
     */
    export const Query = z
      .strictObject({
        // Predicate fields (`address`, `recipient`, `sender`, `token`) are shared
        // verbatim with `transfer` webhook filters via `schema.Predicates`;
        // `address` is overridden below for its list-only mutual-exclusion note.
        ...schema.Predicates.shape,
        address: z
          .optional(Schema.Address)
          .check(
            z.describe(
              'Only include transfers where this account is the sender or recipient. Use this shortcut by itself, not with `sender` or `recipient`.',
            ),
          ),
        'blockNumber.from': Schema.blockNumberBound('transfers', 'from'),
        'blockNumber.to': Schema.blockNumberBound('transfers', 'to'),
        chainId: Schema.ChainIdQuery,
        cursor: Schema.Cursor,
        include: includeQuery,
        limit: Schema.Limit,
        order: Schema.Order,
        page: Schema.Page,
        'timestamp.from': Schema.timestampBound('transfers', 'from'),
        'timestamp.to': Schema.timestampBound('transfers', 'to'),
        'valuation.currency': Schema.Denomination,
      })
      .check(
        z.refine((query) => !(query.address && (query.sender || query.recipient)), {
          error:
            '`address` is a shortcut for `sender` or `recipient`; pass either `address` or one explicit side, not both.',
          path: ['address'],
        }),
        ...Schema.pageChecks(),
      )
      .check(z.describe('Query parameters for listing token transfer events.'))

    /**
     * A single transfer row. The token contract is carried per-row via
     * `sourceToken.address` because `/transfers` is unscoped and rows can span
     * multiple contracts.
     */
    export const Transfer = z
      .object({
        attribution: z
          .optional(z.string())
          .check(
            z.describe(
              'The resolved MPP service name, when the transfer memo matches a known service fingerprint. Returned when you request `include=memo`.',
            ),
            z.meta({ examples: ['example-service'] }),
          ),
        blockNumber: z
          .number()
          .check(
            z.int(),
            z.nonnegative(),
            z.describe('The block number where this transfer was recorded.'),
            z.meta({ examples: [23456789] }),
          ),
        destinationAmount: z
          .optional(valuedTokenAmount)
          .check(
            z.describe(
              'Amount delivered to the recipient for a cross-token transfer. Omitted with `destinationToken` for a same-token transfer.',
            ),
          ),
        destinationToken: z
          .optional(transferToken)
          .check(
            z.describe(
              'The token delivered to the recipient when it differs from `sourceToken`. Omitted for same-token transfers and returned only with `include=crossToken`.',
            ),
          ),
        id: z.string().check(
          z.describe(
            'A stable resource ID for this transfer, built from `${transactionHash}-${logIndex}`.',
          ),
          z.meta({
            examples: ['0x515801d7f9a5ac705e793e85904c9c69b3f1694b465cc8fb6ba3f0298dc82665-0'],
          }),
        ),
        logIndex: z
          .number()
          .check(
            z.int(),
            z.nonnegative(),
            z.describe('The transfer event’s log index within the block.'),
            z.meta({ examples: [0] }),
          ),
        memo: z
          .optional(z.string())
          .check(
            z.describe(
              'The decoded text memo from the matching `TransferWithMemo` event, when one is present. Returned when you request `include=memo`.',
            ),
            z.meta({ examples: ['Payment for invoice 123'] }),
          ),
        recipient: Schema.Address.check(
          z.describe('The account address that received the tokens.'),
        ),
        sender: Schema.Address.check(z.describe('The account address that sent the tokens.')),
        sourceAmount: valuedTokenAmount,
        sourceToken: transferToken.check(z.describe('The token that moved.')),
        timestamp: z.iso
          .datetime()
          .check(
            z.describe('The block timestamp when this transfer was recorded.'),
            z.meta({ examples: ['2024-01-01T00:00:00Z'] }),
          ),
        transactionHash: Schema.Hash.check(
          z.describe('The hash of the transaction that emitted this transfer event.'),
        ),
      })
      .check(
        z.refine(
          (data) =>
            (data.destinationToken === undefined) === (data.destinationAmount === undefined),
          {
            error: '`destinationAmount` and `destinationToken` must be returned together.',
            path: ['destinationAmount'],
          },
        ),
        z.describe(
          'One TIP-20 token transfer event, optionally including a payment memo for reconciliation or invoices.',
        ),
      )

    const transfer = OpenApi.component(Transfer, 'TokenTransfer')

    /** Page-level resources: opt-in counts plus valuation rate provenance. */
    export const Meta = z
      .object({
        valuation: z
          .optional(Valuation.schema.Pricing)
          .check(
            z.describe(
              'Rate provenance for leg valuations. Present when conversion rates were ' +
                'consulted; absent when every value was identity-valued or rates were unavailable.',
            ),
          ),
        totalCount: z.optional(Schema.TotalCount),
        totalCountCapped: z.optional(Schema.TotalCountCapped),
      })
      .check(z.describe('Page-level resources attached to this response.'))

    const meta = OpenApi.component(Meta, 'TokenTransferListMeta')

    /** Page of TIP-20 token transfers across the chain, ordered by block then log index. */
    export const Response = OpenApi.component(
      z
        .object({
          data: z.array(transfer).check(z.describe('The transfers in this page.')),
          // Response-wide `meta` carries the opt-in capped total count
          // (`include=totalCount`) and valuation rate provenance; per-row embeds
          // (`memo`, `token`) attach to each `data` item, not here.
          meta: z
            .optional(meta)
            .check(
              z.describe('Page-level resources: opt-in counts and valuation rate provenance.'),
            ),
          nextCursor: Schema.NextCursor,
        })
        .check(z.describe('A page of TIP-20 token transfer events across the chain.')),
      'TokenTransferList',
    )
  }
}

type UnenrichedTransfer = Omit<
  z.output<typeof schema.getTransfers.Transfer>,
  'destinationAmount' | 'destinationToken' | 'sourceAmount' | 'sourceToken'
> & {
  destinationToken?: CrossToken | undefined
  sourceToken: CrossToken
}

/** Creates transfer handlers. */
export function transfers(options: transfers.Options = {}) {
  const oracle = options.fx?.oracle ?? FxOracle.ecb()
  return new Hono<App.Environment>().get(
    '/v1/transfers',
    Auth.policy({ apiKey: { scopes: ['data:read'] }, mpp: true, public: true }),
    OpenApi.validate('query', schema.getTransfers.Query, {
      code: 'query_invalid',
      message: 'Invalid query parameters',
    }),
    OpenApi.describeRoute({
      description: 'List token transfer events across Tempo.',
      operationId: 'getTransfers',
      responses: OpenApi.responses({
        errors: {
          400: { codes: ['chain_id_invalid', 'chain_id_unsupported', 'query_invalid'] },
          502: 'The indexer or Tempo RPC could not serve this request.',
        },
        success: {
          description: 'A page of token transfer events.',
          schema: schema.getTransfers.Response,
        },
      }),
      summary: 'List transfers',
      tags: ['Transfers'],
    }),
    Cache.response<App.Environment>({
      cacheControl: Cache.policies.feed,
      name: 'tempo-api:transfers:v3',
      key: (c) => Cache.urlKey(c, schema.getTransfers.Query),
    }),
    async (c) => {
      if (Auth.narrowAccess) return Auth.paidAccessError(c)
      if (OpenApi.narrowValidation)
        return OpenApi.validationError(c, {
          code: 'query_invalid',
          message: 'Invalid query parameters',
        })

      const options = c.req.valid('query')
      const chainId = options.chainId ?? c.get('chainId')
      const denomination = options['valuation.currency']

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

      // Filters shared by the page query and the opt-in total count, so the
      // count always matches the page it annotates.
      const predicates = {
        chainId,
        eitherSide: options.address,
        fromBlock: options['blockNumber.from'],
        fromTimestamp: options['timestamp.from'],
        recipient: options.recipient,
        sender: options.sender,
        toBlock: options['blockNumber.to'],
        toTimestamp: options['timestamp.to'],
        token: options.token,
      }

      try {
        // `totalCount` is opt-in and independent of the page rows, so kick it
        // off concurrently with the page query. It is best-effort: a count
        // failure (e.g. upstream timeout) omits `meta` rather than failing the
        // whole page.
        const countPromise = options.include.includes('totalCount')
          ? count(c, {
              ...predicates,
              crossToken: options.include.includes('crossToken'),
              timing: 'transfers_count',
            }).catch(() => undefined)
          : undefined

        const page = await query(c, {
          ...predicates,
          cursor: options.cursor,
          limit: options.limit,
          order: options.order,
          page: options.page,
          timing: 'transfers',
        })

        let data: UnenrichedTransfer[] = page.data.map((row) => ({
          blockNumber: row.blockNumber,
          id: `${row.transactionHash}-${row.logIndex}`,
          logIndex: row.logIndex,
          recipient: row.recipient,
          sender: row.sender,
          sourceToken: { address: row.address, amount: row.amount },
          timestamp: row.timestamp,
          transactionHash: row.transactionHash,
        }))

        const includeCrossToken = options.include.includes('crossToken')
        const includeMemo = options.include.includes('memo')
        // Attach memos before cross-token folding changes the matching fields.
        if (includeMemo && includeCrossToken && data.length > 0) {
          const memoByRow = await memoMap(c, { chainId, rows: data })
          data = data.map((transfer) => {
            const entry = memoByRow.get(
              transferKey(
                transfer.transactionHash,
                transfer.sender,
                transfer.recipient,
                transfer.sourceToken.amount,
                transfer.sourceToken.address,
              ),
            )
            if (!entry) return transfer
            return {
              ...transfer,
              ...(entry.attribution ? { attribution: entry.attribution } : {}),
              ...(entry.memo ? { memo: entry.memo } : {}),
            }
          })
        }
        // Cross-token routing is opt-in. One `Transfer` lookup bounded to the
        // page's transactions folds the account server's approve+swap+forward
        // multicall into a single row on the delivery leg: `sourceToken` becomes
        // the token swapped into the DEX and `destinationToken` the token
        // delivered to the recipient. The intermediate DEX legs and the fee leg
        // are dropped from the page. Folding anchors on the delivery leg (whose
        // own event token is the delivered token), so a `token=` filter matches
        // cross-token rows by their destination token, and each transfer still
        // appears exactly once across pages.
        if (includeCrossToken && data.length > 0) {
          const { fold, suppress } = await crossTokenMap(c, { chainId, rows: data })
          data = data.flatMap((transfer) => {
            const key = eventKey(transfer.transactionHash, transfer.logIndex)
            const folded = fold.get(key)
            if (folded)
              return [
                {
                  ...transfer,
                  destinationToken: folded.destinationToken,
                  recipient: folded.recipient,
                  sender: folded.sender,
                  sourceToken: folded.sourceToken,
                },
              ]
            return suppress.has(key) ? [] : [transfer]
          })
        }
        const includeLogoUri = options.include.includes('token.logoUri')
        const includeVerified = options.include.includes('token.verified')
        // Curated data is needed only for requested valuation or token fields.
        const snapshot =
          data.length > 0 && (denomination !== undefined || includeLogoUri || includeVerified)
            ? await VerifiedTokens.snapshot(c, chainId).catch(() => undefined)
            : undefined
        const uniqueTokens = Array.from(
          new Set(
            data.flatMap((transfer) => [
              transfer.sourceToken.address,
              ...(transfer.destinationToken ? [transfer.destinationToken.address] : []),
            ]),
          ),
        )
        // Memo and token metadata depend only on the page rows, so resolve
        // them concurrently when cross-token folding is not requested.
        const memoPromise =
          includeMemo && !includeCrossToken && data.length > 0
            ? memoMap(c, { chainId, rows: data })
            : undefined
        const tokensPromise = Timing.time(c, 'tokens', async () => {
          const tokensByAddress = new Map<
            string,
            Awaited<ReturnType<typeof Tokens.getTokenMetadata>>
          >()
          for (let index = 0; index < uniqueTokens.length; index += tokenReadChunkSize) {
            const group = uniqueTokens.slice(index, index + tokenReadChunkSize)
            const entries = await Promise.all(
              group.map(async (token) => {
                const [logoUri, metadata] = await Promise.all([
                  includeLogoUri
                    ? Timing.time(c, 'token_logo', () =>
                        Tokens.getTokenLogo(c, { address: token, chainId }),
                      ).catch(() => undefined)
                    : undefined,
                  Timing.time(c, 'token_metadata', () =>
                    Tokens.getTokenMetadata(c, { address: token, chainId }),
                  ),
                ])
                return [
                  token.toLowerCase(),
                  {
                    ...metadata,
                    logoUri: includeLogoUri
                      ? (logoUri ??
                        snapshot?.byAddress.get(token.toLowerCase())?.logoUri ??
                        metadata.logoUri)
                      : undefined,
                  },
                ] as const
              }),
            )
            for (const entry of entries) tokensByAddress.set(...entry)
          }
          return tokensByAddress
        })
        const [memoByRow, tokensByAddress] = await Promise.all([memoPromise, tokensPromise])
        if (memoByRow)
          data = data.map((transfer) => {
            const entry = memoByRow.get(
              transferKey(
                transfer.transactionHash,
                transfer.sender,
                transfer.recipient,
                transfer.sourceToken.amount,
                transfer.sourceToken.address,
              ),
            )
            if (!entry) return transfer
            return {
              ...transfer,
              ...(entry.attribution ? { attribution: entry.attribution } : {}),
              ...(entry.memo ? { memo: entry.memo } : {}),
            }
          })

        // Value each leg via its curated display currency, in the requested
        // denomination.
        const rates = denomination
          ? await Valuation.ratesFor(c, {
              currencies: data.flatMap((transfer) =>
                [
                  transfer.sourceToken.address,
                  ...(transfer.destinationToken ? [transfer.destinationToken.address] : []),
                ].flatMap((address) => {
                  const held = snapshot?.byAddress.get(address.toLowerCase())?.currency
                  return held === undefined ? [] : [held]
                }),
              ),
              denomination,
              oracle,
            })
          : undefined
        const enrich = (leg: CrossToken) => {
          const metadata = tokensByAddress.get(leg.address.toLowerCase())
          if (!metadata) throw new Error(`Token metadata unavailable for ${leg.address}`)
          return {
            amount: {
              ...Value.tokenAmount({
                baseUnits: leg.amount,
                currency: metadata.currency,
                decimals: metadata.decimals,
              }),
              ...(denomination
                ? {
                    valuation: Valuation.valuationFor({
                      amount: BigInt(leg.amount),
                      denomination,
                      rates,
                      token: snapshot?.byAddress.get(leg.address.toLowerCase()),
                    }),
                  }
                : {}),
            },
            token: {
              address: leg.address,
              currency: metadata.currency,
              decimals: metadata.decimals,
              ...(metadata.logoUri !== undefined ? { logoUri: metadata.logoUri } : {}),
              name: metadata.name,
              symbol: metadata.symbol,
              ...(includeVerified && snapshot
                ? { verified: snapshot.byAddress.has(leg.address.toLowerCase()) }
                : {}),
            },
          }
        }
        const responseData: z.output<typeof schema.getTransfers.Transfer>[] = data.map(
          ({ destinationToken, sourceToken, ...transfer }) => {
            const source = enrich(sourceToken)
            if (!destinationToken)
              return {
                ...transfer,
                sourceAmount: source.amount,
                sourceToken: source.token,
              }
            const destination = enrich(destinationToken)
            return {
              ...transfer,
              destinationAmount: destination.amount,
              destinationToken: destination.token,
              sourceAmount: source.amount,
              sourceToken: source.token,
            }
          },
        )

        // Wait at most 2.5s more for a still-pending count (cold indexer
        // reads have stalled 20s+); a timed-out count omits `meta` counts.
        const counts = countPromise
          ? await Promise.race([
              countPromise,
              new Promise<undefined>((resolve) => setTimeout(() => resolve(undefined), 2_500)),
            ])
          : undefined
        const meta = {
          ...(rates ? { valuation: Valuation.pricing(rates, oracle) } : {}),
          ...counts,
        }

        return c.json(
          Response.validated(schema.getTransfers.Response, {
            data: responseData,
            ...(Object.keys(meta).length > 0 ? { meta } : {}),
            nextCursor: page.nextCursor,
          }),
          200,
        )
      } catch (cause) {
        if (cause instanceof Valuation.UnsupportedDenominationError)
          return Response.error(c, { code: 'query_invalid', message: cause.message, status: 400 })
        return Response.upstream(c, cause, { page: options.page })
      }
    },
  )
}

export declare namespace transfers {
  /** Options for the transfer handlers. */
  type Options = {
    /** FX configuration backing leg valuation. */
    fx?: Valuation.addresses.Fx | undefined
  }
}

/**
 * Shared TIP-20 `Transfer` page query. Used by `GET /transfers` and the
 * webhook poller. Builds signature-decoded `Transfer` CTE queries, applies
 * opaque keyset pagination on `(block_num, log_idx)`, and decodes rows into a
 * strongly-typed intermediate shape that each caller maps to its own public
 * response.
 *
 * `eitherSide` is exposed alongside one-sided `sender`/`recipient` filters
 * because it requires two upstream scans. The helper owns their ordered merge.
 *
 * Memoization lives inside the helper under one canonical `transfers:v1:…`
 * key derived from every filter — so equivalent filter combinations share a
 * single cache entry. Callers supply only a `timing` label for Server-Timing.
 */
export function query(c: Context<App.Environment>, options: query.Options) {
  const { chainId, timing, ...rest } = options
  const store = c.get('store')
  const tidx = c.get('getTidx')(chainId)
  const cursor = decodeTransferCursor(options.cursor)

  return Timing.time(c, timing, () =>
    Store.memoize(
      async () => {
        const page = await scan({ tidx }, { chainId, ...rest })
        return {
          data: page.items.map((item) => item.data),
          nextCursor: page.nextCursor,
        }
      },
      {
        key: `transfers:v2:${chainId}:${options.order}:${options.token ?? ''}:${options.eitherSide ?? ''}:${options.sender ?? ''}:${options.recipient ?? ''}:${options.fromBlock ?? ''}:${options.toBlock ?? ''}:${options.fromTimestamp ?? ''}:${options.toTimestamp ?? ''}:${cursor ? `cursor:${cursor.position[0]}:${cursor.position[1]}:${cursor.offset}` : options.page !== undefined && options.page > 1 ? `page:${options.page}` : 'head'}:${options.limit}`,
        store,
        ttl: Ttl.seconds(15),
      },
    ),
  )
}

export declare namespace query {
  /** Options for the shared TIP-20 `Transfer` page query. */
  type Options = {
    chainId: z.output<typeof Schema.ChainId>
    /** Opaque keyset cursor anchoring the page; omit for the head page. */
    cursor?: string | undefined
    /** Restrict to rows where this address appears on either side. */
    eitherSide?: z.output<typeof Schema.Address> | undefined
    fromBlock?: number | undefined
    fromTimestamp?: string | undefined
    limit: number
    /** 1-indexed page number (positional pagination; exclusive with `cursor`). */
    page?: number | undefined
    order: 'asc' | 'desc'
    recipient?: z.output<typeof Schema.Address> | undefined
    sender?: z.output<typeof Schema.Address> | undefined
    /** Server-Timing label for the caller's surface. */
    timing: string
    toBlock?: number | undefined
    toTimestamp?: string | undefined
    /** Restrict to a single TIP-20 token contract. */
    token?: z.output<typeof Schema.Address> | undefined
  }
  /** A decoded `Transfer` row before per-surface shaping. */
  type Row = {
    /** Token contract that emitted the event. */
    address: z.output<typeof Schema.Address>
    amount: string
    blockNumber: number
    logIndex: number
    recipient: z.output<typeof Schema.Address>
    sender: z.output<typeof Schema.Address>
    timestamp: string
    transactionHash: z.output<typeof Schema.Hash>
  }
  type Page = {
    data: readonly Row[]
    nextCursor: string | null
  }
}

/**
 * Context-free core of {@link query}: reads token-scoped and global rows from
 * `token_transfers`, otherwise the decoded `Transfer` CTE, then paginates and
 * decodes them. Also used by the webhook poller.
 */
export async function scan(deps: scan.Deps, options: scan.Options): Promise<scan.Page> {
  const { chainId, limit, order } = options
  const { tidx } = deps
  const direction = order === 'asc' ? 'ASC' : 'DESC'
  const tokenScoped = options.token !== undefined
  // 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 = decodeTransferCursor(options.cursor)
  // 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 pageOffset =
    options.page !== undefined && options.page > 1 ? (options.page - 1) * limit : undefined
  const offset = pageOffset ?? (cursor?.offset === 0 ? undefined : cursor?.offset)
  const address = options.eitherSide
  const addressScoped = address !== undefined
  const recipient = options.recipient
  const recipientDecoded = !tokenScoped && recipient !== undefined
  const decoded = !tokenScoped && (addressScoped || recipientDecoded)
  const splitAddress = addressScoped && !recipientDecoded
  // Only one-sided `"from"`/`"to"` equality shapes hit the planner's
  // deterministic top-N kill, so only they take the capped fallback.
  const sideFiltered =
    addressScoped || options.sender !== undefined || options.recipient !== undefined
  const global = !tokenScoped && !sideFiltered

  // Transfer events filtered by any combination of contract / sender /
  // recipient / block / timestamp. The address shortcut is split below so
  // each upstream query can use one indexed side.
  const filters = global
    ? [
        `token >= '${tip20AddressRange[0]}'`,
        `token < '${tip20AddressRange[1]}'`,
        ...(options.fromBlock !== undefined ? [`block_num >= ${options.fromBlock}`] : []),
        ...(options.toBlock !== undefined ? [`block_num <= ${options.toBlock}`] : []),
        ...(options.fromTimestamp !== undefined
          ? [`block_timestamp >= '${formatClickHouseDateTime(options.fromTimestamp)}'`]
          : []),
        ...(options.toTimestamp !== undefined
          ? [`block_timestamp <= '${formatClickHouseDateTime(options.toTimestamp)}'`]
          : []),
      ]
    : decoded
      ? [
          ...(options.sender !== undefined ? [`"from" = '${options.sender}'`] : []),
          ...(recipient !== undefined ? [`"to" = '${recipient}'`] : []),
          ...(options.fromBlock !== undefined ? [`block_num >= ${options.fromBlock}`] : []),
          ...(options.toBlock !== undefined ? [`block_num <= ${options.toBlock}`] : []),
          ...(options.fromTimestamp !== undefined
            ? [`block_timestamp >= '${options.fromTimestamp}'`]
            : []),
          ...(options.toTimestamp !== undefined
            ? [`block_timestamp <= '${options.toTimestamp}'`]
            : []),
        ]
      : filterClauses(options)
  if (cursor !== undefined && !decoded)
    filters.push(
      Cursor.keyset([
        { literal: Cursor.literal(cursor.position[0], 'int'), name: 'block_num', order },
        { literal: Cursor.literal(cursor.position[1], 'int'), name: 'log_idx', order },
      ]),
    )
  const window = addressScoped ? (offset ?? 0) + limit + 1 : limit + 1
  const fetchRows = async (options_fetch: fetchRows.Options = {}) => {
    const side = options_fetch.side
    const sideFilters =
      side !== undefined && address !== undefined
        ? [`"${side}" = '${address}'`, ...filters]
        : filters
    const queryFilters = [
      ...sideFilters,
      ...(options_fetch.lo !== undefined ? [`block_num >= ${options_fetch.lo}`] : []),
      ...(options_fetch.hi !== undefined ? [`block_num <= ${options_fetch.hi}`] : []),
    ]
    const materialized = tokenScoped || global
    // Keep a distinct alias until tidx.ts parses ClickHouse timestamps as UTC.
    const select = materialized
      ? `"from", "to", token AS address, amount AS value, tx_hash, block_num, log_idx, toString(block_timestamp, 'UTC') AS block_timestamp_utc`
      : `"from", "to", address, value, tx_hash, block_num, log_idx, block_timestamp`
    const table = materialized ? 'token_transfers' : 'Transfer'
    const execute = async (
      filters_execute: readonly string[],
      queryLimit: number,
      queryOffset?: number,
    ): Promise<readonly Record<string, unknown>[]> => {
      const where = filters_execute.length > 0 ? `WHERE ${filters_execute.join(' AND ')}` : ''
      const result = await tidx.fetch({
        chainId,
        ...(materialized ? { engine: 'clickhouse' } : { signatures: [signature] }),
        query: `
          SELECT ${select}
          FROM ${table}
          ${where}
          ORDER BY block_num ${direction}, log_idx ${direction}
          LIMIT ${queryLimit}${queryOffset !== undefined ? ` OFFSET ${queryOffset}` : ''}
        ` as string,
      })
      return result.rows
    }

    if (decoded) {
      const target = (offset ?? 0) + limit + 1
      // Filter the physical token column so ClickHouse can use the materialized table ordering.
      const filters_materialized = [
        `token >= '${tip20AddressRange[0]}'`,
        `token < '${tip20AddressRange[1]}'`,
        ...(options.sender !== undefined ? [`"from" = '${options.sender}'`] : []),
        ...(recipient !== undefined ? [`"to" = '${recipient}'`] : []),
        ...(options.fromBlock !== undefined ? [`block_num >= ${options.fromBlock}`] : []),
        ...(options.toBlock !== undefined ? [`block_num <= ${options.toBlock}`] : []),
        ...(options.fromTimestamp !== undefined
          ? [`block_timestamp >= '${formatClickHouseDateTime(options.fromTimestamp)}'`]
          : []),
        ...(options.toTimestamp !== undefined
          ? [`block_timestamp <= '${formatClickHouseDateTime(options.toTimestamp)}'`]
          : []),
      ]
      if (cursor !== undefined)
        filters_materialized.push(
          Cursor.keyset([
            { literal: Cursor.literal(cursor.position[0], 'int'), name: 'block_num', order },
            { literal: Cursor.literal(cursor.position[1], 'int'), name: 'log_idx', order },
          ]),
        )
      const sideFilters_materialized =
        side !== undefined && address !== undefined
          ? [`"${side}" = '${address}'`, ...filters_materialized]
          : filters_materialized
      const where_materialized =
        sideFilters_materialized.length > 0 ? `WHERE ${sideFilters_materialized.join(' AND ')}` : ''
      try {
        const result = await tidx.fetch({
          chainId,
          engine: 'clickhouse',
          query: `
            SELECT DISTINCT "from", "to", token AS address, amount AS value, tx_hash, block_num, log_idx, toString(block_timestamp, 'UTC') AS block_timestamp_utc
            FROM token_transfers
            ${where_materialized}
            ORDER BY block_num ${direction}, log_idx ${direction}
            LIMIT ${target}
          ` as string,
        })
        return {
          capped: false,
          rows: result.rows.slice(splitAddress ? 0 : (offset ?? 0), target),
          sample: false,
        }
      } catch (error) {
        // Older chains can reject materialized address scans, so retain the
        // bounded decoded scan as a compatibility fallback.
        if (!Tidx.isQueryRejection(error)) throw error
      }

      const rows: Record<string, unknown>[] = []
      let anchor = cursor?.position
      let rawRowCount = 0
      while (rows.length < target && rawRowCount < decodedTransferRowCap) {
        const batchLimit = Math.min(
          Math.max(target - rows.length, decodedTransferBatchSize),
          Schema.maxPageWindow,
          decodedTransferRowCap - rawRowCount,
        )
        const batch = await (async () => {
          if (anchor === undefined) return execute(queryFilters, batchLimit)

          // Signature-decoded queries reject or mis-plan tuple keysets. Read
          // the cursor block and strict outer block range separately.
          const block = Cursor.literal(anchor[0]!, 'int')
          const index = Cursor.literal(anchor[1]!, 'int')
          const comparison = order === 'asc' ? '>' : '<'
          const sameBlock = await execute(
            [...queryFilters, `block_num = ${block}`, `log_idx ${comparison} ${index}`],
            batchLimit,
          )
          if (sameBlock.length >= batchLimit) return sameBlock
          const outerBlocks = await execute(
            [...queryFilters, `block_num ${comparison} ${block}`],
            batchLimit - sameBlock.length,
          )
          return [...sameBlock, ...outerBlocks]
        })()

        rawRowCount += batch.length
        rows.push(...batch.filter(isTip20TransferRow))
        if (batch.length < batchLimit) break
        const last = batch.at(-1)
        const block = Value.toNumber(last?.['block_num'])
        const index = Value.toNumber(last?.['log_idx'])
        if (block === undefined || index === undefined) break
        anchor = [block, index]
      }
      const capped = rows.length < target && rawRowCount >= decodedTransferRowCap
      return {
        capped,
        ...(capped && anchor !== undefined
          ? {
              resume: {
                offset: Math.max(0, (offset ?? 0) - rows.length),
                position: anchor,
              },
            }
          : {}),
        rows: rows.slice(splitAddress ? 0 : (offset ?? 0), target),
        sample: false,
      }
    }

    const queryLimit = options_fetch.limit ?? window
    const queryOffset = options_fetch.offset ?? (!addressScoped ? offset : undefined)
    const where = queryFilters.length > 0 ? `WHERE ${queryFilters.join(' AND ')}` : ''
    try {
      return {
        capped: false,
        rows: await execute(queryFilters, queryLimit, queryOffset),
        sample: false,
      }
    } catch (error) {
      // The planner kills a one-sided top-N walk when its matches sit far
      // from head (422 `db error`). Retry as a plain capped scan, returning
      // the whole sample so truncation stays detectable.
      if (!sideFiltered || !Tidx.isDeterministicError(error)) throw error
      const result = await tidx.fetch({
        chainId,
        ...(materialized ? { engine: 'clickhouse' } : { signatures: [signature] }),
        query: `
          SELECT ${select}
          FROM ${table}
          ${where}
          LIMIT ${Schema.countCap}
        ` as string,
      })
      return { capped: result.rows.length >= Schema.countCap, rows: result.rows, sample: true }
    }
  }

  const windowed = global
  const fetchWindowedRows = async () => {
    const indexedCeiling =
      options.toTimestamp === undefined
        ? await indexedHead(tidx, { chainId })
        : await indexedBlockAtOrBefore(tidx, { chainId, timestamp: options.toTimestamp })
    if (indexedCeiling === undefined) return { capped: false, rows: [], sample: false }

    const floor = options.fromBlock ?? 0
    const ceiling = Math.min(options.toBlock ?? indexedCeiling, indexedCeiling)
    const target = (offset ?? 0) + limit + 1
    const rows: Record<string, unknown>[] = []
    const cursorBlock = cursor?.position[0]
    let resumePosition: readonly [number, number] | undefined
    let scannedWindows = 0
    let anchor =
      order === 'asc'
        ? Math.max(floor, cursorBlock ?? floor)
        : Math.min(ceiling, cursorBlock ?? ceiling)
    while (
      anchor >= floor &&
      anchor <= ceiling &&
      rows.length < target &&
      scannedWindows < transferWindowScanCap
    ) {
      const window_lo =
        order === 'asc' ? anchor : Math.max(floor, anchor - transferWindowBlocks + 1)
      const window_hi =
        order === 'desc' ? anchor : Math.min(ceiling, anchor + transferWindowBlocks - 1)
      let windowOffset = 0
      while (rows.length < target) {
        const queryLimit = Math.min(target - rows.length, Schema.maxPageWindow)
        const result = await fetchRows({
          hi: window_hi,
          limit: queryLimit,
          lo: window_lo,
          offset: windowOffset,
        })
        rows.push(...result.rows)
        windowOffset += result.rows.length
        if (result.rows.length < queryLimit) break
      }
      scannedWindows++
      // Resume after the fully scanned boundary without skipping the next block.
      resumePosition = order === 'asc' ? [window_hi, Number.MAX_SAFE_INTEGER] : [window_lo, 0]
      anchor = order === 'asc' ? window_hi + 1 : window_lo - 1
    }
    const capped =
      rows.length < target &&
      anchor >= floor &&
      anchor <= ceiling &&
      scannedWindows >= transferWindowScanCap
    return {
      capped,
      ...(capped && resumePosition !== undefined
        ? {
            resume: {
              offset: Math.max(0, (offset ?? 0) - rows.length),
              position: resumePosition,
            },
          }
        : {}),
      rows: rows.slice(offset ?? 0),
      sample: false,
    }
  }

  // With `to = recipient`, the either-side address is redundant when equal
  // and otherwise reduces to `from = address`, so one ordered stream suffices.
  const recipientAddressSide =
    recipientDecoded && address !== undefined && address !== recipient ? 'from' : undefined
  // TIDX cannot plan `(from = X OR to = X)` over the decoded Transfer table.
  // Fetch one stream per side only when windowing or recipient reduction does not apply.
  const fetched = windowed
    ? [await fetchWindowedRows()]
    : splitAddress
      ? await Promise.all([fetchRows({ side: 'from' }), fetchRows({ side: 'to' })])
      : [await fetchRows(recipientAddressSide === undefined ? {} : { side: recipientAddressSide })]
  const { capped, resume, rows } = collate(fetched, {
    limit: limit + 1,
    offset: offset ?? 0,
    order,
  })

  // The next page anchors below the last fetched row's `(block, log_idx)`.
  const page = Cursor.paginate({
    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 items: scan.Item[] = []
  for (const row of page.rows) {
    const tokenAddress = Schema.Address.safeParse(row['address'])
    const from = Schema.Address.safeParse(row['from'])
    const to = Schema.Address.safeParse(row['to'])
    const transactionHash = Schema.Hash.safeParse(row['tx_hash'])
    const amount = Value.toIntegerString(row['value'])
    const blockNumber = Value.toNumber(row['block_num'])
    const logIndex = Value.toNumber(row['log_idx'])
    const timestamp = Value.toIsoDateTime(row['block_timestamp_utc'] ?? row['block_timestamp'])
    if (
      tokenAddress.success &&
      from.success &&
      to.success &&
      transactionHash.success &&
      amount !== undefined &&
      blockNumber !== undefined &&
      logIndex !== undefined &&
      timestamp !== undefined
    )
      items.push({
        cursor: [blockNumber, logIndex],
        data: {
          address: tokenAddress.data,
          amount,
          blockNumber,
          logIndex,
          recipient: to.data,
          sender: from.data,
          timestamp,
          transactionHash: transactionHash.data,
        },
      })
  }

  // Every matching row through `resume.position` was inspected, so advancing
  // there cannot skip a match. Positional continuations retain their offset.
  const nextCursor = resume
    ? Cursor.encode(resume.offset > 0 ? [...resume.position, resume.offset] : [...resume.position])
    : capped
      ? null
      : page.nextCursor
  return {
    capped,
    hasMore: resume ? true : capped ? false : page.hasMore,
    items,
    limit,
    nextCursor,
    ...(resume ? { resumeCursor: resume.position } : {}),
  }
}

declare namespace fetchRows {
  /** Source-side, block-window, and pagination constraints for one upstream scan. */
  type Options = {
    /** Inclusive highest block number. */
    hi?: number | undefined
    /** Maximum rows to return. */
    limit?: number | undefined
    /** Inclusive lowest block number. */
    lo?: number | undefined
    /** Rows to skip before returning results. */
    offset?: number | undefined
    /** Indexed event participant side. */
    side?: 'from' | 'to' | undefined
  }
}

/** Reads the latest block available to TIDX for a descending window scan. */
async function indexedHead(tidx: Tidx.Client, options: indexedHead.Options) {
  const result = await tidx.fetch({
    chainId: options.chainId,
    query: 'SELECT num FROM blocks ORDER BY num DESC LIMIT 1',
  })
  return Value.toNumber(result.rows[0]?.['num'])
}

declare namespace indexedHead {
  /** Chain whose indexed head should be read. */
  type Options = {
    /** Target chain id. */
    chainId: number
  }
}

/** Reads the latest indexed block at or before a timestamp. */
async function indexedBlockAtOrBefore(tidx: Tidx.Client, options: indexedBlockAtOrBefore.Options) {
  const { chainId, timestamp } = options
  const result = await tidx.fetch({
    chainId,
    query: `
      SELECT num
      FROM blocks
      WHERE timestamp <= '${Tidx.escape(timestamp)}'
      ORDER BY num DESC
      LIMIT 1
    ` as string,
  })
  return Value.toNumber(result.rows[0]?.['num'])
}

declare namespace indexedBlockAtOrBefore {
  /** Chain and cutoff for the highest qualifying indexed block. */
  type Options = {
    /** Target chain id. */
    chainId: number
    /** Inclusive timestamp cutoff. */
    timestamp: string
  }
}

/** Formats an ISO timestamp for ClickHouse `DateTime64` comparisons. */
function formatClickHouseDateTime(value: string) {
  return value.replace('T', ' ').replace(/Z$/, '')
}

/**
 * Collates per-side transfer streams into one ordered page window: split and
 * sampled streams are deduped, globally ordered, and paged app-side, while a
 * lone engine-paged stream passes through. Ordered capped streams may carry a
 * safe raw continuation; unordered samples cannot.
 */
export function collate(
  fetched: readonly collate.Stream[],
  options: collate.Options,
): collate.Result {
  const capped = fetched.some((stream) => stream.capped)
  const cappedStreams = fetched.filter((stream) => stream.capped)
  const resumes = cappedStreams.flatMap((stream) => (stream.resume ? [stream.resume] : []))
  if (
    fetched.length > 1 &&
    capped &&
    !fetched.some((stream) => stream.sample) &&
    resumes.length === cappedStreams.length
  ) {
    // Advance only through the least-progressed side, so the next strict
    // keyset scan cannot rediscover any returned rows.
    const position = resumes
      .map((resume) => resume.position)
      .reduce((candidate, current) => {
        const comparison = candidate[0] - current[0] || candidate[1] - current[1]
        if (options.order === 'asc') return comparison <= 0 ? candidate : current
        return comparison >= 0 ? candidate : current
      })
    const safeRows = fetched.flatMap((stream) =>
      stream.rows.filter((row) => {
        const blockNumber = Value.toNumber(row['block_num'])
        const logIndex = Value.toNumber(row['log_idx'])
        if (blockNumber === undefined || logIndex === undefined) return false
        const comparison = blockNumber - position[0] || logIndex - position[1]
        return options.order === 'asc' ? comparison <= 0 : comparison >= 0
      }),
    )
    const merged = mergeTransferRows(safeRows, {
      limit: options.offset + options.limit,
      offset: 0,
      order: options.order,
    })
    const rows = merged.slice(options.offset, options.offset + options.limit)
    if (rows.length >= options.limit) return { capped: false, rows }
    return {
      capped: true,
      resume: {
        offset: Math.max(0, options.offset - merged.length),
        position,
      },
      rows,
    }
  }
  const resume = fetched.length === 1 ? fetched[0]?.resume : undefined
  const rows =
    fetched.length > 1 || fetched.some(({ sample }) => sample)
      ? mergeTransferRows(
          fetched.flatMap(({ rows }) => rows),
          options,
        )
      : (fetched[0]?.rows ?? [])
  return { capped, ...(resume ? { resume } : {}), rows }
}

export declare namespace collate {
  /** One fetched transfer stream. */
  type Stream = {
    /** Whether upstream bounds truncated this stream. */
    capped: boolean
    /** Raw TIDX rows. */
    rows: readonly Record<string, unknown>[]
    /** Ordered raw position and unmatched positional offset after a bounded scan. */
    resume?: Resume | undefined
    /** Whether the rows are an unordered, unpaged capped fallback sample. */
    sample: boolean
  }
  /** Progress retained after a bounded ordered scan. */
  type Resume = {
    /** Filtered rows that a positional continuation still needs to skip. */
    offset: number
    /** Last ordered position safely inspected by the bounded scan. */
    position: readonly [blockNumber: number, logIndex: number]
  }
  /** The page window applied to the merged stream. */
  type Options = {
    /** Maximum merged rows to return. */
    limit: number
    /** Number of globally ordered rows to skip. */
    offset: number
    /** Global transfer ordering. */
    order: 'asc' | 'desc'
  }
  /** The collated page window. */
  type Result = {
    /** Whether an upstream work bound truncated any stream. */
    capped: boolean
    /** The ordered page window rows. */
    rows: readonly Record<string, unknown>[]
    /** Ordered continuation state, absent for unordered capped samples. */
    resume?: Resume | undefined
  }
}

/** Deduplicates and globally orders bounded `from`/`to` transfer streams. */
function mergeTransferRows(
  rows: readonly Record<string, unknown>[],
  options: collate.Options,
): Record<string, unknown>[] {
  const positioned: mergeTransferRows.PositionedRow[] = []
  const seen = new Set<string>()
  for (const row of rows) {
    const blockNumber = Value.toNumber(row['block_num'])
    const logIndex = Value.toNumber(row['log_idx'])
    const transactionHash = Value.toText(row['tx_hash'])
    if (blockNumber === undefined || logIndex === undefined || !transactionHash) continue
    const key = eventKey(transactionHash, logIndex)
    if (seen.has(key)) continue
    seen.add(key)
    positioned.push({ blockNumber, logIndex, row })
  }
  positioned.sort((a, b) => {
    const comparison = a.blockNumber - b.blockNumber || a.logIndex - b.logIndex
    return options.order === 'asc' ? comparison : -comparison
  })
  return positioned.slice(options.offset, options.offset + options.limit).map(({ row }) => row)
}

declare namespace mergeTransferRows {
  type PositionedRow = {
    /** Transfer block number. */
    blockNumber: number
    /** Transfer log index. */
    logIndex: number
    /** Raw TIDX row. */
    row: Record<string, unknown>
  }
}

export declare namespace scan {
  /** Dependencies for {@link scan}. */
  type Deps = {
    /** TIDX query client for the target chain. */
    tidx: Tidx.Client
  }
  /** Options for {@link scan} (everything {@link query.Options} has except `timing`). */
  type Options = Omit<query.Options, 'timing'>
  /** A decoded `Transfer` row plus its `(block_num, log_idx)` cursor. */
  type Item = {
    /** Keyset position of the row. */
    cursor: readonly [blockNumber: number, logIndex: number]
    /** The decoded row before per-surface shaping. */
    data: query.Row
  }
  /** A page of scanned `Transfer` rows. */
  type Page = {
    /** Whether an upstream work bound truncated the page. */
    capped: boolean
    hasMore: boolean
    items: readonly Item[]
    limit: number
    nextCursor: string | null
    /** Last ordered position safely inspected, when a capped scan can resume. */
    resumeCursor?: readonly [blockNumber: number, logIndex: number] | undefined
  }
}

/** Decodes ordinary transfer cursors and capped positional continuations. */
function decodeTransferCursor(token: string | undefined): decodeTransferCursor.Result | undefined {
  if (token === undefined) return undefined
  const position = Cursor.decode(token, ['int', 'int'])
  if (position) return { offset: 0, position: position as [blockNumber: number, logIndex: number] }
  const continuation = Cursor.decode(token, ['int', 'int', 'int'])
  if (!continuation) return undefined
  const [blockNumber, logIndex, offset] = continuation as [number, number, number]
  return { offset: Math.max(0, offset), position: [blockNumber, logIndex] }
}

declare namespace decodeTransferCursor {
  /** Decoded raw position plus any filtered rows still to skip. */
  type Result = {
    offset: number
    position: readonly [blockNumber: number, logIndex: number]
  }
}

/**
 * Builds shared transfer filters. Token-scoped queries target `token_transfers`;
 * unscoped queries constrain the decoded `Transfer` emitter to TIP-20 contracts.
 */
function filterClauses(options: filterClauses.Options): string[] {
  const { fromBlock, fromTimestamp, recipient, sender, toBlock, token, toTimestamp } = options
  const filters =
    token === undefined
      ? [`address >= '${tip20AddressRange[0]}'`, `address < '${tip20AddressRange[1]}'`]
      : [`token = '${token}'`]
  if (sender !== undefined) filters.push(`"from" = '${sender}'`)
  if (recipient !== undefined) filters.push(`"to" = '${recipient}'`)
  if (fromBlock !== undefined) filters.push(`block_num >= ${fromBlock}`)
  if (toBlock !== undefined) filters.push(`block_num <= ${toBlock}`)
  if (fromTimestamp !== undefined) filters.push(`block_timestamp >= '${fromTimestamp}'`)
  if (toTimestamp !== undefined) filters.push(`block_timestamp <= '${toTimestamp}'`)
  return filters
}

declare namespace filterClauses {
  /** The non-pagination filter fields of {@link scan.Options}. */
  type Options = Pick<
    scan.Options,
    'fromBlock' | 'fromTimestamp' | 'recipient' | 'sender' | 'toBlock' | 'toTimestamp' | 'token'
  >
}

function isTip20TransferRow(row: Record<string, unknown>) {
  const address = row['address']
  if (typeof address !== 'string') return false
  const normalized = address.toLowerCase()
  return normalized >= tip20AddressRange[0] && normalized < tip20AddressRange[1]
}

/**
 * Capped total-row count for the `Transfer` feed, sharing {@link query}'s
 * filters (minus pagination) so the count matches the page it annotates. Timed
 * and memoized like {@link query}; the result feeds `meta.totalCount`/`totalCountCapped`
 * when a caller opts in via `include=totalCount`.
 */
export function count(c: Context<App.Environment>, options: count.Options): Promise<count.Result> {
  const { chainId, timing, ...rest } = options
  const store = c.get('store')
  const tidx = c.get('getTidx')(chainId)

  return Timing.time(c, timing, () =>
    Store.memoize(() => countScan({ tidx }, { chainId, ...rest }), {
      key: `transfers:count:v2:${chainId}:${rest.crossToken ? 'fold' : 'raw'}:${rest.token ?? ''}:${rest.eitherSide ?? ''}:${rest.sender ?? ''}:${rest.recipient ?? ''}:${rest.fromBlock ?? ''}:${rest.toBlock ?? ''}:${rest.fromTimestamp ?? ''}:${rest.toTimestamp ?? ''}`,
      store,
      ttl: Ttl.seconds(15),
    }),
  )
}

export declare namespace count {
  /** Options for the capped `Transfer` count: {@link query}'s filters plus `timing`. */
  type Options = Pick<
    query.Options,
    | 'chainId'
    | 'eitherSide'
    | 'fromBlock'
    | 'fromTimestamp'
    | 'recipient'
    | 'sender'
    | 'timing'
    | 'toBlock'
    | 'toTimestamp'
    | 'token'
  > & {
    /** Count folded cross-token transfers instead of raw `Transfer` logs. */
    crossToken?: boolean | undefined
  }
  /** A capped count result: `totalCountCapped` is `true` when the count hit `Schema.countCap`. */
  type Result = { totalCountCapped: boolean; totalCount: number }
}

/**
 * Context-free core of {@link count}: counts token-scoped rows from
 * `token_transfers`, otherwise the decoded `Transfer` CTE. The capped inner
 * scan bounds work while preserving each source's established query engine.
 */
export async function countScan(
  deps: scan.Deps,
  options: countScan.Options,
): Promise<count.Result> {
  const { chainId } = options
  const { tidx } = deps
  const filters = filterClauses(options)
  const tokenScoped = options.token !== undefined
  const address = options.eitherSide

  // Folded count: cross-token transfers collapse several `Transfer` legs into
  // one row, so a raw `count(*)` would overcount. Scan the capped `Transfer`
  // rows, classify each transaction, and subtract the folded-away legs. Bounded
  // by `countCap` like the raw count, so it stays within the indexer's budget.
  if (options.crossToken) {
    const global =
      !tokenScoped &&
      address === undefined &&
      options.sender === undefined &&
      options.recipient === undefined
    if (global) {
      // A capped count promises a conservative lower bound, so aggregate the
      // deduplicated capped sample without the global block sort or 10,000-row response.
      const materializedFilters = [
        `token >= '${tip20AddressRange[0]}'`,
        `token < '${tip20AddressRange[1]}'`,
        ...(options.fromBlock !== undefined ? [`block_num >= ${options.fromBlock}`] : []),
        ...(options.toBlock !== undefined ? [`block_num <= ${options.toBlock}`] : []),
        ...(options.fromTimestamp !== undefined
          ? [`block_timestamp >= '${options.fromTimestamp}'`]
          : []),
        ...(options.toTimestamp !== undefined
          ? [`block_timestamp <= '${options.toTimestamp}'`]
          : []),
      ]
      const result = await tidx.fetch({
        chainId,
        engine: 'clickhouse',
        query: `
          SELECT
            count(*) AS total,
            countIf(
              "to" = '${Addresses.stablecoinDex}'
              OR "from" = '${Addresses.stablecoinDex}'
              OR "to" = '${Addresses.feeManager}'
          ) AS suppressed
          FROM (
            SELECT tx_hash, log_idx, "from", "to"
            FROM token_transfers FINAL
            WHERE ${materializedFilters.join(' AND ')}
            LIMIT ${Schema.countCap}
          ) AS capped
        ` as string,
      })
      const total = Value.toNumber(result.rows[0]?.['total']) ?? 0
      const suppressed = Value.toNumber(result.rows[0]?.['suppressed']) ?? 0
      if (total >= Schema.countCap) {
        return {
          totalCount: Math.max(0, total - suppressed),
          totalCountCapped: true,
        }
      }
      if (suppressed === 0) return { totalCount: total, totalCountCapped: false }
    }

    if (address !== undefined) {
      const materializedFilters = [
        ...(options.token === undefined
          ? [`token >= '${tip20AddressRange[0]}'`, `token < '${tip20AddressRange[1]}'`]
          : [`token = '${options.token}'`]),
        ...(options.sender !== undefined ? [`"from" = '${options.sender}'`] : []),
        ...(options.recipient !== undefined ? [`"to" = '${options.recipient}'`] : []),
        ...(options.fromBlock !== undefined ? [`block_num >= ${options.fromBlock}`] : []),
        ...(options.toBlock !== undefined ? [`block_num <= ${options.toBlock}`] : []),
        ...(options.fromTimestamp !== undefined
          ? [`block_timestamp >= '${options.fromTimestamp}'`]
          : []),
        ...(options.toTimestamp !== undefined
          ? [`block_timestamp <= '${options.toTimestamp}'`]
          : []),
      ]
      const sentFilters = [`"from" = '${address}'`, ...materializedFilters]
      const receivedFilters = [
        `"to" = '${address}'`,
        `"from" != '${address}'`,
        ...materializedFilters,
      ]
      const aggregate = await tidx.fetch({
        chainId,
        engine: 'clickhouse',
        query: `
          SELECT
            'sent' AS side,
            count(*) AS total,
            countIf(
              "to" = '${Addresses.stablecoinDex}'
              OR "from" = '${Addresses.stablecoinDex}'
              OR "to" = '${Addresses.feeManager}'
            ) AS suppressed
          FROM (
            SELECT "from", "to"
            FROM token_transfers
            WHERE ${sentFilters.join(' AND ')}
            LIMIT ${Schema.countCap}
          ) AS sent
          UNION ALL
          SELECT
            'received' AS side,
            count(*) AS total,
            countIf(
              "to" = '${Addresses.stablecoinDex}'
              OR "from" = '${Addresses.stablecoinDex}'
              OR "to" = '${Addresses.feeManager}'
            ) AS suppressed
          FROM (
            SELECT "from", "to"
            FROM token_transfers
            WHERE ${receivedFilters.join(' AND ')}
            LIMIT ${Schema.countCap}
          ) AS received
        ` as string,
      })
      const sent = aggregate.rows.find((row) => Value.toText(row['side']) === 'sent')
      const received = aggregate.rows.find((row) => Value.toText(row['side']) === 'received')
      const sentTotal = Value.toNumber(sent?.['total']) ?? 0
      const receivedTotal = Value.toNumber(received?.['total']) ?? 0
      const total = sentTotal + receivedTotal
      const suppressed =
        (Value.toNumber(sent?.['suppressed']) ?? 0) +
        (Value.toNumber(received?.['suppressed']) ?? 0)
      const capped =
        sentTotal >= Schema.countCap || receivedTotal >= Schema.countCap || total >= Schema.countCap
      if (capped)
        return {
          totalCount: Math.min(Math.max(0, total - suppressed), Schema.countCap),
          totalCountCapped: true,
        }
      if (suppressed === 0) return { totalCount: total, totalCountCapped: false }

      // A below-cap aggregate proves these unsorted scans are complete, so the
      // exact bundle classifier does not need a global sort.
      const select = 'tx_hash, block_num, log_idx, "from", "to", token AS address, amount AS value'
      const fetchRows = async (sideFilters: readonly string[]) => {
        const result = await tidx.fetch({
          chainId,
          engine: 'clickhouse',
          query: `
            SELECT ${select}
            FROM token_transfers
            WHERE ${sideFilters.join(' AND ')}
            LIMIT ${Schema.countCap}
          ` as string,
        })
        return result.rows
      }
      const [sentRows, receivedRows] = await Promise.all([
        fetchRows(sentFilters),
        fetchRows(receivedFilters),
      ])
      const rows = [...sentRows, ...receivedRows]
      const transactionHashes = [...legsByTransaction(rows)]
        .filter(([, legs]) => legs.some(isBundleLeg))
        .map(([transactionHash]) => transactionHash)
      const transactionHashSet = new Set(transactionHashes)
      const candidateRows = rows.filter((row) => {
        const transactionHash = Value.toText(row['tx_hash'])
        return (
          transactionHash !== undefined && transactionHashSet.has(transactionHash.toLowerCase())
        )
      })
      const blocks = candidateRows.flatMap((row) => {
        const block = Value.toNumber(row['block_num'])
        return block === undefined ? [] : [block]
      })
      if (blocks.length === 0 || transactionHashes.length === 0)
        return foldCount(rows, { capped: false })
      // Address predicates omit sibling bundle legs. Load each candidate
      // transaction completely, then suppress only address-matched rows.
      const classification = await tidx.fetch({
        chainId,
        engine: 'clickhouse',
        query: `
          SELECT tx_hash, block_num, log_idx, "from", "to", token AS address, amount AS value
          FROM token_transfers
          WHERE block_num BETWEEN ${Math.min(...blocks)} AND ${Math.max(...blocks)}
            AND tx_hash IN (${transactionHashes.map((hash) => `'${hash}'`).join(', ')})
        ` as string,
      })
      return foldCount(rows, { capped: false, classificationRows: classification.rows })
    }

    // Only one-sided `"from"`/`"to"` equality shapes hit the planner's
    // deterministic top-N kill, so only they take the capped fallback.
    const sideFiltered = options.sender !== undefined || options.recipient !== undefined
    const fetchRows = async () => {
      const where = filters.length > 0 ? `WHERE ${filters.join(' AND ')}` : ''
      const select = `tx_hash, block_num, log_idx, "from", "to", ${tokenScoped ? 'token AS address, amount AS value' : 'address, value'}`
      const table = tokenScoped ? 'token_transfers' : 'Transfer'
      try {
        const result = await tidx.fetch({
          chainId,
          ...(tokenScoped ? { engine: 'clickhouse' } : { signatures: [signature] }),
          query: `
            SELECT ${select}
            FROM ${table}
            ${where}
            ORDER BY block_num, log_idx
            LIMIT ${Schema.countCap}
          ` as string,
        })
        return { capped: false, rows: result.rows, sample: false }
      } catch (error) {
        // Same planner kill as the page's side scan; the folded count is
        // already capped, so a plain capped scan keeps its semantics.
        if (!sideFiltered || !Tidx.isDeterministicError(error)) throw error
        const result = await tidx.fetch({
          chainId,
          ...(tokenScoped ? { engine: 'clickhouse' } : { signatures: [signature] }),
          query: `
            SELECT ${select}
            FROM ${table}
            ${where}
            LIMIT ${Schema.countCap}
          ` as string,
        })
        return { capped: result.rows.length >= Schema.countCap, rows: result.rows, sample: true }
      }
    }
    const fetched = [await fetchRows()]
    const { capped, rows } = collate(fetched, { limit: Schema.countCap, offset: 0, order: 'asc' })
    return foldCount(rows, { capped })
  }

  const countRows = async (additionalFilters: readonly string[] = []) => {
    const countFilters = [...additionalFilters, ...filters]
    const countWhere = countFilters.length > 0 ? `WHERE ${countFilters.join(' AND ')}` : ''
    const result = await tidx.fetch({
      chainId,
      ...(tokenScoped ? { engine: 'clickhouse' } : { signatures: [signature] }),
      query: `
        SELECT count(*) AS total
        FROM (
          SELECT DISTINCT tx_hash, log_idx
          FROM ${tokenScoped ? 'token_transfers' : 'Transfer'}
          ${countWhere}
          LIMIT ${Schema.countCap}
        ) AS capped
      ` as string,
    })
    return Value.toNumber(result.rows[0]?.['total']) ?? 0
  }

  if (address !== undefined) {
    // Keep branches disjoint so self-transfers count once.
    const [sent, received] = await Promise.all([
      countRows([`"from" = '${address}'`]),
      countRows([`"to" = '${address}'`, `"from" != '${address}'`]),
    ])
    const rawCount = sent + received
    return {
      totalCount: Math.min(rawCount, Schema.countCap),
      totalCountCapped:
        sent >= Schema.countCap || received >= Schema.countCap || rawCount >= Schema.countCap,
    }
  }

  const totalCount = await countRows()
  return { totalCountCapped: totalCount >= Schema.countCap, totalCount }
}

export declare namespace countScan {
  /** Options for {@link countScan}: the shared filters plus the target chain. */
  type Options = filterClauses.Options &
    Pick<scan.Options, 'chainId' | 'eitherSide'> & {
      /** Count folded cross-token transfers instead of raw `Transfer` logs. */
      crossToken?: boolean | undefined
    }
}

/**
 * Folds a capped `Transfer` scan into the cross-token count. Classified
 * bundles subtract their folded-away legs; a cap-truncated sample can hold
 * partial bundles that defeat classification, so it subtracts every DEX and
 * fee leg instead, keeping the reported lower bound from overcounting.
 */
export function foldCount(
  rows: readonly Record<string, unknown>[],
  options: foldCount.Options,
): count.Result {
  const { capped, classificationRows = rows } = options
  const rowKeys = new Set<string>()
  for (const row of rows) {
    const transactionHash = Value.toText(row['tx_hash'])
    const logIndex = Value.toNumber(row['log_idx'])
    if (transactionHash && logIndex !== undefined) rowKeys.add(eventKey(transactionHash, logIndex))
  }
  let suppressed = 0
  for (const [transactionHash, legs] of legsByTransaction(capped ? rows : classificationRows)) {
    if (capped) {
      suppressed += legs.filter(isBundleLeg).length
      continue
    }
    const bundle = classifyCrossToken(legs)
    if (bundle)
      suppressed += bundle.suppressLogIndexes.filter((logIndex) =>
        rowKeys.has(eventKey(transactionHash, logIndex)),
      ).length
  }
  return {
    totalCount: rows.length - suppressed,
    totalCountCapped: capped || rows.length >= Schema.countCap,
  }
}

export declare namespace foldCount {
  /** Options for {@link foldCount}. */
  type Options = {
    /** Whether the rows include a cap-truncated fallback sample. */
    capped: boolean
    /** Complete transaction legs used to classify the filtered rows. */
    classificationRows?: readonly Record<string, unknown>[] | undefined
  }
}

/** `'tx1', 'tx2', …` SQL `IN` list of the page's distinct transaction hashes. */
function txInList(rows: memoMap.Options['rows']): string {
  return [...new Set(rows.map((row) => row.transactionHash.toLowerCase()))]
    .map((hash) => `'${hash}'`)
    .join(', ')
}

/**
 * `block_num BETWEEN min AND max` bound covering the page's rows. Event-CTE
 * predicates on `tx_hash` cannot prune `logs` (sorted by block position), so
 * a bare `tx_hash IN (…)` scans to genesis; pairing it with the page's block
 * range keeps the lookup a bounded range read. Callers only invoke the memo
 * lookup for non-empty pages.
 */
function blockBound(rows: memoMap.Options['rows']): string {
  const blocks = rows.map((row) => row.blockNumber)
  return `block_num BETWEEN ${Math.min(...blocks)} AND ${Math.max(...blocks)}`
}

/**
 * Resolves the memo of each of a page's transfers from their matching
 * `TransferWithMemo` events, matched back to rows by {@link transferKey}: the
 * decoded text `memo`, and `attribution` (the resolved MPP service name) when
 * the memo is a known service fingerprint. The fingerprint directory is
 * best-effort: a fetch failure leaves a memo unattributed rather than failing
 * the page.
 */
function memoMap(
  c: Context<App.Environment>,
  options: memoMap.Options,
): Promise<Map<string, { attribution?: string | undefined; memo?: string | undefined }>> {
  const { chainId, rows } = options
  const tidx = c.get('getTidx')(chainId)

  return Timing.time(c, 'transfers_memo', async () => {
    const [memos, fingerprintMap] = await Promise.all([
      tidx.fetch({
        chainId,
        query: `
          SELECT tx_hash, "from", "to", amount, address, memo
          FROM TransferWithMemo
          WHERE ${blockBound(rows)} AND tx_hash IN (${txInList(rows)})
        `,
        signatures: [memoSignature],
      }),
      Mpp.fingerprintMap({ store: c.get('store') }).catch(() => ({})),
    ])

    const memoByRow = new Map<
      string,
      { attribution?: string | undefined; memo?: string | undefined }
    >()
    for (const row of memos.rows) {
      const rawMemo = Value.toText(row['memo']) as Hex.Hex
      const attribution = Mpp.resolve({ fingerprintMap, memo: rawMemo })
      const memo = memoToString(rawMemo)
      const amount = Value.toIntegerString(row['amount'])
      const txHash = Value.toText(row['tx_hash'])
      const from = Value.toText(row['from'])
      const to = Value.toText(row['to'])
      const token = Value.toText(row['address'])
      if ((attribution || memo) && amount && txHash && from && to && token)
        memoByRow.set(transferKey(txHash, from, to, amount, token), { attribution, memo })
    }
    return memoByRow
  })
}

declare namespace memoMap {
  /** Options for the per-page memo lookup. */
  type Options = {
    /** Source chain id. */
    chainId: z.output<typeof Schema.ChainId>
    /** The page rows; only their `transactionHash` and `blockNumber` are read. */
    rows: readonly { blockNumber: number; transactionHash: string }[]
  }
}

/** A transfer leg before amount and token metadata are separated. */
type CrossToken = { address: z.output<typeof Schema.Address>; amount: string }

/** A decoded `Transfer` leg used for cross-token classification. */
type Leg = {
  address: z.output<typeof Schema.Address>
  amount: string
  from: z.output<typeof Schema.Address>
  logIndex: number
  to: z.output<typeof Schema.Address>
}

/** Whether a leg is cross-token bundle plumbing: a DEX leg or the fee leg. */
function isBundleLeg(leg: Leg): boolean {
  return (
    Address.isEqual(leg.to, Addresses.stablecoinDex) ||
    Address.isEqual(leg.from, Addresses.stablecoinDex) ||
    Address.isEqual(leg.to, Addresses.feeManager)
  )
}

/** The folded shape of a detected cross-token transfer, anchored on its delivery leg. */
type CrossTokenBundle = {
  /** Log index of the delivery leg the transfer folds onto. */
  deliveryLogIndex: number
  /** The delivered token and amount (the delivery leg's own event token). */
  destinationToken: CrossToken
  /** The recipient the swapped-out token was forwarded to. */
  recipient: z.output<typeof Schema.Address>
  /** The swapper that funded the transfer through the DEX. */
  sender: z.output<typeof Schema.Address>
  /** The token and amount swapped into the DEX. */
  sourceToken: CrossToken
  /** Log indexes of the legs folded away (the two DEX legs and any fee leg). */
  suppressLogIndexes: number[]
}

/**
 * Classifies a single transaction's `Transfer` legs as a cross-token transfer,
 * or returns `undefined`. A cross-token transfer is the account server's
 * approve+swap+forward multicall: the source token is sent into the DEX, the
 * swapped token comes back to the swapper, and that swapped token is forwarded
 * to a different recipient in the same transaction. The transfer folds onto the
 * delivery (forwarding) leg; the two DEX legs and any fee leg fold away. A swap
 * whose output stays with the swapper (no onward leg) is not a cross-token
 * transfer and is left untouched.
 */
function classifyCrossToken(legs: readonly Leg[]): CrossTokenBundle | undefined {
  const ordered = [...legs].sort((a, b) => a.logIndex - b.logIndex)
  const toDex = ordered.find((leg) => Address.isEqual(leg.to, Addresses.stablecoinDex))
  const fromDex = ordered.find((leg) => Address.isEqual(leg.from, Addresses.stablecoinDex))
  if (!toDex || !fromDex) return undefined
  const swapper = fromDex.to
  const destTokenAddress = fromDex.address
  const delivery = ordered.find(
    (leg) =>
      Address.isEqual(leg.address, destTokenAddress) &&
      Address.isEqual(leg.from, swapper) &&
      !Address.isEqual(leg.to, Addresses.stablecoinDex) &&
      !Address.isEqual(leg.to, swapper) &&
      !Address.isEqual(leg.to, Addresses.feeManager),
  )
  // Require a token change so plain same-token DEX activity is not folded.
  if (!delivery || Address.isEqual(toDex.address, delivery.address)) return undefined
  const suppressLogIndexes = ordered
    .filter((leg) => leg.logIndex !== delivery.logIndex && isBundleLeg(leg))
    .map((leg) => leg.logIndex)
  return {
    deliveryLogIndex: delivery.logIndex,
    destinationToken: { address: delivery.address, amount: delivery.amount },
    recipient: delivery.to,
    sender: swapper,
    sourceToken: { address: toDex.address, amount: toDex.amount },
    suppressLogIndexes,
  }
}

/** Groups decoded `Transfer` rows by lowercased transaction hash. */
function legsByTransaction(rows: readonly Record<string, unknown>[]): Map<string, Leg[]> {
  const legsByTx = new Map<string, Leg[]>()
  for (const row of rows) {
    const txHash = Value.toText(row['tx_hash'])
    const from = Schema.Address.safeParse(row['from'])
    const to = Schema.Address.safeParse(row['to'])
    const address = Schema.Address.safeParse(row['address'])
    const amount = Value.toIntegerString(row['value'])
    const logIndex = Value.toNumber(row['log_idx'])
    if (
      txHash &&
      from.success &&
      to.success &&
      address.success &&
      amount !== undefined &&
      logIndex !== undefined
    ) {
      const legs = legsByTx.get(txHash.toLowerCase()) ?? []
      legs.push({ address: address.data, amount, from: from.data, logIndex, to: to.data })
      legsByTx.set(txHash.toLowerCase(), legs)
    }
  }
  return legsByTx
}

/**
 * Detects cross-token transfers among a page's transactions and returns the
 * folding plan: `fold` maps each delivery leg (by `${txHash}-${logIndex}`) to
 * its folded row (swapper, recipient, swapped-in `sourceToken`, delivered
 * `destinationToken`), and `suppress` is the set of the other bundle legs (the
 * two DEX legs and any fee leg) to drop from the page.
 *
 * Bounded to the page's transactions and block range like {@link memoMap}, so it
 * discovers the sibling DEX legs even when they fall outside the current page.
 *
 * TODO(https://tips.sh/1072): once TIP-1072 lands, cross-token conversions emit
 * a single self-describing `Convert(user, sourceToken, destinationToken,
 * backingToken, amount)` event. Replace this extra `Transfer` lookup and the
 * multi-leg {@link classifyCrossToken} heuristic with a direct `Convert` decode
 * (and drop the folded-count scan in {@link countScan}).
 */
function crossTokenMap(
  c: Context<App.Environment>,
  options: memoMap.Options,
): Promise<{
  fold: Map<
    string,
    {
      destinationToken: CrossToken
      recipient: z.output<typeof Schema.Address>
      sender: z.output<typeof Schema.Address>
      sourceToken: CrossToken
    }
  >
  suppress: Set<string>
}> {
  const { chainId, rows } = options
  const tidx = c.get('getTidx')(chainId)

  return Timing.time(c, 'transfers_cross_token', async () => {
    const result = await tidx.fetch({
      chainId,
      query: `
        SELECT tx_hash, log_idx, "from", "to", address, value
        FROM Transfer
        WHERE ${blockBound(rows)} AND tx_hash IN (${txInList(rows)})
        ORDER BY tx_hash, log_idx
      `,
      signatures: [signature],
    })

    const fold = new Map<
      string,
      {
        destinationToken: CrossToken
        recipient: z.output<typeof Schema.Address>
        sender: z.output<typeof Schema.Address>
        sourceToken: CrossToken
      }
    >()
    const suppress = new Set<string>()
    for (const [txHash, legs] of legsByTransaction(result.rows)) {
      const bundle = classifyCrossToken(legs)
      if (!bundle) continue
      fold.set(eventKey(txHash, bundle.deliveryLogIndex), {
        destinationToken: bundle.destinationToken,
        recipient: bundle.recipient,
        sender: bundle.sender,
        sourceToken: bundle.sourceToken,
      })
      for (const logIndex of bundle.suppressLogIndexes) suppress.add(eventKey(txHash, logIndex))
    }
    return { fold, suppress }
  })
}

/** Composite `${transactionHash}-${logIndex}` key for a single transfer event. */
function eventKey(transactionHash: string, logIndex: number): string {
  return `${transactionHash}-${logIndex}`.toLowerCase()
}

/**
 * Attempts to decode a bytes32 memo into a human-readable string. MPP
 * attribution memos encode a service fingerprint, not text — they are
 * suppressed from the display memo (and resolved to `attribution` instead).
 * Shared with the activity feed's transfer entries.
 */
export function memoToString(memo: Hex.Hex): string | undefined {
  if (Mpp.isAttributionMemo(memo)) return undefined
  try {
    const trimmed = Hex.trimLeft(memo)
    if (trimmed === '0x' || trimmed === '0x0') return undefined
    return Hex.toString(trimmed)
  } catch {
    return undefined
  }
}

/** Composite match key shared by the `Transfer` row and its `TransferWithMemo`. */
function transferKey(
  txHash: string,
  from: string,
  to: string,
  amount: string,
  token: string,
): string {
  return `${txHash}:${from}:${to}:${amount}:${token}`.toLowerCase()
}