import * as RequestEvents from './tables/requestEvents.js'

export { createQueueSink } from './tables/requestEvents.js'

/** One org's deduped billable request count for a UTC hour; the reporter's source row. */
export type BillableCount = RequestEvents.BillableCount

/** Options for {@link readBillableRequestCounts}. */
export type BillableCountsOptions = RequestEvents.BillableCountsOptions

/** One organization's request usage and error breakdown. */
export type ProjectUsage = RequestEvents.ProjectUsage

/** Options for reading one organization's request usage. */
export type ProjectUsageOptions = RequestEvents.ProjectUsageOptions

/** API errors grouped by their stable error code. */
export type ErrorBreakdown = RequestEvents.ErrorBreakdown

/** Reads stable API error-code counts for one organization attribution window. */
export function readErrorBreakdown(
  source: Source,
  options: ProjectUsageOptions,
): Promise<ErrorBreakdown[]> {
  return RequestEvents.readErrorBreakdown(get(source), options)
}

/** Reads request usage for an organization and optional attribution terms. */
export function readProjectUsage(
  source: Source,
  options: ProjectUsageOptions,
): Promise<ProjectUsage> {
  return RequestEvents.readProjectUsage(get(source), options)
}

/**
 * Reads deduped billable request counts grouped by `(org, UTC hour)` from
 * `request_events` — the request-metered billing source. Resolves the source
 * at the leaf so a Workers factory builds a fresh connection.
 *
 * @param source - The analytics store or factory.
 * @param options - The read options.
 * @returns One deduped count per `(org, hour)`.
 */
export function readBillableRequestCounts(
  source: Source,
  options: BillableCountsOptions,
): Promise<BillableCount[]> {
  return RequestEvents.readBillableCounts(get(source), options)
}

/**
 * ClickHouse schema; table repository modules (`tables/*`) type their rows
 * through it.
 */
export type Database = {
  request_events: RequestEvents.Table
}

/**
 * A ClickHouse-backed analytics store: row inserts, JSON reads, and schema
 * migrations. Table repository modules (`tables/*`) query through it.
 */
export type Analytics = {
  /** Inserts rows into a table. */
  insert<name extends keyof Database>(table: name, rows: readonly Database[name][]): Promise<void>
  /** Applies any pending schema migrations; safe to re-run. */
  migrate(): Promise<void>
  /** Runs a read query, returning JSON rows. */
  query<row>(sql: string): Promise<row[]>
}

/**
 * An {@link Analytics}, or a factory for per-request construction. Resolve at
 * the leaf with {@link get}.
 */
export type Source = Analytics | (() => Analytics)

/** One stored request row (`request_events`); hosts use it in their queue-body union. */
export type Event = RequestEvents.Table

/** Queue name {@link handleQueue} consumes; the host's queue producer/consumer bindings must reference it. */
export const queueName = 'api-request-analytics'

/**
 * Resolves a {@link Source} to an {@link Analytics}. Call at the request leaf
 * so factories create per-request instances; singletons resolve to themselves.
 *
 * @param source - The analytics store or factory.
 * @returns The resolved {@link Analytics}.
 */
export function get(source: Source): Analytics {
  return typeof source === 'function' ? source() : source
}

/** One schema migration: a single idempotent DDL statement, safe to re-apply. */
type Migration = {
  /** Unique, ordered migration name (e.g. `0001_request_events`). */
  name: string
  /** Idempotent DDL applied by {@link Analytics.migrate}. */
  sql: string
  /**
   * Optional `SELECT` run before {@link Migration.sql}; a nonzero first value
   * skips the migration. Lets a scoped user without an `ALTER` grant (preview)
   * skip an `ALTER` whose columns already exist from the baseline `CREATE`.
   */
  guard?: string
}

/**
 * Ordered migrations for {@link Analytics.migrate}. ClickHouse lacks
 * transactional DDL and advisory locks, so each statement is idempotent
 * (`IF NOT EXISTS`) and re-applied on every run; no applied-migrations ledger.
 */
