import type * as Log from '../../internal/Log.js'
import type * as Metrics from '../../Metrics.js'
import type * as Analytics from '../Analytics.js'

/** Columns of the `request_events` table: one compact row per API request. */
export type Table = {
  /**
   * Whether the owning org's billing was active at request time (`1`/`0`), for
   * authenticated API-key callers; `null` otherwise. The request-metered
   * billing gate for sandbox: only `1` rows bill.
   */
  billing_active: number | null
  /** Tempo chain selected for the request. */
  chain_id: number | null
  /** Total request duration in milliseconds. */
  duration_ms: number
  /** Deployment environment that served the request. */
  environment: string
  /** Stable API error code, when the response was an error envelope. */
  error_code: string | null
  /** API-key environment for authenticated API-key callers. */
  key_environment: 'production' | 'sandbox' | null
  /** API key id for authenticated API-key callers. */
  key_id: string | null
  /** HTTP method. */
  method: string
  /** Organization id for authenticated API-key callers. */
  org_id: string | null
  /** Project id for authenticated API-key callers. */
  project_id: string | null
  /** Caller principal kind. */
  principal_type: 'api_key' | 'public' | 'session' | 'super_admin' | 'unknown'
  /** Redacted request query string, without a leading `?`. */
  query: string
  /** Rate-limit bucket scope, when a rate limit was applied. */
  rate_limit_scope: string | null
  /** Numeric JSON-RPC error code when every failed response shares one code. */
  rpc_error_code: number | null
  /** Stable application error code from JSON-RPC `error.data.code`. */
  rpc_error_data_code: string | null
  /** Number of failed JSON-RPC responses in the request. */
  rpc_error_count: number
  /** Stable per-request id (`entry.requestId`); dedups the billing count via `uniqExact`. */
  request_id: string | null
  /** Bounded matched route pattern. */
  route: string
  /** Service that emitted the event. */
  service: string
  /** HTTP response status code. */
  status: number
  /** UTC request completion timestamp formatted for ClickHouse DateTime64. */
  timestamp: string
  /** Longest Server-Timing operation duration per metric name. */
  timings: Record<string, number>
}

/** One request attribution term for usage reads. */
export type Attribution = {
  /** API key ids matched by this term; absent means all keys. */
  apiKeyIds?: readonly string[] | undefined
  /** API-key environment matched by this term; absent means all environments. */
  environment?: Environment | undefined
  /**
   * Project scope for this term: a project id matches that project, `null`
   * matches rows without project attribution, and `undefined` matches the
   * whole organization (any project, including unattributed rows).
   */
  projectId?: string | null | undefined
}

/** API-key environment for request usage. */
export type Environment = 'production' | 'sandbox'

/** API errors grouped by their stable error code. */
export type ErrorBreakdown = {
  /** Stable API error code, or `unknown` when none was recorded. */
  code: string
  /** Requests returning this error. */
  requests: number
}

/** API key usage grouped for one project. */
export type KeyBreakdown = {
  /** API key id (`key_…`). */
  apiKeyId: string
  /** Average request duration in milliseconds. */
  averageDurationMs: number
  /** Requests whose status was at least 400. */
  errors: number
  /** Total requests. */
  requests: number
}

/** Latest authenticated request timestamp for one API key. */
export type LastUsedAt = {
  /** API key id (`key_…`). */
  apiKeyId: string
  /** Latest request completion timestamp (ISO 8601). */
  lastUsedAt: string
}

/** Time bucket size for the usage series. */
export type Interval = 'day' | 'hour'

/** Route usage grouped for one project. */
export type RouteBreakdown = {
  /** Average request duration in milliseconds. */
  averageDurationMs: number
  /** Requests whose status was at least 400. */
  errors: number
  /** Total requests. */
  requests: number
  /** Matched route pattern. */
  route: string
}

/** One usage series point. */
export type SeriesPoint = {
  /** Requests whose status was at least 400. */
  errors: number
  /** Total requests. */
  requests: number
  /** Bucket start timestamp (ISO 8601). */
  time: string
}

/** Status-code usage grouped for one project. */
export type StatusBreakdown = {
  /** Total requests. */
  requests: number
  /** HTTP status code. */
  status: number
}

