defmodule DaProductApp.Telemetry.MetricsCollector do
  @moduledoc """
  Production metrics collector for CloudLayer observability.

  A supervised GenServer that attaches to `:telemetry` events and maintains
  in-memory counters/gauges/histograms for:

    * HTTP requests (count, status, duration, active requests)
    * Ecto queries (count, duration, errors)
    * BEAM VM (memory, processes, scheduler utilization, uptime)
    * Payment flow (total, success, failure, duration)
    * External provider APIs - Alipay/AANI (total, failure, duration)

  Access the current snapshot via `get_metrics_snapshot/0`, consumed by
  `DaProductApp.Telemetry.PrometheusExporter`.
  """

  use GenServer
  require Logger
  import Ecto.Query, only: [from: 2]

  alias DaProductApp.Groups.Group

  @beam_poll_interval 15_000
  # Merchant counts change slowly (admin-driven, not per-request) - polled on
  # a separate, much longer timer than BEAM/DB-pool sampling, and never on
  # the /monitoring/metrics scrape path.
  @merchant_poll_interval :timer.minutes(5)
  # Short one-off delay before the very first merchant poll, so
  # merchant_total/merchant_active don't stay absent for a full 5 minutes
  # after boot.
  @merchant_initial_poll_delay 1_000
  @histogram_max_samples 1_000
  @histogram_buckets [0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10]

  # Byte-size buckets for http_request_size_bytes / http_response_size_bytes.
  # Unlike @histogram_buckets above (raw-sample reservoir, see note below),
  # these two metrics are stored as true cumulative Prometheus histograms:
  # buckets/sum/count only ever grow, no sample list, no eviction.
  @size_buckets_bytes [64, 256, 1024, 4096, 16384, 65536, 262_144, 1_048_576]

  # Module-namespaced process-dictionary key used to hand the response size
  # computed at [:phoenix, :endpoint, :stop] (real resp_body, no route) over
  # to [:phoenix, :router_dispatch, :stop] (route known, resp_body already
  # nil'd by the Bandit adapter) within the same request-handling process.
  # :telemetry.execute/3 runs handlers synchronously in the calling process,
  # so this never crosses request/process boundaries.
  @response_size_pdict_key {__MODULE__, :http_response_size}

  # Fixed, source-verified allowlist of stably-registered, operationally
  # critical processes polled for beam_mailbox_queue_length. Each entry's
  # registration is confirmed at its own `start_link`:
  #   - DaProductApp.Telemetry.MetricsCollector: name: __MODULE__ (this file)
  #   - DaProductApp.GlobalSettings: name: @name (== __MODULE__), global_settings.ex
  #   - DaProductApp.TransactionRuleCache: name: @name (== __MODULE__), transaction_rule_cache.ex
  # MQTT is deliberately excluded: DaProductApp.MQTT.Supervisor and
  # DaProductApp.MQTT.ClientSupervisor are stably named, but they are plain
  # supervisors that don't process the actual MQTT message traffic: the
  # Tortoise client process that does is registered under Tortoise's own
  # internal registry (keyed by client_id), not a plain atom resolvable via
  # Process.whereis/1 - including it would mean guessing/reaching into
  # Tortoise internals, which is exactly what was ruled out.
  @mailbox_allowlist [
    {DaProductApp.Telemetry.MetricsCollector, "metrics_collector"},
    {DaProductApp.GlobalSettings, "global_settings"},
    {DaProductApp.TransactionRuleCache, "transaction_rule_cache"}
  ]

  # Repos whose DBConnection pools are polled for db_pool_size /
  # db_pool_available / db_pool_checkout_queue_length, alongside the label
  # used for each in the exported metrics.
  @monitored_repos [
    {DaProductApp.Repo, "main"},
    {DaProductApp.Repos.ShukriaMmsRepo, "shukria_mms"}
  ]

  @payment_events [
    [:da_product_app, :payment, :completed],
    [:da_product_app, :payment, :started],
    [:da_product_app, :payment, :finalized]
  ]
  @external_api_events [
    [:da_product_app, :external_api, :call],
    [:da_product_app, :external_api, :timeout],
    [:da_product_app, :external_api, :error]
  ]

  @results ["success", "failure"]

  # Fixed, known {provider, transaction_type} combinations actually supported
  # today — seeded at 0 so their metric families never show HELP/TYPE-only.
  # Deliberately excludes unsupported combinations (e.g. AANI cancel/refund,
  # which have no code path at all).
  @payment_seed_combinations [
    {"alipay", "purchase"},
    {"alipay", "cancel"},
    {"alipay", "refund"},
    {"aani", "purchase"}
  ]

  # Fixed, known {provider, operation} combinations for external provider APIs.
  @external_api_seed_combinations [
    {"alipay", "generate"},
    {"alipay", "inquire"},
    {"alipay", "cancel"},
    {"alipay", "refund"},
    {"aani", "generate"},
    {"aani", "status"}
  ]
  # DaProductApp.Repo and DaProductApp.Repos.ShukriaMmsRepo each emit their
  # own event, derived from their module name by Ecto.
  @db_events [
    [:da_product_app, :repo, :query],
    [:da_product_app, :repos, :shukria_mms_repo, :query]
  ]
  @http_events [
    [:phoenix, :endpoint, :start],
    [:phoenix, :endpoint, :stop],
    [:phoenix, :router_dispatch, :stop]
  ]

  # ---------------------------------------------------------------------------
  # Public API
  # ---------------------------------------------------------------------------

  def start_link(opts \\ []) do
    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
  end

  @doc "Return a snapshot of all current metric values for the Prometheus exporter."
  def get_metrics_snapshot do
    GenServer.call(__MODULE__, :get_metrics_snapshot)
  end

  # ---------------------------------------------------------------------------
  # GenServer init
  # ---------------------------------------------------------------------------

  @impl true
  def init(_opts) do
    :telemetry.attach_many(
      "cloudlayer-metrics-collector",
      @payment_events ++ @external_api_events ++ @db_events ++ @http_events,
      &handle_telemetry_event/4,
      nil
    )

    state = %{
      # --- payment ---
      payment_totals: seed_payment_totals(),
      # {provider, transaction_type, result} => count
      payment_durations: seed_payment_durations(),
      # {provider, transaction_type} => [ms, ...]
      payment_started_totals: %{},
      # {provider, transaction_type} => count
      payment_finalized_totals: %{},
      # {provider, transaction_type, result} => count
      payment_failure_reason_totals: %{},
      # {provider, transaction_type, reason} => count
      payment_amount_totals: %{},
      # {provider, transaction_type, currency} => cumulative amount
      payment_amount_success_totals: %{},
      # {provider, transaction_type, currency} => cumulative amount
      payment_amount_failed_totals: %{},
      # {provider, transaction_type, currency} => cumulative amount

      # --- external provider APIs ---
      external_api_totals: seed_external_api_totals(),
      # {provider, operation, result} => count
      external_api_durations: seed_external_api_durations(),
      # {provider, operation} => [ms, ...]
      external_api_timeout_totals: %{},
      # {provider, operation} => count
      external_api_error_reason_totals: %{},
      # {provider, operation, reason} => count

      # --- http ---
      http_totals: %{},
      # {route, method, status} => count
      http_durations: %{},
      # {route, method} => [ms, ...]
      active_requests: 0,
      http_request_size_totals: %{},
      # {route, method} => cumulative %{buckets:, sum:, count:} (bytes)
      http_response_size_totals: %{},
      # {route, method} => cumulative %{buckets:, sum:, count:} (bytes)

      # --- db (keyed by repo: "main" | "shukria_mms") ---
      db_queries_total: %{},
      db_query_errors_total: %{},
      db_query_durations: %{},
      db_slow_queries_total: %{},
      # repo_label => count
      db_pool_size: %{},
      # repo_label => configured pool_size (gauge)
      db_pool_available: %{},
      # repo_label => ready_conn_count (gauge, from DBConnection.get_connection_metrics/2)
      db_pool_checkout_queue_length: %{},
      # repo_label => checkout_queue_length (gauge, from DBConnection.get_connection_metrics/2)

      # --- merchants (cached gauges, polled every 5 min - never per-scrape) ---
      merchant_total: nil,
      # nil until the first successful poll - never a guessed 0
      merchant_active: nil,
      # nil until the first successful poll - never a guessed 0

      # --- beam vm ---
      beam_vm: initial_beam_vm_snapshot(),
      started_at: System.monotonic_time(),
      start_time_unix_seconds: DateTime.utc_now() |> DateTime.to_unix(:second),
      last_updated: DateTime.utc_now()
    }

    schedule_beam_poll()
    Process.send_after(self(), :poll_merchants, @merchant_initial_poll_delay)

    Logger.info("MetricsCollector started")
    {:ok, state}
  end

  # ---------------------------------------------------------------------------
  # Seed helpers — pre-register known/fixed label combinations at 0 so their
  # metric families always expose sample lines, not just HELP/TYPE.
  # ---------------------------------------------------------------------------

  defp seed_payment_totals do
    for {provider, transaction_type} <- @payment_seed_combinations,
        result <- @results,
        into: %{} do
      {{provider, transaction_type, result}, 0}
    end
  end

  defp seed_payment_durations do
    for {provider, transaction_type} <- @payment_seed_combinations, into: %{} do
      {{provider, transaction_type}, []}
    end
  end

  defp seed_external_api_totals do
    for {provider, operation} <- @external_api_seed_combinations,
        result <- @results,
        into: %{} do
      {{provider, operation, result}, 0}
    end
  end

  defp seed_external_api_durations do
    for {provider, operation} <- @external_api_seed_combinations, into: %{} do
      {{provider, operation}, []}
    end
  end

  # ---------------------------------------------------------------------------
  # Telemetry event handlers (called from :telemetry, must be fast)
  # ---------------------------------------------------------------------------

  @doc false
  def handle_telemetry_event(
        [:da_product_app, :payment, :completed],
        %{duration_ms: duration_ms},
        %{provider: provider, result: result, transaction_type: transaction_type} = metadata,
        _config
      ) do
    GenServer.cast(
      __MODULE__,
      {:payment_completed, provider, transaction_type, result, duration_ms, metadata}
    )
  end

  def handle_telemetry_event(
        [:da_product_app, :payment, :started],
        _measurements,
        %{provider: provider, transaction_type: transaction_type},
        _config
      ) do
    GenServer.cast(__MODULE__, {:payment_started, provider, transaction_type})
  end

  def handle_telemetry_event(
        [:da_product_app, :payment, :finalized],
        _measurements,
        %{provider: provider, result: result, transaction_type: transaction_type},
        _config
      ) do
    GenServer.cast(__MODULE__, {:payment_finalized, provider, result, transaction_type})
  end

  def handle_telemetry_event(
        [:da_product_app, :external_api, :call],
        %{duration_ms: duration_ms},
        %{provider: provider, operation: operation, result: result},
        _config
      ) do
    GenServer.cast(__MODULE__, {:external_api_call, provider, operation, result, duration_ms})
  end

  def handle_telemetry_event(
        [:da_product_app, :external_api, :timeout],
        _measurements,
        %{provider: provider, operation: operation},
        _config
      ) do
    GenServer.cast(__MODULE__, {:external_api_timeout, provider, operation})
  end

  def handle_telemetry_event(
        [:da_product_app, :external_api, :error],
        _measurements,
        %{provider: provider, operation: operation, reason: reason},
        _config
      ) do
    GenServer.cast(__MODULE__, {:external_api_error, provider, operation, reason})
  end

  def handle_telemetry_event(
        [:da_product_app, :repo, :query],
        %{total_time: total_time},
        metadata,
        _config
      ) do
    dispatch_db_query_event("main", total_time, metadata)
  end

  def handle_telemetry_event(
        [:da_product_app, :repos, :shukria_mms_repo, :query],
        %{total_time: total_time},
        metadata,
        _config
      ) do
    dispatch_db_query_event("shukria_mms", total_time, metadata)
  end

  def handle_telemetry_event([:phoenix, :endpoint, :start], _measurements, _metadata, _config) do
    # Defensive cleanup: clear any stale response-size value left behind by a
    # prior request on this same process (e.g. one that crashed before
    # reaching router_dispatch:stop) before this request starts writing its own.
    Process.delete(@response_size_pdict_key)
    GenServer.cast(__MODULE__, :request_started)
  end

  def handle_telemetry_event([:phoenix, :endpoint, :stop], _measurements, %{conn: conn}, _config) do
    # Endpoint-level Plug.Telemetry fires its :stop event from inside
    # Plug.Conn.register_before_send, i.e. before the adapter (Bandit) sends
    # the response and nils out conn.resp_body. This is the only point where
    # the real response body/content-length is still inspectable - stash it
    # here, in the process dictionary of this request's own process, since
    # router_dispatch:stop (which knows the route) fires later with an
    # already-nulled resp_body.
    Process.put(@response_size_pdict_key, response_size_from_conn(conn))
    GenServer.cast(__MODULE__, :request_finished)
  end

  def handle_telemetry_event(
        [:phoenix, :router_dispatch, :stop],
        %{duration: duration},
        %{route: route, conn: conn},
        _config
      ) do
    duration_ms = System.convert_time_unit(duration, :native, :millisecond)
    request_size = request_size_from_conn(conn)
    # Read-and-clear: retrieves the value stashed by endpoint:stop (if any)
    # and removes it so it can never be read again - one response is counted
    # at most once. Do NOT recompute from `conn` here - it's already nil'd.
    response_size = Process.delete(@response_size_pdict_key)

    GenServer.cast(
      __MODULE__,
      {:http_request, route, conn.method, conn.status, duration_ms, request_size, response_size}
    )
  end

  def handle_telemetry_event(_event, _measurements, _metadata, _config), do: :ok

  defp dispatch_db_query_event(repo_label, total_time, metadata) do
    error? = match?({:error, _}, Map.get(metadata, :result))
    duration_ms = System.convert_time_unit(total_time, :native, :millisecond)
    slow? = duration_ms >= slow_query_threshold_ms()
    GenServer.cast(__MODULE__, {:db_query, repo_label, error?, duration_ms, slow?})
  end

  # Read at call time (once per query event) rather than cached, so the
  # threshold stays genuinely runtime-configurable without a restart.
  # Application.get_env/3 is an ETS-backed lookup - cheap enough to call
  # per-query without meaningful overhead; no separate caching layer needed.
  defp slow_query_threshold_ms do
    Application.get_env(:da_product_app, :slow_query_threshold_ms, 500)
  end

  # ---------------------------------------------------------------------------
  # GenServer handle_call
  # ---------------------------------------------------------------------------

  @impl true
  def handle_call(:get_metrics_snapshot, _from, state) do
    uptime_seconds =
      System.convert_time_unit(
        System.monotonic_time() - state.started_at,
        :native,
        :second
      )

    snapshot = %{
      payment: %{
        totals: state.payment_totals,
        duration_seconds: histogram_summaries_seconds(state.payment_durations),
        started_totals: state.payment_started_totals,
        finalized_totals: state.payment_finalized_totals,
        failure_reason_totals: state.payment_failure_reason_totals,
        amount_totals: state.payment_amount_totals,
        amount_success_totals: state.payment_amount_success_totals,
        amount_failed_totals: state.payment_amount_failed_totals
      },
      external_api: %{
        totals: state.external_api_totals,
        duration_seconds: histogram_summaries_seconds(state.external_api_durations),
        timeout_totals: state.external_api_timeout_totals,
        error_reason_totals: state.external_api_error_reason_totals
      },
      http: %{
        totals: state.http_totals,
        duration_seconds: histogram_summaries_seconds(state.http_durations),
        active_requests: state.active_requests,
        request_size_bytes: state.http_request_size_totals,
        response_size_bytes: state.http_response_size_totals
      },
      db: %{
        queries_total: state.db_queries_total,
        errors_total: state.db_query_errors_total,
        duration_seconds: histogram_summaries_seconds(state.db_query_durations),
        slow_queries_total: state.db_slow_queries_total,
        pool_size: state.db_pool_size,
        pool_available: state.db_pool_available,
        pool_checkout_queue_length: state.db_pool_checkout_queue_length
      },
      beam_vm: state.beam_vm,
      app: %{
        version: app_version(),
        uptime_seconds: uptime_seconds,
        start_time_unix_seconds: state.start_time_unix_seconds
      },
      merchants: %{
        total: state.merchant_total,
        active: state.merchant_active
      },
      last_updated: state.last_updated
    }

    {:reply, snapshot, state}
  end

  # ---------------------------------------------------------------------------
  # GenServer handle_cast
  # ---------------------------------------------------------------------------

  @impl true
  def handle_cast(
        {:payment_completed, provider, transaction_type, result, duration_ms, metadata},
        state
      ) do
    key = {provider, transaction_type, Atom.to_string(result)}
    new_totals = Map.update(state.payment_totals, key, 1, &(&1 + 1))

    new_durations =
      append_sample(state.payment_durations, {provider, transaction_type}, duration_ms / 1.0)

    new_failure_reason_totals =
      apply_failure_reason(
        state.payment_failure_reason_totals,
        provider,
        transaction_type,
        result,
        metadata
      )

    {new_amount_totals, new_amount_success_totals, new_amount_failed_totals} =
      apply_amount(
        {state.payment_amount_totals, state.payment_amount_success_totals,
         state.payment_amount_failed_totals},
        provider,
        transaction_type,
        result,
        metadata
      )

    {:noreply,
     %{
       state
       | payment_totals: new_totals,
         payment_durations: new_durations,
         payment_failure_reason_totals: new_failure_reason_totals,
         payment_amount_totals: new_amount_totals,
         payment_amount_success_totals: new_amount_success_totals,
         payment_amount_failed_totals: new_amount_failed_totals,
         last_updated: DateTime.utc_now()
     }}
  end

  @impl true
  def handle_cast({:payment_started, provider, transaction_type}, state) do
    key = {provider, transaction_type}
    new_totals = Map.update(state.payment_started_totals, key, 1, &(&1 + 1))

    {:noreply, %{state | payment_started_totals: new_totals, last_updated: DateTime.utc_now()}}
  end

  @impl true
  def handle_cast({:payment_finalized, provider, result, transaction_type}, state) do
    key = {provider, result, transaction_type}
    new_totals = Map.update(state.payment_finalized_totals, key, 1, &(&1 + 1))

    {:noreply, %{state | payment_finalized_totals: new_totals, last_updated: DateTime.utc_now()}}
  end

  @impl true
  def handle_cast({:external_api_call, provider, operation, result, duration_ms}, state) do
    key = {provider, operation, Atom.to_string(result)}
    new_totals = Map.update(state.external_api_totals, key, 1, &(&1 + 1))

    new_durations =
      append_sample(state.external_api_durations, {provider, operation}, duration_ms / 1.0)

    {:noreply,
     %{
       state
       | external_api_totals: new_totals,
         external_api_durations: new_durations,
         last_updated: DateTime.utc_now()
     }}
  end

  @impl true
  def handle_cast({:external_api_timeout, provider, operation}, state) do
    key = {provider, operation}
    new_totals = Map.update(state.external_api_timeout_totals, key, 1, &(&1 + 1))

    {:noreply,
     %{state | external_api_timeout_totals: new_totals, last_updated: DateTime.utc_now()}}
  end

  @impl true
  def handle_cast({:external_api_error, provider, operation, reason}, state) do
    key = {provider, operation, reason}
    new_totals = Map.update(state.external_api_error_reason_totals, key, 1, &(&1 + 1))

    {:noreply,
     %{state | external_api_error_reason_totals: new_totals, last_updated: DateTime.utc_now()}}
  end

  @impl true
  def handle_cast({:db_query, repo_label, error?, duration_ms, slow?}, state) do
    new_totals = Map.update(state.db_queries_total, repo_label, 1, &(&1 + 1))

    new_errors =
      if error? do
        Map.update(state.db_query_errors_total, repo_label, 1, &(&1 + 1))
      else
        state.db_query_errors_total
      end

    new_durations = append_sample(state.db_query_durations, repo_label, duration_ms / 1.0)

    new_slow_queries =
      if slow? do
        Map.update(state.db_slow_queries_total, repo_label, 1, &(&1 + 1))
      else
        state.db_slow_queries_total
      end

    {:noreply,
     %{
       state
       | db_queries_total: new_totals,
         db_query_errors_total: new_errors,
         db_query_durations: new_durations,
         db_slow_queries_total: new_slow_queries,
         last_updated: DateTime.utc_now()
     }}
  end

  @impl true
  def handle_cast(:request_started, state) do
    {:noreply, %{state | active_requests: state.active_requests + 1}}
  end

  @impl true
  def handle_cast(:request_finished, state) do
    {:noreply, %{state | active_requests: max(0, state.active_requests - 1)}}
  end

  @impl true
  def handle_cast(
        {:http_request, route, method, status, duration_ms, request_size, response_size},
        state
      ) do
    method_label = status_label(method)
    size_key = {route, method_label}
    key = {route, method_label, status_label(status)}
    new_totals = Map.update(state.http_totals, key, 1, &(&1 + 1))

    new_durations = append_sample(state.http_durations, size_key, duration_ms / 1.0)

    new_request_size_totals =
      observe_byte_histogram(state.http_request_size_totals, size_key, request_size)

    new_response_size_totals =
      observe_byte_histogram(state.http_response_size_totals, size_key, response_size)

    {:noreply,
     %{
       state
       | http_totals: new_totals,
         http_durations: new_durations,
         http_request_size_totals: new_request_size_totals,
         http_response_size_totals: new_response_size_totals,
         last_updated: DateTime.utc_now()
     }}
  end

  defp apply_failure_reason(totals, _provider, _transaction_type, :success, _metadata), do: totals

  defp apply_failure_reason(totals, provider, transaction_type, :failure, metadata) do
    case Map.get(metadata, :failure_reason) do
      nil -> totals
      reason -> Map.update(totals, {provider, transaction_type, reason}, 1, &(&1 + 1))
    end
  end

  defp apply_amount(totals, provider, transaction_type, result, metadata) do
    with amount when is_number(amount) <- Map.get(metadata, :amount),
         currency when is_binary(currency) <- Map.get(metadata, :currency) do
      {amount_totals, success_totals, failed_totals} = totals
      key = {provider, transaction_type, currency}
      new_amount_totals = Map.update(amount_totals, key, amount, &(&1 + amount))

      case result do
        :success ->
          {new_amount_totals, Map.update(success_totals, key, amount, &(&1 + amount)),
           failed_totals}

        :failure ->
          {new_amount_totals, success_totals,
           Map.update(failed_totals, key, amount, &(&1 + amount))}
      end
    else
      _ -> totals
    end
  end

  # ---------------------------------------------------------------------------
  # GenServer handle_info — BEAM VM poll timer
  # ---------------------------------------------------------------------------

  @impl true
  def handle_info(:poll_beam_vm, state) do
    new_beam = sample_beam_vm()

    {new_pool_size, new_pool_available, new_pool_checkout_queue_length} =
      sample_db_pools(
        state.db_pool_size,
        state.db_pool_available,
        state.db_pool_checkout_queue_length
      )

    schedule_beam_poll()

    {:noreply,
     %{
       state
       | beam_vm: new_beam,
         db_pool_size: new_pool_size,
         db_pool_available: new_pool_available,
         db_pool_checkout_queue_length: new_pool_checkout_queue_length,
         last_updated: DateTime.utc_now()
     }}
  end

  @impl true
  def handle_info(:poll_merchants, state) do
    new_state =
      case safe_merchant_counts() do
        {:ok, total, active} ->
          %{
            state
            | merchant_total: total,
              merchant_active: active,
              last_updated: DateTime.utc_now()
          }

        :error ->
          Logger.warning(
            "MetricsCollector: could not read merchant counts - keeping last known snapshot"
          )

          state
      end

    schedule_merchant_poll()
    {:noreply, new_state}
  end

  # ---------------------------------------------------------------------------
  # Merchant counts (merchant_total / merchant_active)
  #
  # Confirmed source: `DaProductApp.Groups.Group` (`lib/da_product_app/groups/group.ex`),
  # table "groups" (`priv/repo/migrations/20250515112508_create_groups.exs`),
  # via the default Repo (DaProductApp.Repo - the only repo with migrations
  # in this app). Confirmed active-status literal: "active" (both the
  # schema's `field :status, :string, default: "active"` and the migration's
  # `status VARCHAR(255) DEFAULT 'active'` agree).
  #
  # Polled on a separate 5-minute timer, never on the /monitoring/metrics
  # scrape path and never on the 15s BEAM/DB-pool timer - merchant counts
  # change slowly (admin-driven), so a much longer interval is enough and
  # keeps DB load negligible. Both counts use Repo.aggregate/2,3 (COUNT(*)
  # pushed down to SQL - rows are never loaded into memory to be counted).
  # Any failure keeps the last known good values rather than publishing a
  # guessed 0; before the first successful poll, both stay `nil` and are
  # omitted from the exported text entirely.
  # ---------------------------------------------------------------------------

  defp safe_merchant_counts do
    total = DaProductApp.Repo.aggregate(Group, :count)
    active = DaProductApp.Repo.aggregate(from(g in Group, where: g.status == "active"), :count)
    {:ok, total, active}
  rescue
    _ -> :error
  catch
    :exit, _ -> :error
  end

  defp schedule_merchant_poll,
    do: Process.send_after(self(), :poll_merchants, @merchant_poll_interval)

  # ---------------------------------------------------------------------------
  # DB pool sampling (db_pool_size / db_pool_available / db_pool_checkout_queue_length)
  #
  # Polled on the same cadence as the BEAM VM sampler above, not per-scrape:
  # pool state doesn't change per-request, and polling avoids adding overhead
  # to /monitoring/metrics itself.
  #
  # IMPORTANT: the repo module's registered name (e.g. `DaProductApp.Repo`)
  # belongs to `Ecto.Repo.Supervisor`, NOT the DBConnection pool. Calling
  # `DBConnection.get_connection_metrics/2` directly on the repo module sends
  # a GenServer.call the supervisor cannot handle and crashes it. The real
  # pool PID must be resolved via the public, documented
  # `Ecto.Adapter.lookup_meta/1` (backed by `Ecto.Repo.Registry`'s public ETS
  # table + a non-blocking `GenServer.whereis/1` - no message sent to the
  # repo/supervisor/pool during lookup itself, no supervisor child
  # enumeration, no private GenServer state inspection). The pid is resolved
  # fresh on every poll - never cached - since a repo restart replaces it.
  #
  # Defensive per requirement: any failure (repo not registered, dead pid,
  # unexpected metrics shape, or the pool call itself raising/exiting) keeps
  # that repo's last known good gauge values rather than publishing a guess
  # or crashing the collector.
  # ---------------------------------------------------------------------------

  defp sample_db_pools(prev_sizes, prev_available, prev_queue_lengths) do
    Enum.reduce(@monitored_repos, {prev_sizes, prev_available, prev_queue_lengths}, fn
      {repo, label}, {sizes, available, queue_lengths} ->
        case safe_pool_metrics(repo) do
          {:ok, pool_size, ready_conn_count, checkout_queue_length} ->
            {
              Map.put(sizes, label, pool_size),
              Map.put(available, label, ready_conn_count),
              Map.put(queue_lengths, label, checkout_queue_length)
            }

          :error ->
            Logger.warning(
              "MetricsCollector: could not read DBConnection pool metrics for #{label} repo - keeping last known snapshot"
            )

            {sizes, available, queue_lengths}
        end
    end)
  end

  defp safe_pool_metrics(repo) do
    pool_size = repo.config()[:pool_size]

    with {:ok, %{pid: pool_pid}} <- safe_lookup_meta(repo),
         true <- is_pid(pool_pid) and Process.alive?(pool_pid),
         metrics when is_list(metrics) <- safe_get_connection_metrics(pool_pid),
         [%{ready_conn_count: ready_conn_count, checkout_queue_length: checkout_queue_length} | _] <-
           metrics,
         true <- is_integer(pool_size) do
      {:ok, pool_size, ready_conn_count, checkout_queue_length}
    else
      _ -> :error
    end
  end

  # Ecto.Adapter.lookup_meta/1 raises (a plain, catchable exception in this
  # process) if the repo isn't registered - never sends a message to any
  # other process, so there is nothing here that can crash the repo.
  defp safe_lookup_meta(repo) do
    {:ok, Ecto.Adapter.lookup_meta(repo)}
  rescue
    _ -> :error
  end

  # The only call in this module that message-passes to the pool itself.
  # Guarded by the is_pid/Process.alive? check above, but still wrapped
  # defensively in case the pool exits between that check and this call.
  defp safe_get_connection_metrics(pool_pid) do
    DBConnection.get_connection_metrics(pool_pid)
  rescue
    _ -> :error
  catch
    :exit, _ -> :error
  end

  # ---------------------------------------------------------------------------
  # BEAM VM sampling (mirrors Mercury's SwitchMetrics collector)
  # ---------------------------------------------------------------------------

  defp initial_beam_vm_snapshot do
    %{
      memory_total_bytes: 0,
      memory_processes_bytes: 0,
      memory_binary_bytes: 0,
      memory_ets_bytes: 0,
      memory_atom_bytes: 0,
      process_count: 0,
      process_limit: 0,
      run_queue_length: 0,
      scheduler_count: 0,
      port_count: 0,
      uptime_seconds: 0,
      scheduler_utilization: %{},
      gc_total: 0,
      reductions_total: 0,
      io_input_bytes: 0,
      io_output_bytes: 0,
      mailbox_lengths: %{},
      mailbox_max: nil
    }
  end

  defp sample_beam_vm do
    mem = :erlang.memory()
    {wall_ms, _} = :erlang.statistics(:wall_clock)
    {gc_total, _words_reclaimed, _} = :erlang.statistics(:garbage_collection)
    {reductions_total, _since_last_call} = :erlang.statistics(:reductions)
    {{:input, io_input_bytes}, {:output, io_output_bytes}} = :erlang.statistics(:io)
    mailbox_lengths = sample_mailbox_lengths()

    %{
      memory_total_bytes: Keyword.get(mem, :total, 0),
      memory_processes_bytes: Keyword.get(mem, :processes, 0),
      memory_binary_bytes: Keyword.get(mem, :binary, 0),
      memory_ets_bytes: Keyword.get(mem, :ets, 0),
      memory_atom_bytes: Keyword.get(mem, :atom, 0),
      process_count: :erlang.system_info(:process_count),
      process_limit: :erlang.system_info(:process_limit),
      run_queue_length: :erlang.statistics(:run_queue),
      scheduler_count: :erlang.system_info(:schedulers_online),
      port_count: :erlang.system_info(:port_count),
      uptime_seconds: div(wall_ms, 1_000),
      scheduler_utilization: sample_scheduler_utilization(),
      gc_total: gc_total,
      reductions_total: reductions_total,
      io_input_bytes: io_input_bytes,
      io_output_bytes: io_output_bytes,
      mailbox_lengths: mailbox_lengths,
      mailbox_max: mailbox_max(mailbox_lengths)
    }
  end

  # ---------------------------------------------------------------------------
  # beam_mailbox_queue_length / beam_mailbox_queue_length_max
  #
  # Only the fixed @mailbox_allowlist is ever queried - never
  # :erlang.processes/0. A process that isn't currently registered (not yet
  # started, or mid-restart) simply contributes no sample rather than a
  # guessed 0; beam_mailbox_queue_length_max is computed only from whatever
  # was successfully sampled this tick, and is `nil` (omitted from export)
  # if none were.
  # ---------------------------------------------------------------------------

  defp sample_mailbox_lengths do
    Enum.reduce(@mailbox_allowlist, %{}, fn {name, label}, acc ->
      case safe_mailbox_length(name) do
        {:ok, length} -> Map.put(acc, label, length)
        :error -> acc
      end
    end)
  end

  defp safe_mailbox_length(name) do
    case Process.whereis(name) do
      nil ->
        :error

      pid ->
        case Process.info(pid, :message_queue_len) do
          {:message_queue_len, length} -> {:ok, length}
          nil -> :error
        end
    end
  rescue
    _ -> :error
  end

  defp mailbox_max(lengths) when map_size(lengths) == 0, do: nil
  defp mailbox_max(lengths), do: lengths |> Map.values() |> Enum.max()

  defp sample_scheduler_utilization do
    try do
      :erlang.system_flag(:scheduler_wall_time, true)
      t1 = :erlang.statistics(:scheduler_wall_time)
      Process.sleep(100)
      t2 = :erlang.statistics(:scheduler_wall_time)

      Enum.reduce(Enum.zip(Enum.sort(t1), Enum.sort(t2)), %{}, fn
        {{id, a1, t1_val}, {id, a2, t2_val}}, acc ->
          util =
            if t2_val - t1_val > 0,
              do: Float.round((a2 - a1) / (t2_val - t1_val), 4),
              else: 0.0

          Map.put(acc, id, util)
      end)
    rescue
      _ -> %{}
    end
  end

  # ---------------------------------------------------------------------------
  # Histogram helpers (mirrors Mercury's percentile summarizer)
  # ---------------------------------------------------------------------------

  # Nil-safe label conversion for Prometheus label values (e.g. conn.status
  # can be nil for a halted connection that never sent a response).
  defp status_label(nil), do: "unknown"
  defp status_label(value), do: to_string(value)

  defp append_sample(samples_map, key, value) do
    Map.update(samples_map, key, [value], fn existing ->
      [value | existing] |> Enum.take(@histogram_max_samples)
    end)
  end

  defp histogram_summaries_seconds(samples_map) do
    Enum.reduce(samples_map, %{}, fn {key, values_ms}, acc ->
      Map.put(acc, key, summarize_seconds(values_ms))
    end)
  end

  defp summarize_seconds([]) do
    %{count: 0, sum: 0.0, buckets: Enum.map(@histogram_buckets, &{&1, 0}) ++ [{:infinity, 0}]}
  end

  defp summarize_seconds(values_ms) do
    values_seconds = Enum.map(values_ms, &(&1 / 1000))
    count = length(values_seconds)
    sum_seconds = Enum.sum(values_seconds)

    buckets =
      Enum.map(@histogram_buckets, fn le ->
        {le, Enum.count(values_seconds, &(&1 <= le))}
      end) ++ [{:infinity, count}]

    %{
      count: count,
      sum: Float.round(sum_seconds, 6),
      buckets: buckets
    }
  end

  # ---------------------------------------------------------------------------
  # http_request_size_bytes / http_response_size_bytes
  #
  # True cumulative Prometheus histograms: buckets/sum/count are updated
  # incrementally on every observation and never shrink or evict, unlike the
  # raw-sample-list reservoir used by summarize_seconds/1 above.
  # ---------------------------------------------------------------------------

  defp request_size_from_conn(conn) do
    case parse_content_length(conn.req_headers) do
      {:ok, size} -> size
      :missing when conn.method in ["GET", "HEAD"] -> 0
      :missing -> nil
      :invalid -> nil
    end
  end

  defp response_size_from_conn(conn) do
    case safe_iodata_length(conn.resp_body) do
      {:ok, size} ->
        size

      :unavailable ->
        case parse_content_length(conn.resp_headers) do
          {:ok, size} -> size
          _ -> nil
        end
    end
  end

  defp safe_iodata_length(nil), do: :unavailable

  defp safe_iodata_length(body) do
    {:ok, IO.iodata_length(body)}
  rescue
    _ -> :unavailable
  end

  defp parse_content_length(headers) do
    case List.keyfind(headers, "content-length", 0) do
      nil ->
        :missing

      {_key, value} ->
        case Integer.parse(value) do
          {n, ""} when n >= 0 -> {:ok, n}
          _ -> :invalid
        end
    end
  end

  defp observe_byte_histogram(totals, _key, nil), do: totals

  defp observe_byte_histogram(totals, key, value) when is_integer(value) and value >= 0 do
    Map.update(totals, key, fresh_byte_summary(value), &apply_byte_observation(&1, value))
  end

  defp observe_byte_histogram(totals, _key, _invalid), do: totals

  defp fresh_byte_summary(value) do
    empty = %{
      count: 0,
      sum: 0,
      buckets: Enum.map(@size_buckets_bytes, &{&1, 0}) ++ [{:infinity, 0}]
    }

    apply_byte_observation(empty, value)
  end

  defp apply_byte_observation(summary, value) do
    new_buckets =
      Enum.map(summary.buckets, fn
        {:infinity, count} -> {:infinity, count + 1}
        {le, count} when value <= le -> {le, count + 1}
        {le, count} -> {le, count}
      end)

    %{summary | buckets: new_buckets, sum: summary.sum + value, count: summary.count + 1}
  end

  defp app_version do
    case Application.spec(:da_product_app, :vsn) do
      nil -> "unknown"
      vsn -> to_string(vsn)
    end
  end

  # ---------------------------------------------------------------------------
  # Timers
  # ---------------------------------------------------------------------------

  defp schedule_beam_poll, do: Process.send_after(self(), :poll_beam_vm, @beam_poll_interval)
end