import { Hono, type Context, type MiddlewareHandler } from 'hono'
import { Hex } from 'ox'
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'

/** Example block number surfaced in OpenAPI docs. */
const exampleBlockNumber = '1000000'

/** The EVM block tags accepted by the `:block` selector. */
const blockTags = ['latest', 'finalized'] as const

/**
 * ISO 8601 timestamp selector for the `:block` parameter. Shared between the
 * route's param schema (validation + OpenAPI `date-time` format) and
 * {@link parseBlockIdentifier} (runtime dispatch) so the two cannot drift.
 */
const blockTimestamp = z.iso.datetime({ offset: true })

/** Zod schemas owned by the block handlers. */
export namespace schema {
  /** The verbatim JSON-RPC payloads, surfaced under each resource's `rpc` field. */
  export namespace Rpc {
    // TODO(ox/zod): replace with the zod schema for `ox/Block.Rpc` once ox
    // ships a `zod` entrypoint, so the wire format stays in lockstep with
    // ox/viem.
    /** A JSON-RPC block, passed through verbatim. */
    export const Block = z
      .looseObject({
        baseFeePerGas: z
          .nullish(Schema.Quantity)
          .check(z.describe('Base fee per gas for EIP-1559-style fee markets, if present.')),
        difficulty: z
          .optional(Schema.Quantity)
          .check(
            z.describe('Block difficulty value from legacy proof-of-work fields, when present.'),
          ),
        extraData: z
          .optional(Schema.Hex)
          .check(
            z.describe('Extra data bytes included by the block producer.'),
            z.meta({ examples: ['0x00'] }),
          ),
        gasLimit: Schema.Quantity.check(
          z.describe('Maximum gas available for all transactions in this block.'),
        ),
        gasUsed: Schema.Quantity.check(
          z.describe('Total gas used by every transaction in this block.'),
        ),
        hash: z
          .nullable(Schema.Hash)
          .check(z.describe('The block hash, or `null` for a pending block.')),
        logsBloom: z
          .nullish(Schema.Hex)
          .check(
            z.describe('Bloom filter summarizing logs in the block, or `null` while pending.'),
            z.meta({ examples: ['0x00'] }),
          ),
        miner: Schema.Address.check(
          z.describe('The address of the block producer, also called the proposer or miner.'),
        ),
        mixHash: z
          .optional(Schema.Hash)
          .check(z.describe('Consensus mix hash carried in the JSON-RPC block payload.')),
        nonce: z
          .nullish(Schema.Hex)
          .check(
            z.describe('Legacy block nonce, or `null` while pending.'),
            z.meta({ examples: ['0x00'] }),
          ),
        number: z
          .nullable(Schema.Quantity)
          .check(z.describe('The block number, or `null` for a pending block.')),
        parentHash: Schema.Hash.check(z.describe('The hash of the previous block in the chain.')),
        receiptsRoot: Schema.Hash.check(
          z.describe('Root hash of the receipts trie for this block.'),
        ),
        sha3Uncles: z
          .optional(Schema.Hash)
          .check(z.describe('Keccak-256 hash of the uncle blocks list.')),
        size: Schema.Quantity.check(z.describe('Size of the block in bytes.')),
        stateRoot: Schema.Hash.check(
          z.describe('Root hash of the world-state trie after this block.'),
        ),
        timestamp: Schema.Quantity.check(z.describe('Block timestamp as Unix seconds.')),
        totalDifficulty: z
          .optional(Schema.Quantity)
          .check(z.describe('Total cumulative difficulty through this block, when present.')),
        transactions: z
          .array(Schema.Hash)
          .check(z.describe('Transaction hashes included in this ordered batch.')),
        transactionsRoot: Schema.Hash.check(
          z.describe('Root hash of the transactions trie for this block.'),
        ),
        uncles: z
          .optional(z.array(Schema.Hash))
          .check(z.describe('Hashes of uncle blocks included in this block, when present.')),
        withdrawals: z
          .optional(z.array(z.looseObject({})))
          .check(z.describe('EIP-4895 withdrawals included in this block, when present.')),
        withdrawalsRoot: z
          .optional(Schema.Hash)
          .check(z.describe('Root hash of the withdrawals trie, when present.')),
      })
      .check(z.describe('The block exactly as returned by JSON-RPC.'))
  }

