import type { Metrics } from '../Metrics.js'
import type * as Webhooks from '../Webhooks.js'
import type * as Log from './Log.js'
import * as WebhookDestination from './WebhookDestination.js'

/** Webhook event type, including the synthetic `ping` carried on test envelopes. */
export type EventType = Webhooks.Envelope['type']

/** Where a delivery attempt originated. */
export type Trigger = 'manual_retry' | 'ping' | 'queue'

/** Bounded step names inside one block-fanout scan, used by `webhook:scan-step`. */
export type ScanStep =
  | 'block_rpc'
  | 'enrich'
  | 'finalized_rpc'
  | 'logs_rpc'
  | 'match'
  | 'stage'
  | 'subscription_read'

/**
 * A webhook metric event, recorded by the {@link webhooks} sink. The
 * block-fanout scan path emits head and scan measurements; the delivery path
 * emits `webhook:delivery`.
 */
export type WebhookEvent =
  | {
      /** Tempo chain whose listener was checked. */
      chainId: number
      /** Heads refused in a row by the monotonic guard; zero on a delivering listener. */
      skippedHeads: number
      /** Identifies heads received and discarded rather than heads not received. */
      type: 'webhook:head-skipped'
    }
  | {
      chainId: number
      /** True once finality has deferred this block at least one round. */
      deferred: boolean
      /** Rows matched when dispatched; zero otherwise. */
      matches: number
      /** Block timestamp to scan completion; absent when the block was not read. */
      scanLagMs?: number | undefined
      /** Block timestamp to head observation; absent without a fresh stamp. */
      observationLagMs?: number | undefined
      /** How the head that enqueued this block was observed. */
      origin: string
      /** One scanned block's outcome. */
      outcome: 'deferred' | 'dispatched' | 'empty'
      /** Head observation to this scan attempt starting; absent without a stamp. */
      queueWaitMs?: number | undefined
      /** Identifies one block-fanout scan. */
      type: 'webhook:scan-block'
      /** Wall-clock work inside this scan attempt. */
      workMs: number
    }
  | {
      chainId: number
      /** Times this step ran across the batch, so the total can be normalized. */
      count: number
      /** Wall-clock duration of this step summed across the batch. */
      durationMs: number
      /** Bounded step name. */
      step: ScanStep
      /** Identifies per-step scan attribution. */
      type: 'webhook:scan-step'
    }
  | {
      chainId: number
      /** Age of the oldest tracked block, absent when none are pending. */
      oldestPendingAgeMs?: number | undefined
      /** Tracked blocks awaiting completion reports. */
      pendingCount: number
      /** Identifies the coordinator's tracked-scan backlog. */
      type: 'webhook:scan-backlog'
      /** Blocks between the chain head and the enqueue watermark. */
      watermarkLagBlocks?: number | undefined
    }
  | {
      chainId: number
      /** Blocks between the observed head and the finalized head. */
      lagBlocks: number
      /** Identifies the finality gate's settling distance. */
      type: 'webhook:scan-finality'
    }
  | {
      /** Atomic ledger claim outcome. */
      outcome: 'attempting' | 'claimed' | 'missing' | 'scheduled' | 'terminal'
      /** Identifies a delivery-obligation claim. */
      type: 'webhook:delivery-job-claim'
    }
  | {
      /** Obligations represented by this outcome. */
      count: number
      /** Whether staging created or found the obligation. */
      outcome: 'created' | 'existing'
      /** Identifies obligation staging ahead of Queue handoff. */
      type: 'webhook:delivery-job-ensure'
    }
  | {
      /** Delivery attempt ordinal: the ledger claim count on the queue path. */
      attempt: number
      envelope: Webhooks.Envelope
      /** Cloudflare Queue delivery attempt, absent outside the Queue path. */
      queueAttempt?: number | undefined
      result: Webhooks.Result
      subscription: Webhooks.Subscription
      timings?: {
        deliveryLogMs?: number | undefined
        envelopeToQueueMs?: number | undefined
        endToEndMs?: number | undefined
        /** Block timestamp to the attempt leaving; the total latency SLI. */
        eventToAttemptMs?: number | undefined
        eventToEnvelopeMs?: number | undefined
        /** Head observation to the attempt leaving; the controllable SLI. */
        observedToAttemptMs?: number | undefined
        /** Dequeue to the attempt leaving: claim, reads, signing. */
        preflightMs?: number | undefined
        processingMs?: number | undefined
        queueWaitMs?: number | undefined
        /** Response headers to bookkeeping complete; occupancy, not latency. */
        settleMs?: number | undefined
        stateWriteMs?: number | undefined
        subscriptionReadMs?: number | undefined
      }
      trigger: Trigger
      type: 'webhook:delivery'
    }