/** Overall usage totals for one project. */
export type Totals = {
  /** Average request duration in milliseconds. */
  averageDurationMs: number
  /** Requests whose status was at least 400. */
  errors: number
  /** Total requests. */
  requests: number
}

/** Usage read options for one project. */
export type ProjectUsageOptions = {
  /** Attribution terms that define which request rows count. */
  attributions: readonly Attribution[]
  /** Inclusive lower timestamp bound (ISO 8601). */
  from: string
  /** Bucket size for the time series. */
  interval: Interval
  /** Organization id (`org_…`). */
  orgId: string
  /** Inclusive upper timestamp bound (ISO 8601). */
  to: string
}

/** Project usage response body. */
export type ProjectUsage = {
  /** Usage grouped by API key. */
  byKey: readonly KeyBreakdown[]
  /** Usage grouped by route. */
  byRoute: readonly RouteBreakdown[]
  /** Usage grouped by status code. */
  byStatus: readonly StatusBreakdown[]
  /** Inclusive lower timestamp bound used for the read (ISO 8601). */
  from: string
  /** Bucket size used for the time series. */
  interval: Interval
  /** Usage grouped into small time buckets. */
  series: readonly SeriesPoint[]
  /** Inclusive upper timestamp bound used for the read (ISO 8601). */
  to: string
  /** Overall usage totals. */
  totals: Totals
}

/**
 * Builds a `Metrics.cloudflare` analytics factory that enqueues one compact
 * request row per entry onto `queue`. Pass the result directly as the
 * backend's `analytics` option; it reuses the backend's
 * `enabled`/`environment`/`service` context and self-disables when `enabled`
 * is false.
 *
 * @param queue - The analytics queue producer.
 * @returns The analytics factory.
 */
export function createQueueSink(queue: createQueueSink.Queue): createQueueSink.Factory {
  return (context) => async (entry) => {
    if (!context.enabled) return
    const row = event(entry, context)
    try {
      await queue.send(row)
    } catch {
      // Transient Queues internal errors (e.g. code 15000) usually clear on retry.
      await queue.send(row)
    }
  }
}

export declare namespace createQueueSink {
  /** Analytics factory accepted by `Metrics.cloudflare`'s `analytics` option. */
  type Factory = (context: Metrics.cloudflare.Context) => Log.Emit

  /** Minimal Cloudflare Queue producer shape the sink sends rows to. */
  type Queue = {
    /** Enqueues one row; the result is awaited and discarded. */
    send(message: Table): Promise<unknown>
  }
}

/**
 * Inserts a batch of queued request rows, acking each message on success and
 * retrying the batch on failure.
 *
 * @param analytics - The analytics store.
 * @param messages - The queued messages.
 */
export async function insertMessages(
  analytics: Analytics.Analytics,
  messages: readonly insertMessages.Message[],
  options: insertMessages.Options = {},
): Promise<void> {
  if (messages.length === 0) return

  try {
    await analytics.insert(
      'request_events',
      messages.map((message) => message.body),
    )
    for (const message of messages) message.ack()
    reportResult({ messages: messages.length, outcome: 'acked' })
  } catch (error) {
    console.error('ClickHouse analytics insert failed', error)
    for (const message of messages) message.retry()
    reportResult({ cause: error, messages: messages.length, outcome: 'retried' })
  }

  function reportResult(result: insertMessages.Result) {
    try {
      options.onResult?.(result)
    } catch (error) {
      console.error('Analytics queue result callback failed', error)
    }
  }
}

export declare namespace insertMessages {
  /** Minimal Cloudflare Queue message shape carrying one row. */
  type Message = {
    /** Acknowledges the message. */
    ack(): void
    /** Queued row. */
    body: Table
    /** Marks the message for redelivery. */
    retry(): void
  }

  /** Queue insert observability options. */
  type Options = {
    /** Receives one bounded batch result after messages are acknowledged or retried. */
    onResult?: ((result: Result) => void) | undefined
  }

  /** Outcome of one analytics insert batch. */
  type Result =
    | {
        /** Number of messages acknowledged. */
        messages: number
        /** Successful queue disposition. */
        outcome: 'acked'
      }
    | {
        /** Original insert failure. */
        cause: unknown
        /** Number of messages marked for retry. */
        messages: number
        /** Failed queue disposition. */
        outcome: 'retried'
      }
}

