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 = OpenApi.component(
    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.'),
            z.meta({ examples: [['num', 'hash', 'timestamp']] }),
          ),
        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`.',
            ),
            z.meta({ examples: [[[1_234_567, `0x${'aa'.repeat(32)}`, 1_718_668_800]]] }),
          ),
        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.')),
    'IndexerQueryResponse',
  )

  /** Error envelope returned by the upstream indexer. */
  export const Error = OpenApi.component(
    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.'),
            z.meta({ examples: ['Only SELECT queries are supported'] }),
          ),
      })
      .check(z.describe('Error response returned by the upstream Tempo indexer.')),
    'IndexerQueryError',
  )

  /** Tempo error envelope returned before a query reaches the indexer. */
  export const RequestError = OpenApi.component(
    Schema.errorResponse(['api_key_malformed', 'chain_id_invalid', 'chain_id_unsupported']),
    'IndexerQueryRequestError',
  )

  /** Either error envelope that a rejected indexer request can return. */
  export const BadRequest = OpenApi.component(
    z
      .union([Error, RequestError])
      .check(
        z.describe('An upstream query error or a Tempo authentication or chain-selection error.'),
      ),
    'IndexerQueryBadRequest',
  )

  /** Tempo error envelope returned when the upstream indexer cannot be reached. */
  export const TempoUpstreamError = OpenApi.component(
    Schema.errorResponse(['upstream_error']),
    'IndexerQueryTempoUpstreamError',
  )

  /** Either error envelope returned when the indexer cannot complete a request. */
  export const UpstreamFailure = OpenApi.component(
    z
      .union([Error, TempoUpstreamError])
      .check(z.describe('An upstream indexer error or a Tempo upstream error.')),
    'IndexerQueryUpstreamFailure',
  )

  /** Server-Sent Event stream returned for live indexer queries. */
  export const EventStream = OpenApi.component(
    z.string().check(
      z.describe('A stream of indexer result, error, and lagged events.'),
      z.meta({
        examples: [
          'event: result\ndata: {"ok":true,"columns":["num"],"rows":[[1234567]],"row_count":1}\n\n',
        ],
      }),
    ),
    'IndexerQueryEventStream',
  )
}

/** 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: false,
          schema: {
            anyOf: [
              { enum: ['mainnet', 'testnet'], type: 'string' },
              { 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.EventStream) },
          },
          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 covers both the upstream indexer shape and Tempo's error envelope.
        // 422 remains the upstream indexer shape; the other errors are Tempo envelopes.
        400: {
          content: { 'application/json': { schema: OpenApi.resolver(schema.BadRequest) } },
          description:
            'The query is invalid, the API key is malformed, or the selected chain is invalid or unsupported. Query errors use the indexer envelope; Tempo errors use the standard API envelope.',
        },
        401: OpenApi.standardError(401, 'The API key is missing or invalid.'),
        403: OpenApi.standardError(403, 'The API key cannot query the selected indexer.'),
        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, 'The request exceeded its rate limit.'),
        500: OpenApi.standardError(500, 'The Tempo API encountered an internal server error.'),
        502: {
          content: { 'application/json': { schema: OpenApi.resolver(schema.UpstreamFailure) } },
          description:
            'The indexer could not complete the request. Indexer HTTP errors retain the indexer envelope; network failures use the Tempo envelope.',
        },
        504: OpenApi.standardError(504, 'The request timed out before it completed.'),
      },
      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.text(
          'event: result\ndata: {"ok":true,"columns":[],"rows":[],"row_count":0}\n\n',
          200,
        )
      if (narrowPassthrough)
        return Response.error(c, {
          code: 'chain_id_invalid',
          message: 'Invalid chain id',
          status: 400,
        })
      if (narrowPassthrough)
        return Response.error(c, {
          code: 'chain_id_unsupported',
          message: 'Unsupported chain id',
          status: 400,
        })
      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,
        )
      if (narrowPassthrough)
        return c.json(
          Response.validated(schema.Error, { error: 'Indexer unavailable', ok: false }),
          502,
        )
      if (narrowPassthrough) return Response.upstream(c, new Error('Indexer passthrough failed'))
      if (narrowPassthrough)
        return Response.error(c, {
          code: 'request_timeout',
          message: 'Request timed out',
          status: 504,
        })
      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 { body, 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.
      if (Tidx.isQueryRejection(response))
        return { body: response.body, failure: undefined, response }
      if (live)
        return {
          body: response.body,
          failure: response.ok ? undefined : Tidx.providerFailure(response),
          response,
        }
      const inspection = await Tidx.inspectResponse(response)
      return { ...inspection, 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(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())
}