/** Records request (HTTP) metrics for one request-log entry. Built by {@link requests}. */
export type RequestSink = (entry: Log.Entry) => void

/** Records webhook reliability metrics. Built by {@link webhooks}. */
export type WebhookSink = {
  /** Ships buffered metrics (call once per scan batch / delivery). */
  flush(): void
  /** Records one webhook event. */
  record(event: WebhookEvent): void
}

/** Low-cardinality tags shared by every HTTP metric, derived from a request log entry. */
type RouteTags = {
  cache: 'hit' | 'miss'
  method: string
  principal_environment: 'none' | 'production' | 'sandbox'
  principal_type: 'api_key' | 'public' | 'session' | 'super_admin' | 'unknown'
  route: string
}

/**
 * Builds a request (HTTP) metrics sink over `backend`: emits the bounded Worker
 * metrics derived from one request-log entry, then flushes (one entry == one
 * request).
 *
 * Metric tags deliberately exclude caller ids, organization ids, request ids,
 * raw paths, and IP-derived identities — those belong in logs or analytics, not
 * StatsD-style series.
 */
export function requests(backend: Metrics): RequestSink {
  return (entry) => {
    const tags = routeTags(entry)

    backend.count('http_response_count', 1, {
      ...tags,
      ...(entry.errorCode === undefined ? {} : { error_code: entry.errorCode }),
      ...(entry.payment === undefined ? {} : { payment: entry.payment }),
      status: entry.status,
    })
    backend.histogram('http_response_duration_ms', entry.duration, tags)
    if (entry.status >= 500)
      backend.count('http_server_error_count', 1, {
        ...tags,
        ...(entry.errorCode === undefined ? {} : { error_code: entry.errorCode }),
        status: entry.status,
      })

    if (entry.rpc) {
      backend.count('rpc_response_error_count', entry.rpc.errors, {
        ...tags,
        rpc_error_code: rpcErrorCode(entry.rpc.code),
      })
      if ((entry.rpc.serverErrors ?? 0) > 0)
        backend.count('rpc_response_server_error_count', entry.rpc.serverErrors ?? 0, {
          ...tags,
          rpc_error_code: rpcErrorCode(entry.rpc.code),
        })
    }

    if (entry.status === 429 || entry.errorCode === 'rate_limit_exceeded')
      backend.count('http_rate_limit_count', 1, {
        ...tags,
        rate_limit_scope: entry.rateLimit?.scope ?? 'unknown',
        status: entry.status,
      })

    for (const [operation, duration] of Object.entries(entry.timings ?? {}))
      backend.histogram('upstream_duration_ms', duration, { ...tags, operation })

    for (const attempt of entry.fundingProviderAttempts ?? []) {
      const tags_attempt = {
        ...(attempt.failure === undefined ? {} : { failure: attempt.failure }),
        operation: attempt.operation,
        outcome: attempt.outcome,
        provider: attempt.id,
      }
      backend.count('funding_quote_provider_attempt_count', 1, tags_attempt)
      backend.histogram('funding_quote_provider_duration_ms', attempt.durationMs, tags_attempt)
    }

    if (entry.fundingDepositCountFailed)
      backend.count('funding_deposit_count_failure_count', 1, tags)

    if (entry.fundingTransferCountFailed)
      backend.count('funding_transfer_count_failure_count', 1, tags)

    for (const failure of entry.providerFailures ?? (entry.provider ? [entry.provider] : []))
      backend.count('upstream_failure_count', 1, {
        ...tags,
        chain_id: String(failure.chainId ?? 'unknown'),
        failure: failure.failure,
        operation: failure.operation,
        provider: failure.id,
        status: String(failure.status ?? 'unknown'),
      })

    if ((entry.sponsorship?.internalErrors ?? 0) > 0)
      backend.count('sponsorship_internal_error_count', entry.sponsorship?.internalErrors ?? 0, {
        ...tags,
        chain_id: String(entry.sponsorship?.chainId ?? 'mixed'),
        method: entry.sponsorship?.method ?? 'mixed',
      })

    if (entry.mpp) {
      const tags_mpp = {
        chain_id: String(entry.mpp.chainId),
        ...(entry.mpp.errorCode === undefined ? {} : { error_code: entry.mpp.errorCode }),
        fee_payer: entry.mpp.feePayer,
        operation: entry.mpp.operation,
        outcome: entry.mpp.outcome,
      }
      backend.count('mpp_relay_operation_count', 1, tags_mpp)
      backend.histogram('mpp_relay_operation_duration_ms', entry.duration, tags_mpp)
      if (entry.mpp.idempotency)
        backend.count('mpp_relay_idempotency_claim_count', 1, {
          chain_id: String(entry.mpp.chainId),
          outcome: entry.mpp.idempotency,
        })
    }

    backend.flush()
  }
}

