import { Hono, type Context, type TypedResponse } from 'hono'
import { ZoneRpcAuthentication } from 'ox/tempo'
import * as z from 'zod/mini'

import type * as App from '../../../App.js'
import * as Auth from '../../../internal/Auth.js'
import * as Cache from '../../../internal/Cache.js'
import * as OpenApi from '../../../internal/OpenApi.js'
import * as Path from '../../../internal/Path.js'
import * as Response from '../../../internal/Response.js'
import * as Schema from '../../../internal/Schema.js'
import * as Tidx from '../../../internal/Tidx.js'
import * as Timing from '../../../internal/Timing.js'

// Request headers that must not be forwarded upstream: hop-by-hop headers, the
// inbound host, and any caller auth so the consumer's Tempo API credentials never
// reach the indexer (the proxy injects its own TIDX basic auth instead).
const strippedHeaders = new Set([
  'authorization',
  'connection',
  'cookie',
  'host',
  'keep-alive',
  'proxy-authorization',
  'te',
  'tempo-api-key',
  'trailer',
  'transfer-encoding',
  'upgrade',
  'x-api-key',
  ZoneRpcAuthentication.headerName.toLowerCase(),
])

/** Never reached; carries typed passthrough response variants into Hono client inference. */
const narrowPassthrough = false as boolean

/** Zod schemas owned by the indexer passthrough. */
export namespace schema {
  /** Structured result returned by the indexer `query` endpoint. */
  export const Response = z
    .object({
      ok: z
        .boolean()
        .check(
          z.describe('Whether the SQL query completed successfully.'),
          z.meta({ examples: [true] }),
        ),
      columns: z
        .array(z.string())
        .check(z.describe('Column names returned by the query, in row-value order.')),
      rows: z
        .array(z.array(z.unknown()))
        .check(
          z.describe(
            'Query results. Each row is an array of values in the same order as `columns`.',
          ),
        ),
      row_count: z
        .number()
        .check(
          z.int(),
          z.nonnegative(),
          z.describe('Number of rows returned in this response.'),
          z.meta({ examples: [1] }),
        ),
      engine: z
        .optional(z.enum(['postgres', 'clickhouse', 'tiered']))
        .check(
          z.describe('Indexer engine that ran the query.'),
          z.meta({ examples: ['postgres'] }),
        ),
      query_time_ms: z
        .optional(z.number())
        .check(
          z.describe('Server-side query execution time in milliseconds.'),
          z.meta({ examples: [12] }),
        ),
    })
    .check(z.describe('Structured result from a Tempo indexer SQL query.'))

  /** Error envelope returned by the upstream indexer. */
  export const Error = z
    .object({
      ok: z.literal(false).check(z.describe('Always `false` when the indexer returns an error.')),
      error: z.string().check(z.describe('Human-readable error message from the indexer.')),
    })
    .check(z.describe('Error response returned by the upstream Tempo indexer.'))
}