/**
 * Reads the latest authenticated request timestamp for each requested API key.
 * Keys with no request events are omitted.
 */
export async function readLastUsedAt(
  analytics: Analytics.Analytics,
  apiKeyIds: readonly string[],
): Promise<LastUsedAt[]> {
  if (apiKeyIds.length === 0) return []
  const rows = await analytics.query<LastUsedAtRow>(
    `
      SELECT
        key_id,
        formatDateTime(max(timestamp), '%Y-%m-%dT%H:%i:%S.000Z', 'UTC') AS lastUsedAt
      FROM ${table}
      WHERE principal_type = 'api_key'
        AND key_id IN (${apiKeyIds.map(ClickHouseValue.literal).join(', ')})
      GROUP BY key_id
    `,
  )
  return rows.map((row) => ({
    apiKeyId: String(row.key_id),
    lastUsedAt: String(row.lastUsedAt),
  }))
}

/**
 * Reads request usage for one project from the `request_events` table.
 *
 * @param analytics - The analytics store.
 * @param options - The usage read options.
 * @returns The project usage.
 */
export async function readProjectUsage(
  analytics: Analytics.Analytics,
  options: ProjectUsageOptions,
): Promise<ProjectUsage> {
  const where = usageWhere(options)
  const average = `toFloat64(coalesce(round(avgOrNull(duration_ms), 1), 0))`
  const bucket = options.interval === 'day' ? 'toStartOfDay(timestamp)' : 'toStartOfHour(timestamp)'
  const [byKey, byRoute, byStatus, series, totals] = await Promise.all([
    analytics.query<KeyRow>(
      `
        SELECT
          key_id,
          count() AS requests,
          countIf(status >= 400 OR rpc_error_count > 0) AS errors,
          ${average} AS averageDurationMs
        FROM ${table}
        WHERE ${where} AND key_id IS NOT NULL
        GROUP BY key_id
        ORDER BY requests DESC, key_id ASC
        LIMIT 100
      `,
    ),
    analytics.query<RouteRow>(
      `
        SELECT
          route,
          count() AS requests,
          countIf(status >= 400 OR rpc_error_count > 0) AS errors,
          ${average} AS averageDurationMs
        FROM ${table}
        WHERE ${where}
        GROUP BY route
        ORDER BY requests DESC, route ASC
        LIMIT 100
      `,
    ),
    analytics.query<StatusRow>(
      `
        SELECT
          status,
          count() AS requests
        FROM ${table}
        WHERE ${where}
        GROUP BY status
        ORDER BY status ASC
      `,
    ),
    analytics.query<SeriesRow>(
      `
        SELECT
          formatDateTime(${bucket}, '%Y-%m-%dT%H:%i:%S.000Z', 'UTC') AS time,
          count() AS requests,
          countIf(status >= 400 OR rpc_error_count > 0) AS errors
        FROM ${table}
        WHERE ${where}
        GROUP BY ${bucket}
        ORDER BY ${bucket} ASC
      `,
    ),
    analytics.query<TotalsRow>(
      `
        SELECT
          count() AS requests,
          countIf(status >= 400 OR rpc_error_count > 0) AS errors,
          ${average} AS averageDurationMs
        FROM ${table}
        WHERE ${where}
      `,
    ),
  ])
  return {
    byKey: byKey.map((row) => ({
      apiKeyId: String(row.key_id),
      averageDurationMs: ClickHouseValue.number(row.averageDurationMs),
      errors: ClickHouseValue.number(row.errors),
      requests: ClickHouseValue.number(row.requests),
    })),
    byRoute: byRoute.map((row) => ({
      averageDurationMs: ClickHouseValue.number(row.averageDurationMs),
      errors: ClickHouseValue.number(row.errors),
      requests: ClickHouseValue.number(row.requests),
      route: String(row.route),
    })),
    byStatus: byStatus.map((row) => ({
      requests: ClickHouseValue.number(row.requests),
      status: ClickHouseValue.number(row.status),
    })),
    from: options.from,
    interval: options.interval,
    series: series.map((row) => ({
      errors: ClickHouseValue.number(row.errors),
      requests: ClickHouseValue.number(row.requests),
      time: String(row.time),
    })),
    to: options.to,
    totals: totals[0]
      ? {
          averageDurationMs: ClickHouseValue.number(totals[0].averageDurationMs),
          errors: ClickHouseValue.number(totals[0].errors),
          requests: ClickHouseValue.number(totals[0].requests),
        }
      : { averageDurationMs: 0, errors: 0, requests: 0 },
  }
}