/** Maps arbitrary upstream JSON-RPC codes into bounded metric tags. */
function rpcErrorCode(code: number | undefined) {
  if (code === undefined) return 'mixed'
  if ([-32700, -32600, -32601, -32602, -32603].includes(code)) return String(code)
  if (code >= -32099 && code <= -32000) return 'server'
  return 'other'
}

function routeTags(entry: Log.Entry): RouteTags {
  return {
    cache: entry.cache ?? 'miss',
    method: entry.method,
    principal_environment: entry.principal?.environment ?? 'none',
    principal_type: entry.principal?.type ?? 'unknown',
    route: entry.route,
  }
}

/** Bounded attempt-count buckets so the attempt tag never explodes cardinality. */
type AttemptBucket = '1' | '2' | '3-5' | '6-10' | 'gt10'

/** Low-cardinality delivery destination kind, tagged on `webhook:delivery`. */
type DestinationType = Webhooks.Subscription['destination']['type']

/** Low-cardinality delivery outcome class derived from an HTTP status / error. */
type StatusClass = '2xx' | '4xx' | '5xx' | 'network' | 'timeout' | 'unknown' | 'url_rejected'

/**
 * Builds a webhook delivery-reliability sink over `backend`. Tags stay
 * low-cardinality (`chain_id`, `event_type`, `destination_type`, `status_class`,
 * `attempt_bucket`, `failure`, `queue_attempt_bucket`, `outcome`, `stage`) — never subscription/org/event ids, which belong
 * in logs or analytics, not StatsD-style series.
 */