  const rpcBlock = OpenApi.component(Rpc.Block, 'RpcBlock')

  /**
   * Metadata for a block: the verbatim JSON-RPC block (always present). Held in
   * a `meta` envelope so future opt-in `include` resources (e.g. embedded
   * transactions) can layer in without breaking the response shape.
   */
  export const BlockMeta = OpenApi.component(
    Schema.describe(
      z.object({
        rpc: rpcBlock.check(z.describe('The original JSON-RPC block payload.')),
      }),
      'Metadata for this block: the original JSON-RPC payload plus any future `include` resources.',
    ),
    'BlockMeta',
  )

  /**
   * A humanized block: decimal-string amounts, ISO 8601 timestamps, and plain
   * numbers for block fields. The verbatim JSON-RPC payload is always
   * available under `meta.rpc`. Transactions are omitted from the response;
   * clients filter `GET /v1/transactions` to this block number.
   */
  export const Block = z
    .object({
      baseFeePerGas: z
        .nullable(Schema.DecimalString)
        .check(
          z.describe(
            'Base fee per gas for EIP-1559-style fee markets, or `null` when unsupported.',
          ),
        ),
      gasLimit: z
        .number()
        .check(
          z.int(),
          z.nonnegative(),
          z.describe('Maximum gas available for all transactions in this block.'),
          z.meta({ examples: [21000] }),
        ),
      gasUsed: z
        .number()
        .check(
          z.int(),
          z.nonnegative(),
          z.describe('Total gas used by transactions in this block.'),
          z.meta({ examples: [21000] }),
        ),
      hash: Schema.Hash.check(
        z.describe(
          'A 32-byte hash, `0x`-prefixed and returned in lowercase, that identifies this block.',
        ),
      ),
      id: Schema.Hash.check(
        z.describe('Stable resource ID for this API response; it is the block hash.'),
      ),
      meta: BlockMeta.check(
        z.describe('The original JSON-RPC payload plus any resources requested with `include`.'),
      ),
      miner: Schema.Address.check(
        z.describe('The address of the block producer, also called the proposer or miner.'),
      ),
      number: z
        .number()
        .check(
          z.int(),
          z.nonnegative(),
          z.describe('The block height, starting from genesis block 0.'),
          z.meta({ examples: [23456789] }),
        ),
      parentHash: Schema.Hash.check(z.describe('The hash of the previous block in the chain.')),
      size: z
        .number()
        .check(
          z.int(),
          z.nonnegative(),
          z.describe('Size of the block in bytes.'),
          z.meta({ examples: [1024] }),
        ),
      timestamp: z.iso
        .datetime()
        .check(
          z.describe('Block timestamp formatted as ISO 8601.'),
          z.meta({ examples: ['2024-01-01T00:00:00Z'] }),
        ),
      transactionCount: z
        .number()
        .check(
          z.int(),
          z.nonnegative(),
          z.describe('Number of transactions included in this block.'),
          z.meta({ examples: [25] }),
        ),
    })
    .check(
      z.describe(
        'A block formatted for API clients, with number, timestamp, gas usage, producer, and original JSON-RPC data under `meta.rpc`.',
      ),
    )

  const block = OpenApi.component(Block, 'Block')