export const migrations: readonly Migration[] = [
  {
    name: '0001_request_events',
    sql: `
      CREATE TABLE IF NOT EXISTS request_events (
        timestamp DateTime64(3, 'UTC'),
        service LowCardinality(String),
        environment LowCardinality(String),
        chain_id Nullable(UInt32),
        method LowCardinality(String),
        route LowCardinality(String),
        status UInt16,
        duration_ms Float64,
        principal_type LowCardinality(String),
        key_id Nullable(String),
        key_environment Nullable(String),
        org_id Nullable(String),
        project_id Nullable(String),
        query String,
        rate_limit_scope Nullable(String),
        error_code Nullable(String),
        rpc_error_code Nullable(Int32),
        rpc_error_data_code Nullable(String),
        rpc_error_count UInt16 DEFAULT 0,
        request_id Nullable(String),
        billing_active Nullable(UInt8),
        timings Map(String, Float64)
      )
      ENGINE = MergeTree
      ORDER BY (timestamp)
    `,
  },
  {
    name: '0002_request_events_billing',
    // Patches tables created before the billing columns were folded into 0001
    // (production). Fresh databases already carry them from the CREATE, so the
    // guard skips this ALTER there, where the scoped migrate user (preview)
    // lacks the ALTER grant. ADD COLUMN IF NOT EXISTS keeps it idempotent.
    guard: `
      SELECT toUInt8(count() = 2) FROM system.columns
      WHERE database = currentDatabase()
        AND table = 'request_events'
        AND name IN ('request_id', 'billing_active')
    `,
    sql: `
      ALTER TABLE request_events
        ADD COLUMN IF NOT EXISTS request_id Nullable(String),
        ADD COLUMN IF NOT EXISTS billing_active Nullable(UInt8)
    `,
  },
  {
    name: '0003_request_events_rpc_errors',
    // Patches tables created before RPC columns entered the baseline. Fresh
    // tables include them, allowing scoped preview users without ALTER to skip.
    guard: `
      SELECT toUInt8(count() = 3) FROM system.columns
      WHERE database = currentDatabase()
        AND table = 'request_events'
        AND name IN ('rpc_error_code', 'rpc_error_data_code', 'rpc_error_count')
    `,
    sql: `
      ALTER TABLE request_events
        ADD COLUMN IF NOT EXISTS rpc_error_code Nullable(Int32),
        ADD COLUMN IF NOT EXISTS rpc_error_data_code Nullable(String),
        ADD COLUMN IF NOT EXISTS rpc_error_count UInt16 DEFAULT 0
    `,
  },
  {
    name: '0004_request_events_timings',
    // Patches tables created before request timings entered the baseline. Fresh
    // tables skip this ALTER, allowing scoped preview users to migrate.
    guard: `
      SELECT toUInt8(count() = 1) FROM system.columns
      WHERE database = currentDatabase()
        AND table = 'request_events'
        AND name = 'timings'
    `,
    sql: `
      ALTER TABLE request_events
        ADD COLUMN IF NOT EXISTS timings Map(String, Float64)
    `,
  },
  {
    name: '0005_query_benchmark_results',
    sql: `
      CREATE TABLE IF NOT EXISTS query_benchmark_results (
        timestamp DateTime64(3, 'UTC'),
        run_id String,
        repository LowCardinality(String),
        git_sha String,
        git_ref LowCardinality(String),
        chain_id UInt32,
        chain LowCardinality(String),
        method LowCardinality(String),
        route LowCardinality(String),
        query String,
        metric LowCardinality(String),
        mean_ms Float64,
        min_ms Float64,
        max_ms Float64,
        samples UInt16,
        tidx_queries Array(String),
        rpc_url String,
        tidx_url String
      )
      ENGINE = ReplacingMergeTree
      PARTITION BY toYYYYMM(timestamp)
      ORDER BY (chain_id, query, metric, timestamp, run_id)
    `,
  },
  {
    name: '0006_endpoint_latency_dimensions',
    // Patches production while fresh preview tables skip the unavailable ALTER.
    guard: `
      SELECT toUInt8(count() = 1) FROM system.columns
      WHERE database = currentDatabase()
        AND table = 'request_events'
        AND name = 'chain_id'
    `,
    sql: `
      ALTER TABLE request_events
        ADD COLUMN IF NOT EXISTS chain_id Nullable(UInt32) AFTER environment
    `,
  },
  {
    name: '0007_query_benchmark_dimensions',
    // Patches production while fresh preview tables skip the unavailable ALTER.
    guard: `
      SELECT toUInt8(count() = 2) FROM system.columns
      WHERE database = currentDatabase()
        AND table = 'query_benchmark_results'
        AND name IN ('method', 'route')
    `,
    sql: `
      ALTER TABLE query_benchmark_results
        ADD COLUMN IF NOT EXISTS method LowCardinality(String) AFTER chain,
        ADD COLUMN IF NOT EXISTS route LowCardinality(String) AFTER method
    `,
  },
  {
    name: '0008_endpoint_latency_results',
    sql: `
      CREATE OR REPLACE VIEW endpoint_latency_results AS
      SELECT
        timestamp,
        'production' AS source,
        chain_id,
        method,
        route AS endpoint,
        '' AS variant,
        toNullable(status) AS status,
        if(status >= 200 AND status < 300, 'success', 'error') AS outcome,
        measurement.1 AS metric,
        measurement.2 AS mean_ms,
        measurement.2 AS min_ms,
        measurement.2 AS max_ms,
        toUInt64(1) AS samples,
        '' AS run_id,
        '' AS git_sha
      FROM request_events
      ARRAY JOIN arrayConcat(
        [('request', duration_ms)],
        arrayFilter(
          measurement -> measurement.1 != 'request',
          arrayZip(mapKeys(timings), mapValues(timings))
        )
      ) AS measurement
      WHERE environment = 'production' AND service = 'api'

      UNION ALL

      SELECT
        timestamp,
        'benchmark' AS source,
        toNullable(chain_id) AS chain_id,
        method,
        if(route = '', query, route) AS endpoint,
        query AS variant,
        CAST(NULL, 'Nullable(UInt16)') AS status,
        'success' AS outcome,
        if(result.metric = 'benchmark', 'request', result.metric) AS metric,
        mean_ms,
        min_ms,
        max_ms,
        toUInt64(samples) AS samples,
        run_id,
        git_sha
      FROM query_benchmark_results AS result
      WHERE result.metric != 'request'
    `,
  },
  {
    name: '0009_endpoint_latency_production_service',
    sql: `
      CREATE OR REPLACE VIEW endpoint_latency_results AS
      SELECT
        timestamp,
        'production' AS source,
        chain_id,
        method,
        route AS endpoint,
        '' AS variant,
        toNullable(status) AS status,
        if(status >= 200 AND status < 300, 'success', 'error') AS outcome,
        measurement.1 AS metric,
        measurement.2 AS mean_ms,
        measurement.2 AS min_ms,
        measurement.2 AS max_ms,
        toUInt64(1) AS samples,
        '' AS run_id,
        '' AS git_sha
      FROM request_events
      ARRAY JOIN arrayConcat(
        [('request', duration_ms)],
        arrayFilter(
          measurement -> measurement.1 != 'request',
          arrayZip(mapKeys(timings), mapValues(timings))
        )
      ) AS measurement
      WHERE environment = 'production' AND service = 'cadent-api'

      UNION ALL

      SELECT
        timestamp,
        'benchmark' AS source,
        toNullable(chain_id) AS chain_id,
        method,
        if(route = '', query, route) AS endpoint,
        query AS variant,
        CAST(NULL, 'Nullable(UInt16)') AS status,
        'success' AS outcome,
        if(result.metric = 'benchmark', 'request', result.metric) AS metric,
        mean_ms,
        min_ms,
        max_ms,
        toUInt64(samples) AS samples,
        run_id,
        git_sha
      FROM query_benchmark_results AS result
      WHERE result.metric != 'request'
    `,
  },
  {
    name: '0010_request_events_query',
    // Patches tables created before redacted request queries entered the baseline.
    // Fresh tables skip this ALTER, allowing scoped preview users to migrate.
    guard: `
      SELECT toUInt8(count() = 1) FROM system.columns
      WHERE database = currentDatabase()
        AND table = 'request_events'
        AND name = 'query'
    `,
    sql: `
      ALTER TABLE request_events
        ADD COLUMN IF NOT EXISTS query String AFTER project_id
    `,
  },
  {
    name: '0011_endpoint_latency_query',
    sql: `
      CREATE OR REPLACE VIEW endpoint_latency_results AS
      SELECT
        timestamp,
        'production' AS source,
        chain_id,
        method,
        route AS endpoint,
        query AS variant,
        toNullable(status) AS status,
        if(status >= 200 AND status < 300, 'success', 'error') AS outcome,
        measurement.1 AS metric,
        measurement.2 AS mean_ms,
        measurement.2 AS min_ms,
        measurement.2 AS max_ms,
        toUInt64(1) AS samples,
        coalesce(request_id, '') AS request_id,
        '' AS run_id,
        '' AS git_sha
      FROM request_events
      ARRAY JOIN arrayConcat(
        [('request', duration_ms)],
        arrayFilter(
          measurement -> measurement.1 != 'request',
          arrayZip(mapKeys(timings), mapValues(timings))
        )
      ) AS measurement
      WHERE environment = 'production' AND service = 'cadent-api'

      UNION ALL

      SELECT
        timestamp,
        'benchmark' AS source,
        toNullable(chain_id) AS chain_id,
        method,
        if(route = '', query, route) AS endpoint,
        query AS variant,
        CAST(NULL, 'Nullable(UInt16)') AS status,
        'success' AS outcome,
        if(result.metric = 'benchmark', 'request', result.metric) AS metric,
        mean_ms,
        min_ms,
        max_ms,
        toUInt64(samples) AS samples,
        '' AS request_id,
        run_id,
        git_sha
      FROM query_benchmark_results AS result
      WHERE result.metric != 'request'
    `,
  },
]