export function webhooks(backend: Metrics): WebhookSink {
  return {
    flush() {
      backend.flush()
    },
    record(event) {
      switch (event.type) {
        case 'webhook:head-skipped':
          backend.gauge('webhook_head_skipped_count', event.skippedHeads, {
            chain_id: String(event.chainId),
          })
          return
        case 'webhook:scan-block': {
          const tags = {
            chain_id: String(event.chainId),
            origin: event.origin,
            outcome: event.outcome,
          }
          backend.count('webhook_scan_block_count', 1, tags)
          if (event.matches > 0) backend.count('webhook_scan_match_count', event.matches, tags)
          backend.histogram('webhook_scan_work_ms', event.workMs, tags)
          // Deferred rounds are chain finality, not queue depth, so the wait
          // carries the distinction rather than silently averaging the two.
          if (event.queueWaitMs !== undefined)
            backend.histogram('webhook_scan_queue_wait_ms', event.queueWaitMs, {
              ...tags,
              deferred: String(event.deferred),
            })
          if (event.observationLagMs !== undefined)
            backend.histogram('webhook_scan_observation_lag_ms', event.observationLagMs, {
              chain_id: String(event.chainId),
              origin: event.origin,
            })
          // Block timestamp to scan completion: how far behind the chain the
          // scanner is, reported by every scanned block rather than only by
          // blocks that produce a delivery.
          if (event.scanLagMs !== undefined)
            backend.histogram('webhook_scan_lag_ms', event.scanLagMs, {
              chain_id: String(event.chainId),
            })
          return
        }
        case 'webhook:scan-step': {
          const tags = { chain_id: String(event.chainId), step: event.step }
          // Batch totals, so the count is required to read a per-invocation
          // mean: a ten-block batch otherwise reports ten times a one-block
          // batch at identical per-call latency.
          backend.histogram('webhook_scan_step_duration_ms', event.durationMs, tags)
          backend.count('webhook_scan_step_count', event.count, tags)
          return
        }
        case 'webhook:scan-backlog': {
          const tags = { chain_id: String(event.chainId) }
          backend.gauge('webhook_scan_pending_count', event.pendingCount, tags)
          backend.gauge('webhook_scan_oldest_pending_age_ms', event.oldestPendingAgeMs ?? 0, tags)
          if (event.watermarkLagBlocks !== undefined)
            backend.gauge('webhook_scan_watermark_lag_blocks', event.watermarkLagBlocks, tags)
          return
        }
        case 'webhook:scan-finality': {
          backend.gauge('webhook_scan_finality_lag_blocks', event.lagBlocks, {
            chain_id: String(event.chainId),
          })
          return
        }
        case 'webhook:delivery-job-claim': {
          backend.count('webhook_delivery_job_claim_count', 1, { outcome: event.outcome })
          return
        }
        case 'webhook:delivery-job-ensure': {
          backend.count('webhook_delivery_job_ensure_count', event.count, {
            outcome: event.outcome,
          })
          return
        }
        case 'webhook:delivery': {
          const { envelope, result, subscription, timings } = event
          const queueTags =
            event.queueAttempt === undefined
              ? {}
              : {
                  queue_attempt_bucket: attemptBucket(event.queueAttempt),
                  // Retryable results ack the message and schedule a ledger
                  // retry; the Queue transport never redelivers them.
                  queue_disposition: WebhookDestination.isRetryable(result)
                    ? 'scheduled_retry'
                    : 'acked',
                }
          const tags = {
            chain_id: String(envelope.chainId),
            destination_type: subscription.destination.type satisfies DestinationType,
            event_type: envelope.type,
          }
          const outcome = result.ok ? 'succeeded' : 'failed'
          backend.count('webhook_delivery_attempt_count', 1, {
            ...tags,
            attempt_bucket: attemptBucket(event.attempt),
            outcome,
            ...queueTags,
            status_class: statusClass(result),
            trigger: event.trigger,
          })
          if (result.durationMs !== undefined)
            backend.histogram('webhook_delivery_duration_ms', result.durationMs, {
              ...tags,
              outcome,
            })
          if (timings?.deliveryLogMs !== undefined)
            backend.histogram('webhook_delivery_log_write_ms', timings.deliveryLogMs, tags)
          if (timings?.preflightMs !== undefined)
            backend.histogram('webhook_delivery_preflight_ms', timings.preflightMs, tags)
          if (timings?.settleMs !== undefined)
            backend.histogram('webhook_delivery_settle_ms', timings.settleMs, tags)
          // The two headline latency SLIs, both ending when the attempt leaves
          // rather than when bookkeeping finishes. Attempt-bucketed for the
          // same reason as end-to-end: retries carry designed backoff.
          if (timings?.eventToAttemptMs !== undefined)
            backend.histogram('webhook_event_to_attempt_ms', timings.eventToAttemptMs, {
              ...tags,
              attempt_bucket: attemptBucket(event.attempt),
              outcome,
            })
          if (timings?.observedToAttemptMs !== undefined)
            backend.histogram('webhook_observed_to_attempt_ms', timings.observedToAttemptMs, {
              ...tags,
              attempt_bucket: attemptBucket(event.attempt),
              outcome,
            })
          if (timings?.envelopeToQueueMs !== undefined)
            backend.histogram(
              'webhook_delivery_envelope_to_queue_ms',
              timings.envelopeToQueueMs,
              tags,
            )
          if (timings?.endToEndMs !== undefined)
            backend.histogram('webhook_delivery_end_to_end_ms', timings.endToEndMs, {
              ...tags,
              // Unlimited-horizon ledger retries emit hour-plus samples;
              // the bucket keeps first-attempt percentiles interpretable.
              attempt_bucket: attemptBucket(event.attempt),
              outcome,
            })
          if (timings?.eventToEnvelopeMs !== undefined)
            backend.histogram(
              'webhook_delivery_event_to_envelope_ms',
              timings.eventToEnvelopeMs,
              tags,
            )
          if (timings?.processingMs !== undefined)
            backend.histogram('webhook_delivery_processing_ms', timings.processingMs, tags)
          if (timings?.queueWaitMs !== undefined)
            backend.histogram('webhook_delivery_queue_wait_ms', timings.queueWaitMs, {
              ...tags,
              outcome,
              ...queueTags,
            })
          if (timings?.stateWriteMs !== undefined)
            backend.histogram('webhook_delivery_state_write_ms', timings.stateWriteMs, tags)
          if (timings?.subscriptionReadMs !== undefined)
            backend.histogram(
              'webhook_delivery_subscription_read_ms',
              timings.subscriptionReadMs,
              tags,
            )
          return
        }
      }
    },
  }
}

/** Buckets a 1-based attempt number into a bounded set of attempt labels. */
function attemptBucket(attempt: number): AttemptBucket {
  if (attempt <= 1) return '1'
  if (attempt === 2) return '2'
  if (attempt <= 5) return '3-5'
  if (attempt <= 10) return '6-10'
  return 'gt10'
}

/** Maps a delivery result to a low-cardinality status class. */
function statusClass(result: Webhooks.Result): StatusClass {
  if (result.ok) return '2xx'
  const { error, status } = result
  if (status !== undefined) {
    if (status >= 300 && status < 400) return 'url_rejected'
    if (status >= 400 && status < 500) return '4xx'
    if (status >= 500) return '5xx'
  }
  if (error === undefined) return 'unknown'
  if (/redirect/i.test(error)) return 'url_rejected'
  if (/timeout|abort/i.test(error)) return 'timeout'
  return 'network'
}