  /**
   * A lightweight block row, returned by the `/blocks` list endpoint. Full
   * fields (parentHash, miner, gasUsed/gasLimit, …) would require one RPC
   * roundtrip per row; the list endpoint stays cheap and clients fetch
   * `/blocks/:block` for individual blocks they care about. The block hash is
   * selected from the index so `id` stays tied to the resolved block identity.
   */
  export const BlockSummary = z
    .object({
      hash: Schema.Hash.check(
        z.describe(
          'A 32-byte hash, `0x`-prefixed and returned in lowercase, that identifies this block.',
        ),
      ),
      id: Schema.Hash.check(
        z.describe('Stable resource ID for this API response; it is the block hash.'),
      ),
      number: z
        .number()
        .check(
          z.int(),
          z.nonnegative(),
          z.describe('The block height, starting from genesis block 0.'),
          z.meta({ examples: [23456789] }),
        ),
      timestamp: z.iso
        .datetime()
        .check(
          z.describe('Block timestamp formatted as ISO 8601.'),
          z.meta({ examples: ['2024-01-01T00:00:00Z'] }),
        ),
      transactionCount: z
        .number()
        .check(
          z.int(),
          z.nonnegative(),
          z.describe('Number of transactions included in this block.'),
          z.meta({ examples: [25] }),
        ),
    })
    .check(z.describe('A lightweight block summary for list responses.'))

  const blockSummary = OpenApi.component(BlockSummary, 'BlockSummary')

  /** Schemas for the getBlocks operation. */
  export namespace getBlocks {
    /** Query parameters for `GET /blocks`. */
    export const Query = z
      .strictObject({
        chainId: Schema.ChainIdQuery,
        cursor: Schema.Cursor,
        limit: Schema.Limit,
        order: Schema.Order,
        page: Schema.Page,
      })
      .check(...Schema.pageChecks(), z.describe('Query parameters for listing blocks.'))

    /** Page of lightweight block rows, including empty blocks. */
    export const Response = OpenApi.component(
      z
        .object({
          data: z.array(blockSummary).check(z.describe('Blocks on this page.')),
          nextCursor: Schema.NextCursor,
        })
        .check(z.describe('A page of blocks ordered by block number.')),
      'BlockList',
    )
  }

  /** Schemas for the getBlock operation. */
  export namespace getBlock {
    /**
     * Path parameter for block requests. Resolves a single block from any of:
     * a block tag (`latest`, `finalized`); a decimal or `0x`-prefixed hex block
     * number; a `0x`-prefixed 32-byte block hash; or an ISO 8601 timestamp
     * (`2023-11-14T22:13:20Z`), which resolves to the first block at or after
     * that instant.
     */
    export const Params = z
      .object({
        block: z
          .union([
            // Block tag (`latest`, `finalized`).
            z.enum(blockTags),
            // Block number (decimal or hex) or a 0x-prefixed 32-byte hash.
            z.string().check(z.regex(/^(0x[0-9a-fA-F]+|[0-9]+)$/)),
            // ISO 8601 timestamp, resolved to the first block at or after it.
            blockTimestamp,
          ])
          .check(
            z.describe(
              'Block selector: `latest`, `finalized`, a decimal or hex block number, a 32-byte block hash, or an ISO 8601 timestamp.',
            ),
            z.meta({ examples: [exampleBlockNumber, 'latest', '2023-11-14T22:13:20Z'] }),
          ),
      })
      .check(z.describe('Path parameter used to select a block.'))

    /** Query parameters for `GET /blocks/:block`. */
    export const Query = z
      .strictObject({
        chainId: Schema.ChainIdQuery,
      })
      .check(z.describe('Query parameters for looking up a block.'))

    /** A humanized block, with the verbatim JSON-RPC payload under `meta.rpc`. */
    export const Response = block
  }

  /**
   * Schemas for the legacy `getBlockByTimestamp` compatibility operation
   * (`GET /chains/:chainId/blocks/by-timestamp`). Hidden from the docs; kept to
   * keep pre-existing clients of the prior Tempo API working unchanged.
   */
  export namespace getBlockByTimestamp {
    /** Path parameters: the numeric Tempo chain id. */
    export const Params = z
      .object({ chainId: Schema.ChainId })
      .check(z.describe('Path parameters for the legacy block-by-timestamp lookup.'))