/** Reads stable API error-code counts for one organization attribution window. */
export async function readErrorBreakdown(
  analytics: Analytics.Analytics,
  options: ProjectUsageOptions,
): Promise<ErrorBreakdown[]> {
  const rows = await analytics.query<ErrorRow>(
    `
      SELECT
        coalesce(
          error_code,
          rpc_error_data_code,
          if(rpc_error_count > 0 AND rpc_error_code IS NOT NULL, concat('rpc_', toString(rpc_error_code)), NULL),
          if(rpc_error_count > 0, 'rpc_multiple', 'unknown')
        ) AS code,
        count() AS requests
      FROM ${table}
      WHERE ${usageWhere(options)} AND (status >= 400 OR rpc_error_count > 0)
      GROUP BY code
      ORDER BY requests DESC, code ASC
      LIMIT 25
    `,
  )
  return rows.map((row) => ({
    code: String(row.code),
    requests: ClickHouseValue.number(row.requests),
  }))
}

/** One org's deduped billable request count for a single UTC hour. */
export type BillableCount = {
  /** UTC hour start (ISO 8601). */
  bucketStart: string
  /** Deduped billable request count (`uniqExact(request_id)`). */
  count: number
  /** Organization id (`org_…`). */
  orgId: string
}

/** Options for {@link readBillableCounts}. */
export type BillableCountsOptions = {
  /** API-key environment to bill; `sandbox` rows require request-time billing. */
  environment: Environment
  /** Inclusive lower timestamp bound (ISO 8601). */
  from: string
  /** Exclusive upper timestamp bound (ISO 8601); the seal cutoff. */
  to: string
}

/**
 * Reads deduped billable request counts grouped by `(org, UTC hour)` for the
 * request-metered billing reporter. Counts dedupe on `request_id` via
 * `uniqExact` (the queue → ClickHouse path can duplicate rows); production rows
 * always bill, sandbox rows only when `billing_active = 1` at request time.
 *
 * @param analytics - The analytics store.
 * @param options - The read options.
 * @returns One deduped count per `(org, hour)`.
 */
export async function readBillableCounts(
  analytics: Analytics.Analytics,
  options: BillableCountsOptions,
): Promise<BillableCount[]> {
  const sandboxGate = options.environment === 'production' ? '' : `AND billing_active = 1`
  const rows = await analytics.query<BillableCountRow>(
    `
      SELECT
        org_id AS orgId,
        formatDateTime(toStartOfHour(timestamp), '%Y-%m-%dT%H:%i:%S.000Z', 'UTC') AS bucketStart,
        uniqExact(request_id) AS count
      FROM ${table}
      WHERE principal_type = 'api_key'
        AND org_id IS NOT NULL
        AND request_id IS NOT NULL
        AND key_environment = ${ClickHouseValue.literal(options.environment)}
        ${sandboxGate}
        AND timestamp >= parseDateTime64BestEffort(${ClickHouseValue.literal(options.from)}, 3, 'UTC')
        AND timestamp < parseDateTime64BestEffort(${ClickHouseValue.literal(options.to)}, 3, 'UTC')
      GROUP BY org_id, toStartOfHour(timestamp)
    `,
  )
  return rows.map((row) => ({
    bucketStart: String(row.bucketStart),
    count: ClickHouseValue.number(row.count),
    orgId: String(row.orgId),
  }))
}

type BillableCountRow = {
  bucketStart: string
  count: number | string
  orgId: string
}

/** Table name queries read from; the store scopes the database. */
const table = 'request_events'

type KeyRow = {
  averageDurationMs: number | string
  errors: number | string
  key_id: string
  requests: number | string
}

type LastUsedAtRow = {
  key_id: string
  lastUsedAt: string
}

type ErrorRow = {
  code: string
  requests: number | string
}

type RouteRow = {
  averageDurationMs: number | string
  errors: number | string
  requests: number | string
  route: string
}

type SeriesRow = {
  errors: number | string
  requests: number | string
  time: string
}