/** Creates the raw indexer (TIDX) passthrough handler. */
export function indexer() {
  const queryCache = Cache.response<App.Environment>({
    cacheControl: Cache.policies.feed,
    name: 'tempo-api:indexer:v1',
    key: (c) => {
      const url = new URL(c.req.url)
      if (!url.searchParams.has('chainId') && !url.searchParams.has('chain_id'))
        url.searchParams.set('chainId', String(c.get('chainId')))
      url.searchParams.sort()
      return url.toString()
    },
  })

  return new Hono<App.Environment>().get(
    '/v1/indexer/query',
    Auth.policy({ apiKey: { scopes: ['indexer:query'] }, mpp: true }),
    OpenApi.describeRoute({
      description:
        'Runs a read-only SQL-style query against Tempo’s indexed chain data, with optional live streaming as new blocks arrive.',
      operationId: 'indexerQuery',
      parameters: [
        {
          description:
            'Read-only SQL query to run against Tempo’s indexed chain data. Use `SELECT` statements only.',
          example: 'SELECT num, hash, timestamp FROM blocks ORDER BY num DESC LIMIT 10',
          in: 'query',
          name: 'sql',
          required: true,
          schema: { type: 'string' },
        },
        {
          description:
            'Which Tempo network to query. Use the alias `mainnet` or `testnet`, or a numeric chain id (mainnet is `4217`).',
          example: 4217,
          in: 'query',
          name: 'chainId',
          required: true,
          schema: { minimum: 1, type: 'integer' },
        },
        {
          description:
            'ABI event signature to expose as a named SQL CTE, so you can query decoded event fields directly. Repeat this parameter to add multiple signatures.',
          example: 'Transfer(address indexed from, address indexed to, uint256 value)',
          explode: true,
          in: 'query',
          name: 'signature',
          required: false,
          schema: { items: { type: 'string' }, type: 'array' },
          style: 'form',
        },
        {
          description:
            'Choose the indexer engine yourself. By default the indexer routes automatically; `clickhouse` cannot be used with `live=true`.',
          example: 'postgres',
          in: 'query',
          name: 'engine',
          required: false,
          schema: { enum: ['postgres', 'clickhouse'], type: 'string' },
        },
        {
          description:
            'Stream results as Server-Sent Events and re-run the query on every new block. This cannot be combined with `engine=clickhouse`.',
          example: false,
          in: 'query',
          name: 'live',
          required: false,
          schema: { default: false, type: 'boolean' },
        },
        {
          description:
            'Maximum number of rows to return. The server clamps this to the range 1 through 10,000.',
          example: 10,
          in: 'query',
          name: 'limit',
          required: false,
          schema: { default: 10000, maximum: 10000, minimum: 1, type: 'integer' },
        },
        {
          description:
            'Per-query timeout in milliseconds. The server clamps this to the range 100 through 30,000.',
          example: 5000,
          in: 'query',
          name: 'timeout_ms',
          required: false,
          schema: { default: 5000, maximum: 30000, minimum: 100, type: 'integer' },
        },
      ],
      responses: {
        200: {
          content: {
            'application/json': {
              schema: OpenApi.resolver(schema.Response),
              example: {
                ok: true,
                columns: ['num', 'hash', 'timestamp'],
                rows: [[1234567, '0x515801d7f9a5ac705e793e85904c9c69b3f1694b465cc8fb6ba3f0298dc82665', 1718668800]], // prettier-ignore
                row_count: 1,
                engine: 'postgres',
                query_time_ms: 12,
              },
            },
            // `live=true` streams Server-Sent Events: `result` events carry this
            // same shape, alongside `error` and `lagged` events.
            'text/event-stream': { schema: OpenApi.resolver(schema.Response) },
          },
          description:
            'Structured result from a Tempo indexer SQL query. When `live=true` the response is a `text/event-stream` (SSE) of `result` events with this same shape, interleaved with `error` and `lagged` events, re-run on every new block.',
          headers: OpenApi.successHeaders,
        },
        // 400 and 422 are the upstream indexer's own error shape (`{ ok: false, error }`),
        // not the Tempo API JSON error envelope; the auth/rate-limit/proxy errors below are.
        400: {
          content: { 'application/json': { schema: OpenApi.resolver(schema.Error) } },
          description:
            'The query parameters are invalid; this uses the upstream indexer’s error shape.',
        },
        401: OpenApi.standardError(401, 'Your API key is missing or invalid.'),
        422: {
          content: { 'application/json': { schema: OpenApi.resolver(schema.Error) } },
          description:
            'The SQL failed validation or execution; this uses the upstream indexer’s error shape.',
        },
        429: OpenApi.standardError(429, 'You have exceeded the rate limit.'),
        500: OpenApi.standardError(500, 'The Tempo API encountered an internal server error.'),
        502: OpenApi.standardError(
          502,
          'The upstream Tempo indexer could not complete the request.',
        ),
      },
      summary: 'Query indexed chain data',
      tags: ['Indexer'],
    }),
    (c, next) => (isLive(c) ? next() : queryCache(c, next)),
    (c) => {
      if (Auth.narrowAccess) return Auth.paidAccessError(c)
      if (narrowPassthrough)
        return c.json(
          Response.validated(schema.Response, { columns: [], ok: true, row_count: 0, rows: [] }),
          200,
        )
      if (narrowPassthrough)
        return c.json(
          Response.validated(schema.Error, { error: 'Invalid query parameters', ok: false }),
          400,
        )
      if (narrowPassthrough)
        return c.json(
          Response.validated(schema.Error, {
            error: 'SQL validation or query execution error',
            ok: false,
          }),
          422,
        )
      return passthrough(c, 'v1')
    },
  )
}