    /** Query parameters: a required unix timestamp and the search direction. */
    export const Query = z
      .strictObject({
        closest: z
          ._default(z.enum(['before', 'after']), 'before')
          .check(
            z.describe(
              'Choose whether the closest block should be at or before the timestamp, or at or after it.',
            ),
          ),
        timestamp: z.coerce
          .number()
          .check(z.int(), z.nonnegative(), z.describe('Unix timestamp in seconds.')),
      })
      .check(z.describe('Query parameters for the legacy block-by-timestamp lookup.'))

    /** The legacy response envelope: `{ block: { blockNumber, blockTimestamp } }`. */
    export const Response = z
      .object({
        block: z
          .object({
            blockNumber: z
              .number()
              .check(
                z.int(),
                z.nonnegative(),
                z.describe('The block height, starting from genesis block 0.'),
              ),
            blockTimestamp: z
              .number()
              .check(z.int(), z.nonnegative(), z.describe('Block timestamp as Unix seconds.')),
          })
          .check(z.describe('The closest indexed block found for the requested timestamp.')),
      })
      .check(z.describe('The closest indexed block for the supplied timestamp.'))
  }
}

/** Window, in seconds, searched on either side of a `by-timestamp` lookup. */
const byTimestampWindowSeconds = 60

/** Creates block handlers. */
export function blocks() {
  // Only cursor pages of the block list are cacheable: they are anchored below
  // the head and effectively immutable. The head page is a point-in-time read
  // of the live chain tip, so it skips the cache entirely and carries no
  // `Cache-Control` header at all.
  const cacheBlocksList = Cache.response<App.Environment>({
    cacheControl: Cache.policies.immutable,
    name: 'tempo-api:blocks:v1',
    key: (c) => Cache.urlKey(c, schema.getBlocks.Query),
  })
  const cacheBlocksListIfCursor: MiddlewareHandler<App.Environment> = (c, next) =>
    c.req.query('cursor') === undefined ? next() : cacheBlocksList(c, next)

  return new Hono<App.Environment>()
    .get(
      '/v1/blocks',
      Auth.policy({ apiKey: { scopes: ['data:read'] }, mpp: true, public: true }),
      OpenApi.validate('query', schema.getBlocks.Query, {
        code: 'query_invalid',
        message: 'Invalid query parameters',
      }),
      OpenApi.describeRoute({
        description: 'List the ordered batches of transactions that make up Tempo.',
        operationId: 'getBlocks',
        responses: OpenApi.responses({
          errors: {
            400: { codes: ['chain_id_invalid', 'chain_id_unsupported', 'query_invalid'] },
            502: 'Could not read indexed block data from TIDX.',
          },
          success: { description: 'A page of block summaries.', schema: schema.getBlocks.Response },
        }),
        summary: 'List blocks',
        tags: ['Blocks'],
      }),
      cacheBlocksListIfCursor,
      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 listBlocks(c, {
            chainId,
            cursor: query.cursor,
            limit: query.limit,
            order: query.order,
            page: query.page,
          })
          return c.json(Response.validated(schema.getBlocks.Response, page), 200)
        } catch (cause) {
          return Response.upstream(c, cause, { page: query.page })
        }
      },
    )
    .get(
      '/v1/blocks/:block',
      Auth.policy({ apiKey: { scopes: ['data:read'] }, mpp: true, public: true }),
      OpenApi.validate('param', schema.getBlock.Params, {
        code: 'block_invalid',
        message: 'Invalid block identifier',
      }),
      OpenApi.validate('query', schema.getBlock.Query, {
        code: 'query_invalid',
        message: 'Invalid query parameters',
      }),
      OpenApi.describeRoute({
        description: 'Get one block by tag, number, hash, or timestamp selector.',
        operationId: 'getBlock',
        responses: OpenApi.responses({
          errors: {
            400: {
              codes: ['block_invalid', 'chain_id_invalid', 'chain_id_unsupported', 'query_invalid'],
            },
            404: {
              description:
                'No block matched the selector, or no indexed block exists at or after the timestamp.',
              codes: ['block_not_found'],
            },
            502: 'Could not read block data from the upstream JSON-RPC node or TIDX.',
          },
          success: {
            description:
              'A single block with number, timestamp, gas usage, producer, and original JSON-RPC data.',
            schema: schema.getBlock.Response,
          },
        }),
        summary: 'Get a block by selector',
        tags: ['Blocks'],
      }),
      Cache.response<App.Environment>({
        // Default to no-store so a 404 cannot mask the eventual mined block.
        // The handler upgrades a mined block to `immutable` below.
        cacheControl: Cache.policies.noStore,
        name: 'tempo-api:blocks:v1',
        key: (c) => Cache.urlKey(c, schema.getBlock.Query),
      }),
      async (c) => {
        if (Auth.narrowAccess) return Auth.paidAccessError(c)
        if (OpenApi.narrowValidation)
          return OpenApi.validationError(c, {
            code: 'block_invalid',
            message: 'Invalid request parameters',
          })
        if (OpenApi.narrowValidation)
          return OpenApi.validationError(c, {
            code: 'query_invalid',
            message: 'Invalid request parameters',
          })

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

        const identifier = parseBlockIdentifier(block)
        if (!identifier)
          return Response.error(c, {
            code: 'block_invalid',
            message: 'Invalid block identifier',
            status: 400,
          })

        try {
          // A timestamp selector resolves to a concrete block number via the
          // indexer first (the `blocks` table records every block, empty ones
          // included, so this is exact rather than rounding to the next
          // non-empty block). A distinct 404 separates "no indexed block at or
          // after this instant" from a missing block by number/hash.
          let timestampBlock: number | undefined
          if (identifier.kind === 'timestamp') {
            const tidx = c.get('getTidx')(chainId)
            const store = c.get('store')
            const iso = identifier.iso
            timestampBlock = await Timing.time(c, 'block_by_timestamp', () =>
              Store.memoize(
                async () => {
                  const result = await tidx.fetch({
                    chainId,
                    query: `
                      SELECT num
                      FROM blocks
                      WHERE timestamp >= '${iso}'
                      ORDER BY num ASC
                      LIMIT 1
                    ` as string,
                  })
                  const row = result.rows[0]
                  if (!row) return undefined
                  return Value.toNumber(row['num'])
                },
                {
                  key: `blocks:v1:${chainId}:by_timestamp:${iso}`,
                  store,
                  ttl: Ttl.seconds(30),
                },
              ),
            )
            if (timestampBlock === undefined)
              return Response.error(c, {
                code: 'block_not_found',
                message: 'No indexed block at or after this timestamp',
                status: 404,
              })
          }

          // Dispatch by selector shape: a 32-byte hash uses
          // `eth_getBlockByHash`; tags, numbers, and the timestamp-resolved
          // number all go through `eth_getBlockByNumber`. The moving tags
          // (`latest`/`finalized`) are deliberately never cached — not even
          // briefly — because they are point-in-time reads of the live tip.
          const rpc = await Timing.time(c, 'block', () => {
            if (identifier.kind === 'hash')
              return client.request({
                method: 'eth_getBlockByHash',
                params: [identifier.hash, false],
              })
            const block =
              identifier.kind === 'tag'
                ? identifier.tag
                : identifier.kind === 'number'
                  ? Hex.fromNumber(identifier.number)
                  : Hex.fromNumber(BigInt(timestampBlock!))
            return client.request({ method: 'eth_getBlockByNumber', params: [block, false] })
          })
          if (!rpc)
            return Response.error(c, {
              code: 'block_not_found',
              message: 'Block not found',
              status: 404,
            })

          // Cache policy by selector: a specific block (number or hash) is
          // immutable; a timestamp can land on the moving head or a historic
          // block, so `state` is the safe middle; the moving tags (`latest`,
          // `finalized`) keep the route's no-store default.
          if (identifier.kind === 'number' || identifier.kind === 'hash')
            Cache.setPolicy(c, Cache.policies.immutable)
          else if (identifier.kind === 'timestamp') Cache.setPolicy(c, Cache.policies.state)

          const data = humanizeBlock(Response.validated(schema.Rpc.Block, rpc))
          return c.json(Response.validated(schema.Block, data), 200)
        } catch (cause) {
          return Response.upstream(c, cause)
        }
      },
    )
}