/**
 * Creates a ClickHouse-backed {@link Analytics} over the HTTP interface.
 * Requests scope to `database` (the `?database=` parameter); table names are
 * code-owned by the `tables/*` modules, not configuration.
 *
 * @param options - ClickHouse connection options.
 * @returns The analytics store.
 */
export function clickhouse(options: clickhouse.Options): Analytics {
  const authorization = `Basic ${btoa(`${options.user}:${options.password}`)}`

  function endpoint() {
    const url = new URL(options.url)
    url.searchParams.set('database', options.database)
    url.searchParams.set('date_time_input_format', 'best_effort')
    return url
  }

  // Runs a migration guard SELECT; true (skip) when its first value is nonzero.
  // A failed guard returns false so the migration still runs and reports.
  async function guarded(sql: string): Promise<boolean> {
    const response = await fetch(endpoint(), {
      body: `${sql} FORMAT JSON`,
      headers: { Authorization: authorization, 'Content-Type': 'text/plain' },
      method: 'POST',
    })
    if (!response.ok) return false
    try {
      const body = (await response.json()) as { data?: Array<Record<string, unknown>> }
      const first = body.data?.[0]
      return first ? Number(Object.values(first)[0]) > 0 : false
    } catch {
      return false
    }
  }

  return {
    async insert(table, rows) {
      if (rows.length === 0) return
      const url = endpoint()
      url.searchParams.set('query', `INSERT INTO ${quote(table)} FORMAT JSONEachRow`)
      const response = await fetch(url, {
        body: `${rows.map((row) => JSON.stringify(row)).join('\n')}\n`,
        headers: { Authorization: authorization, 'Content-Type': 'application/x-ndjson' },
        method: 'POST',
      })
      if (response.ok) return
      throw new InsertError(response.status, await response.text())
    },
    async migrate() {
      for (const migration of migrations) {
        if (migration.guard && (await guarded(migration.guard))) continue
        const response = await fetch(endpoint(), {
          body: migration.sql,
          headers: { Authorization: authorization, 'Content-Type': 'text/plain' },
          method: 'POST',
        })
        if (response.ok) continue
        throw new MigrateError(migration.name, response.status, await response.text())
      }
    },
    async query<row>(sql: string): Promise<row[]> {
      const response = await fetch(endpoint(), {
        body: `${sql} FORMAT JSON`,
        headers: { Authorization: authorization, 'Content-Type': 'text/plain' },
        method: 'POST',
      })
      if (!response.ok) throw new QueryError(response.status, await response.text())
      const body = (await response.json()) as { data?: row[] | undefined }
      return body.data ?? []
    },
  }
}