type StatusRow = {
  requests: number | string
  status: number | string
}

type TotalsRow = {
  averageDurationMs: number | string
  errors: number | string
  requests: number | string
}

/** Maps one request log entry to a stored row. */
function event(entry: Log.Entry, context: Metrics.cloudflare.Context): Table {
  const principal = entry.principal
  return {
    billing_active: principal?.type === 'api_key' ? Number(principal.billingActive === true) : null,
    chain_id: entry.chainId ?? null,
    duration_ms: entry.duration,
    environment: context.environment ?? 'unknown',
    error_code: entry.errorCode ?? null,
    key_environment: principal?.type === 'api_key' ? (principal.environment ?? null) : null,
    key_id: principal?.type === 'api_key' ? principal.id : null,
    method: entry.method,
    org_id: principal?.type === 'api_key' ? (principal.orgId ?? null) : null,
    project_id: principal?.type === 'api_key' ? (principal.projectId ?? null) : null,
    principal_type: principal?.type ?? 'unknown',
    query: entry.query ?? '',
    rate_limit_scope: entry.rateLimit?.scope ?? null,
    request_id: entry.requestId ?? null,
    route: entry.route,
    rpc_error_code: entry.rpc?.code ?? null,
    rpc_error_count: entry.rpc?.errors ?? 0,
    rpc_error_data_code: entry.rpc?.dataCode ?? null,
    service: context.service ?? 'unknown',
    status: entry.status,
    timestamp: ClickHouseValue.timestamp(),
    timings: entry.timings ?? {},
  }
}

function usageWhere(options: ProjectUsageOptions) {
  const attributions = options.attributions.length
    ? options.attributions
        .map((attribution) => `(${attributionWhere(attribution, options)})`)
        .join(' OR ')
    : '0'
  return [
    `principal_type = 'api_key'`,
    `(${attributions})`,
    `timestamp >= parseDateTime64BestEffort(${ClickHouseValue.literal(options.from)}, 3, 'UTC')`,
    `timestamp <= parseDateTime64BestEffort(${ClickHouseValue.literal(options.to)}, 3, 'UTC')`,
  ].join(' AND ')
}

function attributionWhere(attribution: Attribution, options: ProjectUsageOptions) {
  if (attribution.projectId === null) {
    if (!attribution.apiKeyIds?.length) return '0'
    return [
      `project_id IS NULL`,
      `key_id IN (${attribution.apiKeyIds.map(ClickHouseValue.literal).join(', ')})`,
      ...(attribution.environment === undefined
        ? []
        : [keyEnvironmentWhere(attribution.environment, true)]),
    ].join(' AND ')
  }

  // Absent `projectId` scopes to the whole organization (any project).
  if (attribution.projectId === undefined) {
    return [
      `org_id = ${ClickHouseValue.literal(options.orgId)}`,
      ...(attribution.apiKeyIds === undefined
        ? []
        : [`key_id IN (${attribution.apiKeyIds.map(ClickHouseValue.literal).join(', ')})`]),
      ...(attribution.environment === undefined
        ? []
        : [keyEnvironmentWhere(attribution.environment, false)]),
    ].join(' AND ')
  }

  return [
    `org_id = ${ClickHouseValue.literal(options.orgId)}`,
    `project_id = ${ClickHouseValue.literal(attribution.projectId)}`,
    ...(attribution.apiKeyIds === undefined
      ? []
      : [`key_id IN (${attribution.apiKeyIds.map(ClickHouseValue.literal).join(', ')})`]),
    ...(attribution.environment === undefined
      ? []
      : [keyEnvironmentWhere(attribution.environment, false)]),
  ].join(' AND ')
}

function keyEnvironmentWhere(environment: Environment, allowLegacyNull: boolean) {
  const exact = `key_environment = ${ClickHouseValue.literal(environment)}`
  if (!allowLegacyNull) return exact
  return `(${exact} OR key_environment IS NULL)`
}

namespace ClickHouseValue {
  export function literal(value: string) {
    return `'${value.replaceAll('\\', '\\\\').replaceAll("'", "\\'")}'`
  }

  export function number(value: number | string) {
    return Number(value)
  }

  export function timestamp() {
    return new Date().toISOString().replace('T', ' ').replace('Z', '')
  }
}