/**
 * Mounts the hidden legacy `GET /chains/:chainId/blocks/by-timestamp`
 * compatibility route at the app root. The prior Tempo API exposed this shape
 * (chain id in the path, a required unix `timestamp`, and a `closest` direction)
 * and returned `{ block: { blockNumber, blockTimestamp } }`; this preserves it
 * so existing clients keep working while the canonical surface is
 * `GET /v1/blocks/:block` with an ISO 8601 selector.
 *
 * Like the original it searches a ±{@link byTimestampWindowSeconds}s window and
 * returns `404` when no indexed block falls within it. A local supported-chain
 * check keeps standalone composition safe.
 */
export function blocksByTimestampCompat() {
  return new Hono<App.Environment>().get(
    '/chains/:chainId/blocks/by-timestamp',
    Auth.policy({ apiKey: { scopes: ['data:read'] }, mpp: true, public: true }),
    OpenApi.validate('param', schema.getBlockByTimestamp.Params, {
      code: 'chain_id_invalid',
      message: 'Invalid chain id',
    }),
    OpenApi.validate('query', schema.getBlockByTimestamp.Query, {
      code: 'query_invalid',
      message: 'Invalid query parameters',
    }),
    OpenApi.describeRoute({
      hide: true,
      operationId: 'getBlockByTimestamp',
      responses: OpenApi.responses({
        errors: {
          400: { codes: ['chain_id_invalid', 'chain_id_unsupported', 'query_invalid'] },
          404: {
            description: 'No indexed block was found within the timestamp search window.',
            codes: ['block_not_found'],
          },
          502: 'Could not read indexed block data from TIDX.',
        },
        success: {
          description: 'The closest indexed block for the supplied timestamp.',
          schema: schema.getBlockByTimestamp.Response,
        },
      }),
      summary: 'Get a legacy block by timestamp',
      tags: ['Blocks'],
    }),
    Cache.response<App.Environment>({
      cacheControl: Cache.policies.state,
      name: 'tempo-api:blocks:v1',
      key: (c) => Cache.urlKey(c, schema.getBlockByTimestamp.Query),
    }),
    async (c) => {
      if (Auth.narrowAccess) return Auth.paidAccessError(c)
      if (OpenApi.narrowValidation)
        return OpenApi.validationError(c, {
          code: 'query_invalid',
          message: 'Invalid request parameters',
        })

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

      // Keep this route safe when mounted without the shared data app.
      const supportedChainIds = c.get('supportedChainIds')
      if (!supportedChainIds.has(chainId))
        return Response.unsupportedChainId(c, chainId, supportedChainIds)

      try {
        const block = await findBlockByTimestamp(c, { chainId, closest, timestamp })
        if (!block)
          return Response.error(c, {
            code: 'block_not_found',
            message: 'No indexed block matches the supplied timestamp',
            status: 404,
          })
        return c.json(Response.validated(schema.getBlockByTimestamp.Response, { block }), 200)
      } catch (cause) {
        return Response.upstream(c, cause)
      }
    },
  )
}