export declare namespace clickhouse {
  /** ClickHouse connection options. */
  type Options = {
    /** Database the connection scopes to (e.g. `tempo_api`). */
    database: string
    /** Password for `user`. */
    password: string
    /** ClickHouse HTTPS endpoint. */
    url: string
    /** ClickHouse user; ingest and read deployments differ only here. */
    user: string
  }
}

/**
 * Consumes one queue batch when it is the analytics queue ({@link queueName}):
 * batch-inserts the queued rows, acking on success and retrying on failure.
 * Returns whether the batch was handled, so a multi-queue worker can chain:
 *
 * ```ts
 * async queue(batch) {
 *   if (await Analytics.handleQueue(analytics, batch)) return
 *   // … other queues …
 * }
 * ```
 *
 * @param source - The analytics store or factory.
 * @param batch - The queue batch.
 * @param options - Options.
 * @returns Whether the batch belonged to the analytics queue.
 */
export async function handleQueue(
  source: Source,
  batch: handleQueue.Batch,
  options: handleQueue.Options = {},
): Promise<boolean> {
  if (batch.queue !== (options.queue ?? queueName)) return false
  await RequestEvents.insertMessages(
    get(source),
    batch.messages as readonly RequestEvents.insertMessages.Message[],
    { onResult: options.onResult },
  )
  return true
}