async function passthrough(
  c: Context<App.Environment>,
  version: string,
): Promise<globalThis.Response & TypedResponse<never, never, never>> {
  const tidx = c.get('tidx')

  // Resolve the upstream base URL per chain so mainnet and testnet
  // requests reach the correct indexer.
  const chainIdParam = c.req.query('chainId') ?? c.req.query('chain_id')
  const parsed = chainIdParam ? Schema.ChainId.safeParse(chainIdParam) : undefined
  const chainId = parsed?.success ? parsed.data : c.get('chainId')
  // Resolve the upstream for this caller. A resolver `tidx` can route by
  // principal (e.g. anonymous `type: 'public'` callers to a public, no-auth
  // indexer); a static config ignores the principal.
  const { baseUrl, basicAuth, bearerAuth, zoneHeaders } = Tidx.resolve(tidx, {
    chainId,
    principal: Auth.getPrincipal(c),
    zone: c.get('zones').get(chainId),
  })
  // The chain id is syntactically valid but this deployment has no indexer
  // upstream configured for it (no `tidx` config and no built-in default).
  if (!baseUrl)
    return Response.error(c, {
      code: 'chain_id_unsupported',
      message: 'Unsupported chain id',
      status: 400,
    }) as never

  // Strip the `<basePath>/<version>/indexer` mount prefix to recover the
  // upstream path.
  const prefix = Path.join(c.get('basePath'), version, 'indexer')
  const requestUrl = new URL(c.req.url)
  const rest = requestUrl.pathname.slice(prefix.length) || '/'

  const upstream = new URL(baseUrl)
  upstream.pathname = Path.join(upstream.pathname, rest)
  upstream.search = requestUrl.search
  if (parsed?.success)
    for (const key of ['chainId', 'chain_id'])
      if (upstream.searchParams.has(key)) upstream.searchParams.set(key, String(chainId))

  // Enforce the documented execution bounds server-side: an arbitrary SELECT
  // from an anonymous caller is always bounded even if the upstream's own
  // defaulting changes. Caller-supplied values pass through (the upstream
  // clamps them to its documented ranges).
  if (!upstream.searchParams.has('timeout_ms')) upstream.searchParams.set('timeout_ms', '5000')
  if (!upstream.searchParams.has('limit')) upstream.searchParams.set('limit', '10000')

  const headers = new Headers()
  for (const [name, value] of c.req.raw.headers)
    if (!strippedHeaders.has(name.toLowerCase())) headers.set(name, value)
  if (zoneHeaders) for (const [name, value] of new Headers(zoneHeaders)) headers.set(name, value)
  if (basicAuth) headers.set('authorization', `Basic ${btoa(basicAuth)}`)
  else if (bearerAuth) headers.set('authorization', `Bearer ${bearerAuth}`)

  const method = c.req.method

  // `duplex` is required when streaming a request body, but is not yet in
  // the DOM `RequestInit` type.
  const init: RequestInit & { duplex?: 'half' } = { headers, method, redirect: 'manual' }
  const body = method === 'GET' || method === 'HEAD' ? null : c.req.raw.body
  if (body) {
    init.body = body
    init.duplex = 'half'
  }
  // Bound buffered queries so a hung upstream connection cannot pin the
  // request forever: the upstream's own per-query ceiling is 30 s, so 35 s
  // only fires when the connection itself stalls. The SSE `live` stream is
  // long-lived by design and gets no timeout.
  const live = isLive(c)
  if (!live) init.signal = AbortSignal.timeout(35_000)

  try {
    const { failure, response } = await Timing.time(c, 'tidx', async () => {
      const response = await fetch(new Request(upstream, init))
      // Syntax/validation failures belong to the caller, not the dependency's
      // operational error ratio.
      const failure = Tidx.isQueryRejection(response)
        ? undefined
        : live
          ? response.ok
            ? undefined
            : Tidx.providerFailure(response)
          : await Tidx.responseFailure(response)
      return { failure, response }
    })
    c.set('providerFailure', failure ? { ...failure, chainId } : undefined)

    // Stream the upstream response back verbatim so both JSON queries and
    // SSE `live` streams pass through untouched.
    return c.newResponse(response.body, response) as globalThis.Response &
      TypedResponse<never, never, never>
  } catch (cause) {
    // Timeouts and network failures become the standard 502 envelope instead
    // of an unhandled 500.
    c.set('providerFailure', { ...Tidx.providerFailure(cause), chainId })
    return Response.upstream(c, cause) as never
  }
}

/**
 * Whether the request asked for the SSE `live` stream. Anything other than an
 * explicit falsy value counts as live, so an ambiguous value is never buffered
 * into the response cache or cut off by the passthrough timeout.
 */
function isLive(c: Context<App.Environment>): boolean {
  const live = c.req.query('live')
  if (live === undefined) return false
  return !['', '0', 'false'].includes(live.toLowerCase())
}