/**
 * Resolves the indexed block closest to `timestamp` within a
 * ±{@link byTimestampWindowSeconds}s window, in the requested direction. Reads
 * the `blocks` table (every block, empty ones included) so the match is exact.
 */
function findBlockByTimestamp(c: Context<App.Environment>, options: findBlockByTimestamp.Options) {
  const { chainId, closest, timestamp } = options
  const store = c.get('store')
  const tidx = c.get('getTidx')(chainId)
  const at = new Date(timestamp * 1000).toISOString()
  const bound = new Date(
    (closest === 'before'
      ? timestamp - byTimestampWindowSeconds
      : timestamp + byTimestampWindowSeconds) * 1000,
  ).toISOString()

  return Timing.time(c, 'block_by_timestamp', () =>
    Store.memoize(
      async () => {
        // `before`: latest block in [at - window, at]; `after`: earliest block
        // in [at, at + window]. Order on timestamp then `num` so ties resolve to
        // the block nearest the requested instant.
        const where =
          closest === 'before'
            ? `timestamp <= '${at}' AND timestamp >= '${bound}'`
            : `timestamp >= '${at}' AND timestamp <= '${bound}'`
        const direction = closest === 'before' ? 'DESC' : 'ASC'

        const result = await tidx.fetch({
          chainId,
          query: `
            SELECT num, timestamp
            FROM blocks
            WHERE ${where}
            ORDER BY timestamp ${direction}, num ${direction}
            LIMIT 1
          ` as string,
        })
        const row = result.rows[0]
        if (!row) return undefined
        const blockNumber = Value.toNumber(row['num'])
        const iso = Value.toIsoDateTime(row['timestamp'])
        if (blockNumber === undefined || iso === undefined) return undefined
        return { blockNumber, blockTimestamp: Math.floor(Date.parse(iso) / 1000) }
      },
      {
        key: `blocks:v1:${chainId}:by_timestamp_compat:${closest}:${timestamp}`,
        store,
        ttl: Ttl.seconds(30),
      },
    ),
  )
}