export declare namespace handleQueue {
  /** Options for {@link handleQueue}. */
  type Options = {
    /** Receives the final insert disposition for metrics and error reporting. */
    onResult?: RequestEvents.insertMessages.Options['onResult']
    /** Queue name to match instead of {@link queueName} (e.g. a per-preview queue). */
    queue?: string | undefined
  }

  /** Minimal queue batch shape; Cloudflare's `MessageBatch` satisfies it. */
  type Batch = {
    /** Queued messages; bodies are {@link Event} rows when `queue` matches. */
    messages: readonly Message[]
    /** Originating queue name. */
    queue: string
  }

  /** One queued message. */
  type Message = {
    /** Acknowledges the message. */
    ack(): void
    /** Queued body; an {@link Event} when the batch is the analytics queue. */
    body: unknown
    /** Marks the message for redelivery. */
    retry(): void
  }
}

/** Valid unquoted ClickHouse identifier. */
const identifier = /^[A-Za-z_][A-Za-z0-9_]*$/

function quote(value: string) {
  if (identifier.test(value)) return `\`${value}\``
  throw new ConfigError(value)
}

/** Rejected ClickHouse identifier. */
export class ConfigError extends Error {
  constructor(value: string) {
    super(`Invalid ClickHouse identifier: ${value}`)
    this.name = 'Analytics.ConfigError'
  }
}

/** Failed ClickHouse insert (non-2xx HTTP response). */
export class InsertError extends Error {
  /** HTTP response status returned by ClickHouse. */
  status: number

  constructor(status: number, body: string) {
    super(`ClickHouse insert failed with status ${status}: ${body.slice(0, 500)}`)
    this.name = 'Analytics.InsertError'
    this.status = status
  }
}

/** Failed ClickHouse schema migration (non-2xx HTTP response). */
export class MigrateError extends Error {
  constructor(name: string, status: number, body: string) {
    super(`ClickHouse migration "${name}" failed with status ${status}: ${body.slice(0, 500)}`)
    this.name = 'Analytics.MigrateError'
  }
}

/** Failed ClickHouse read query (non-2xx HTTP response). */
export class QueryError extends Error {
  constructor(status: number, body: string) {
    super(`ClickHouse query failed with status ${status}: ${body.slice(0, 500)}`)
    this.name = 'Analytics.QueryError'
  }
}