export declare namespace findBlockByTimestamp {
  type Options = {
    chainId: z.output<typeof Schema.ChainId>
    closest: 'before' | 'after'
    timestamp: number
  }
}

/**
 * Parses a `:block` path parameter, dispatching by shape. Returns the
 * discriminated identifier, or `undefined` when the input matches no known
 * shape (the schema should have caught this earlier; this guards the type at
 * the use site).
 *
 * Selectors are unambiguous by shape: a block tag is a fixed keyword; the
 * 66-char `0x`-prefixed length identifies a 32-byte hash; any other
 * `0x`-prefixed hex (or plain decimal) is a block number; and an ISO 8601
 * timestamp (validated with the same {@link blockTimestamp} schema as the
 * route) is resolved to its UTC instant.
 */
export function parseBlockIdentifier(value: string): parseBlockIdentifier.Result | undefined {
  if ((blockTags as readonly string[]).includes(value))
    return { kind: 'tag', tag: value as parseBlockIdentifier.BlockTag }
  if (/^0x[0-9a-fA-F]{64}$/.test(value))
    return { kind: 'hash', hash: value.toLowerCase() as Hex.Hex }
  if (/^0x[0-9a-fA-F]+$/.test(value)) return { kind: 'number', number: BigInt(value) }
  if (/^[0-9]+$/.test(value)) return { kind: 'number', number: BigInt(value) }
  if (blockTimestamp.safeParse(value).success)
    return { kind: 'timestamp', iso: new Date(Date.parse(value)).toISOString() }
  return undefined
}

export declare namespace parseBlockIdentifier {
  /** An EVM block tag accepted by the `:block` selector. */
  type BlockTag = (typeof blockTags)[number]

  /** A parsed block identifier, dispatching by shape. */
  type Result =
    | { kind: 'tag'; tag: BlockTag }
    | { kind: 'hash'; hash: Hex.Hex }
    | { kind: 'number'; number: bigint }
    | { kind: 'timestamp'; iso: string }
}

/**
 * Lists blocks from the indexer. Reads every block from the `blocks` table
 * (so empty blocks are included) and resolves transaction counts via a
 * second narrow `txs` query keyed to the page's block numbers.
 */
export function listBlocks(c: Context<App.Environment>, options: listBlocks.Options) {
  const { chainId, limit, order } = options
  const store = c.get('store')
  const tidx = c.get('getTidx')(chainId)
  const sortDirection = order === 'asc' ? 'ASC' : 'DESC'
  // Single-column keyset cursor on `num` — block timestamps aren't strictly
  // monotonic on every chain, so the block number alone is the stable order.
  const cursor = options.cursor ? Cursor.decode(options.cursor, ['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, 'blocks', () =>
    Store.memoize(
      async () => {
        const filters: string[] = []
        if (cursor !== undefined)
          filters.push(
            Cursor.keyset([{ literal: Cursor.literal(cursor[0]!, 'int'), name: 'num', order }]),
          )
        const where = filters.length > 0 ? `WHERE ${filters.join(' AND ')}` : ''

        const blocks = await tidx.fetch({
          chainId,
          query: `
            SELECT hash, num, timestamp
            FROM blocks
            ${where}
            ORDER BY num ${sortDirection}
            LIMIT ${limit + 1}${offset !== undefined ? ` OFFSET ${offset}` : ''}
          ` as string,
        })

        const page = Cursor.paginate({
          rows: blocks.rows,
          limit,
          key: (row) => {
            const block = Value.toNumber(row['num'])
            return block !== undefined ? [block] : undefined
          },
        })

        // Pull transaction counts in a single follow-up query scoped to the
        // page's block numbers so empty blocks correctly report `0`.
        const numbers: number[] = []
        for (const row of page.rows) {
          const block = Value.toNumber(row['num'])
          if (block !== undefined) numbers.push(block)
        }
        const countsByBlock = new Map<number, number>()
        if (numbers.length > 0) {
          const counts = await tidx.fetch({
            chainId,
            engine: 'clickhouse',
            query: `
              SELECT block_num, count() AS transaction_count
              FROM txs
              WHERE block_num IN (${numbers.join(', ')})
              GROUP BY block_num
            ` as string,
          })
          for (const row of counts.rows) {
            const block = Value.toNumber(row['block_num'])
            const count = Value.toNumber(row['transaction_count'])
            if (block !== undefined && count !== undefined) countsByBlock.set(block, count)
          }
        }

        const data: z.output<typeof schema.BlockSummary>[] = []
        for (const row of page.rows) {
          const hash = Value.toText(row['hash'])?.toLowerCase() as Hex.Hex | undefined
          const number = Value.toNumber(row['num'])
          const timestamp = Value.toIsoDateTime(row['timestamp'])
          if (hash === undefined || number === undefined || timestamp === undefined) continue
          data.push({
            hash,
            id: hash,
            number,
            timestamp,
            transactionCount: countsByBlock.get(number) ?? 0,
          })
        }

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

export declare namespace listBlocks {
  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'
  }
}

/**
 * Humanizes a JSON-RPC block: converts hex quantities to decimal strings, hex
 * block fields to numbers, and timestamps to ISO 8601, while preserving the
 * verbatim payload under `meta.rpc`. Transactions are surfaced as a count;
 * full transaction objects live on `GET /transactions` (equal `blockNumber.*` bounds).
 */
function humanizeBlock(rpc: z.output<typeof schema.Rpc.Block>): z.output<typeof schema.Block> {
  // A non-pending block always has `number`, `hash`, and `timestamp`. The
  // pending case is uninteresting for this surface (the latest endpoint asks
  // for `'latest'`, which is mined), so these are treated as required after
  // the upstream call returns a non-null result.
  return {
    baseFeePerGas: Value.fromHex(rpc.baseFeePerGas) ?? null,
    gasLimit: Value.hexToNumber(rpc.gasLimit) ?? 0,
    gasUsed: Value.hexToNumber(rpc.gasUsed) ?? 0,
    hash: rpc.hash as Hex.Hex,
    id: rpc.hash as Hex.Hex,
    meta: { rpc },
    miner: rpc.miner,
    number: Value.hexToNumber(rpc.number) ?? 0,
    parentHash: rpc.parentHash,
    size: Value.hexToNumber(rpc.size) ?? 0,
    timestamp: Value.hexSecondsToIso(rpc.timestamp) ?? new Date(0).toISOString(),
    transactionCount: rpc.transactions.length,
  }
}