super_cache

defmodule SuperCache.Cluster.Stats do
  @moduledoc """
  Observability utilities for a running SuperCache cluster.

  ## Available reports

  | Function                  | What it shows                                        |
  |---------------------------|------------------------------------------------------|
  | `cluster/0`               | Full cluster overview — nodes, partitions, ETS sizes |
  | `partitions/0`            | Per-partition primary/replica assignment + ETS count |
  | `primary_partitions/0`    | Partitions owned (primary) by this node              |
  | `replica_partitions/0`    | Partitions this node replicates but does not own     |
  | `node_partitions/1`       | Partition ownership for any specific node            |
  | `three_phase_commit/0`    | Counters for the 3PC coordinator (this node)         |
  | `api/0`                   | Per-operation call counters + latency percentiles    |
  | `print/1`                 | Pretty-print any stats map to the console            |

  ## Quick start

      alias SuperCache.Cluster.Stats

      Stats.cluster()        |> Stats.print()
      Stats.partitions()     |> Stats.print()
      Stats.three_phase_commit() |> Stats.print()
      Stats.api()            |> Stats.print()

  ## Telemetry integration

  All counters are stored in a dedicated ETS table owned by
  `SuperCache.Cluster.Metrics` and are updated via
  `SuperCache.Cluster.Stats.record/2`.  You can attach your own
  `:telemetry` handler or periodically call `Stats.api/0` to ship
  metrics to an external system.
  """

  alias SuperCache.{Storage, Partition, Config}
  alias SuperCache.Cluster.{Manager, TxnRegistry, Metrics}

  # ── Cluster overview ─────────────────────────────────────────────────────────

  @doc """
  Return a full cluster overview map.

  Fields:

  - `:nodes` — list of live nodes
  - `:node_count` — number of live nodes
  - `:replication_factor` — configured replication factor
  - `:replication_mode` — `:async | :sync | :strong`
  - `:num_partitions` — total partition count
  - `:partitions` — list of per-partition maps (see `partitions/0`)
  - `:total_records` — sum of ETS record counts across all local partitions

  ## Example

      iex> SuperCache.Cluster.Stats.cluster()
      %{
        nodes: [:"primary@127.0.0.1", :"replica@127.0.0.1"],
        node_count: 2,
        replication_factor: 2,
        replication_mode: :async,
        num_partitions: 8,
        total_records: 1_042,
        partitions: [...]
      }
  """
  @spec cluster() :: map
  def cluster() do
    parts = partitions()
    total = parts |> Enum.map(& &1.local_record_count) |> Enum.sum()

    %{
      nodes:              Manager.live_nodes(),
      node_count:         Manager.live_nodes() |> length(),
      replication_factor: Config.get_config(:replication_factor, 1),
      replication_mode:   Config.get_config(:replication_mode, :async),
      num_partitions:     Partition.get_num_partition(),
      total_records:      total,
      partitions:         parts
    }
  end

  # ── Partition maps ───────────────────────────────────────────────────────────

  @doc """
  Return one map per partition describing ownership and local ETS size.

  Each map contains:

  - `:idx` — partition index
  - `:table` — ETS table atom
  - `:primary` — node that owns this partition
  - `:replicas` — list of replica nodes
  - `:role` — `:primary | :replica | :none` (this node's role)
  - `:local_record_count` — number of records in the local ETS table

  ## Example

      iex> SuperCache.Cluster.Stats.partitions()
      [
        %{idx: 0, table: :"SuperCache.Storage.Ets_0", primary: :"a@host",
          replicas: [:"b@host"], role: :primary, local_record_count: 128},
        ...
      ]
  """
  @spec partitions() :: [map]
  def partitions() do
    me = node()

    Enum.map(0..(Partition.get_num_partition() - 1), fn idx ->
      {primary, replicas} = Manager.get_replicas(idx)
      table     = Partition.get_partition_by_idx(idx)
      {_, count} = Storage.stats(table)

      role =
        cond do
          primary == me    -> :primary
          me in replicas   -> :replica
          true             -> :none
        end

      %{
        idx:                 idx,
        table:               table,
        primary:             primary,
        replicas:            replicas,
        role:                role,
        local_record_count:  count
      }
    end)
  end

  @doc """
  Return only the partitions for which this node is the **primary**.

  ## Example

      iex> SuperCache.Cluster.Stats.primary_partitions()
      [
        %{idx: 0, primary: :"a@host", replicas: [:"b@host"],
          role: :primary, local_record_count: 128},
        ...
      ]
  """
  @spec primary_partitions() :: [map]
  def primary_partitions() do
    partitions() |> Enum.filter(&(&1.role == :primary))
  end

  @doc """
  Return only the partitions for which this node is a **replica**.

  ## Example

      iex> SuperCache.Cluster.Stats.replica_partitions()
      [
        %{idx: 3, primary: :"b@host", replicas: [:"a@host"],
          role: :replica, local_record_count: 95},
        ...
      ]
  """
  @spec replica_partitions() :: [map]
  def replica_partitions() do
    partitions() |> Enum.filter(&(&1.role == :replica))
  end

  @doc """
  Return the partition ownership summary for `target_node`.

  Returns a map with:

  - `:node` — the queried node
  - `:primary_count` — number of partitions it owns as primary
  - `:replica_count` — number of partitions it replicates
  - `:primary_partitions` — list of partition indices where it is primary
  - `:replica_partitions` — list of partition indices where it is replica

  ## Example

      iex> SuperCache.Cluster.Stats.node_partitions(:"replica@127.0.0.1")
      %{
        node: :"replica@127.0.0.1",
        primary_count: 4,
        replica_count: 4,
        primary_partitions: [1, 3, 5, 7],
        replica_partitions: [0, 2, 4, 6]
      }
  """
  @spec node_partitions(node) :: map
  def node_partitions(target_node) do
    all = partitions()

    primary_idxs =
      all
      |> Enum.filter(&(&1.primary == target_node))
      |> Enum.map(& &1.idx)

    replica_idxs =
      all
      |> Enum.filter(&(target_node in &1.replicas))
      |> Enum.map(& &1.idx)

    %{
      node:                target_node,
      primary_count:       length(primary_idxs),
      replica_count:       length(replica_idxs),
      primary_partitions:  primary_idxs,
      replica_partitions:  replica_idxs
    }
  end

  # ── Three-phase commit metrics ───────────────────────────────────────────────

  @doc """
  Return three-phase commit counters for this node.

  Counters:

  - `:in_flight` — number of transactions currently in the TxnRegistry
  - `:committed` — total successfully committed transactions
  - `:aborted` — total aborted transactions (any phase)
  - `:prepare_failures` — transactions that failed at PREPARE phase
  - `:pre_commit_failures` — transactions that failed at PRE_COMMIT phase
  - `:commit_failures` — transactions that failed at COMMIT phase
  - `:recovered_committed` — in-doubt :pre_committed entries resolved on startup
  - `:recovered_aborted` — in-doubt :prepared entries rolled back on startup
  - `:avg_commit_latency_us` — rolling average commit latency in microseconds
  - `:p99_commit_latency_us` — 99th-percentile commit latency (sampled window)

  ## Example

      iex> SuperCache.Cluster.Stats.three_phase_commit()
      %{
        in_flight: 0,
        committed: 1_200,
        aborted: 3,
        prepare_failures: 1,
        pre_commit_failures: 1,
        commit_failures: 1,
        recovered_committed: 0,
        recovered_aborted: 0,
        avg_commit_latency_us: 1_840,
        p99_commit_latency_us: 4_210
      }
  """
  @spec three_phase_commit() :: map
  def three_phase_commit() do
    c = Metrics.get_all(:tpc)

    {avg, p99} = latency_stats(:tpc_commit_latency_us)

    %{
      in_flight:              TxnRegistry.count(),
      committed:              Map.get(c, :committed, 0),
      aborted:                Map.get(c, :aborted, 0),
      prepare_failures:       Map.get(c, :prepare_failures, 0),
      pre_commit_failures:    Map.get(c, :pre_commit_failures, 0),
      commit_failures:        Map.get(c, :commit_failures, 0),
      recovered_committed:    Map.get(c, :recovered_committed, 0),
      recovered_aborted:      Map.get(c, :recovered_aborted, 0),
      avg_commit_latency_us:  avg,
      p99_commit_latency_us:  p99
    }
  end

  # ── API call metrics ─────────────────────────────────────────────────────────

  @doc """
  Return per-operation call counters and latency statistics for this node.

  Each entry in the returned map is keyed by operation name and contains:

  - `:calls` — total number of calls
  - `:errors` — calls that returned `{:error, _}`
  - `:avg_us` — rolling average latency in microseconds
  - `:p99_us` — 99th-percentile latency (sampled window)

  Operations tracked: `:put`, `:get_local`, `:get_primary`, `:get_quorum`,
  `:delete`, `:delete_all`, `:delete_match`, `:replicate_async`,
  `:replicate_sync`, `:replicate_strong`.

  ## Example

      iex> SuperCache.Cluster.Stats.api()
      %{
        put:              %{calls: 5_000, errors: 2, avg_us: 210, p99_us: 890},
        get_local:        %{calls: 18_000, errors: 0, avg_us: 12,  p99_us: 45},
        get_primary:      %{calls: 300,   errors: 0, avg_us: 540, p99_us: 1_200},
        get_quorum:       %{calls: 50,    errors: 0, avg_us: 620, p99_us: 1_400},
        delete:           %{calls: 800,   errors: 0, avg_us: 190, p99_us: 750},
        delete_all:       %{calls: 3,     errors: 0, avg_us: 4_200, p99_us: 6_000},
        delete_match:     %{calls: 120,   errors: 0, avg_us: 310, p99_us: 900},
        replicate_async:  %{calls: 5_000, errors: 5, avg_us: 95,  p99_us: 400},
        replicate_sync:   %{calls: 0,     errors: 0, avg_us: 0,   p99_us: 0},
        replicate_strong: %{calls: 0,     errors: 0, avg_us: 0,   p99_us: 0}
      }
  """
  @spec api() :: map
  def api() do
    ops = [
      :put, :get_local, :get_primary, :get_quorum,
      :delete, :delete_all, :delete_match,
      :replicate_async, :replicate_sync, :replicate_strong
    ]

    Map.new(ops, fn op ->
      c = Metrics.get_all({:api, op})
      {avg, p99} = latency_stats({:api_latency_us, op})

      {op, %{
        calls:   Map.get(c, :calls, 0),
        errors:  Map.get(c, :errors, 0),
        avg_us:  avg,
        p99_us:  p99
      }}
    end)
  end

  # ── Pretty printer ───────────────────────────────────────────────────────────

  @doc """
  Pretty-print any stats map or list returned by this module to stdout.

  Handles the nested structures returned by `cluster/0`, `partitions/0`,
  `three_phase_commit/0`, and `api/0`.

  ## Example

      SuperCache.Cluster.Stats.cluster() |> SuperCache.Cluster.Stats.print()

      # ╔══════════════════════════════════════╗
      # ║       SuperCache Cluster Stats       ║
      # ╚══════════════════════════════════════╝
      # nodes            : [a@host, b@host]
      # node_count       : 2
      # replication_mode : async
      # ...
  """
  @spec print(map | [map]) :: :ok
  def print(stats) when is_map(stats) do
    IO.puts(banner("SuperCache Stats"))
    print_map(stats, 0)
    IO.puts("")
    :ok
  end

  def print(stats) when is_list(stats) do
    IO.puts(banner("SuperCache Stats"))
    stats
    |> Enum.with_index(1)
    |> Enum.each(fn {item, i} ->
      IO.puts("  [#{i}]")
      print_map(item, 4)
    end)
    IO.puts("")
    :ok
  end

  # ── Instrumentation helpers (called by Router, Replicator, 3PC) ──────────────

  @doc """
  Record a completed API call.

  Called by `SuperCache.Cluster.Router` and `SuperCache.Cluster.Replicator`
  after each operation.

      Stats.record({:api, :put}, %{latency_us: 210, error: false})
  """
  @spec record(term, map) :: :ok
  def record(key, %{latency_us: lat, error: err}) do
    Metrics.increment({:api, key}, :calls)
    if err, do: Metrics.increment({:api, key}, :errors)
    Metrics.push_latency({:api_latency_us, key}, lat)
    :ok
  end

  @doc """
  Record a completed 3PC transaction.

      Stats.record_tpc(:committed, latency_us: 1_840)
      Stats.record_tpc(:aborted, phase: :prepare)
  """
  @spec record_tpc(atom, keyword) :: :ok
  def record_tpc(:committed, opts) do
    Metrics.increment(:tpc, :committed)
    if lat = Keyword.get(opts, :latency_us) do
      Metrics.push_latency(:tpc_commit_latency_us, lat)
    end
    :ok
  end

  def record_tpc(:aborted, opts) do
    Metrics.increment(:tpc, :aborted)
    case Keyword.get(opts, :phase) do
      :prepare    -> Metrics.increment(:tpc, :prepare_failures)
      :pre_commit -> Metrics.increment(:tpc, :pre_commit_failures)
      :commit     -> Metrics.increment(:tpc, :commit_failures)
      _           -> :ok
    end
    :ok
  end

  def record_tpc(:recovered_committed, _opts), do: Metrics.increment(:tpc, :recovered_committed)
  def record_tpc(:recovered_aborted,   _opts), do: Metrics.increment(:tpc, :recovered_aborted)
  def record_tpc(_event, _opts),               do: :ok

  # ── Private ──────────────────────────────────────────────────────────────────

  # Compute {avg, p99} from the latency sample ring buffer.
  defp latency_stats(key) do
    case Metrics.get_latency_samples(key) do
      [] ->
        {0, 0}

      samples ->
        sorted = Enum.sort(samples)
        n      = length(sorted)
        avg    = div(Enum.sum(sorted), n)
        p99    = Enum.at(sorted, trunc(n * 0.99))
        {avg, p99}
    end
  end

  defp banner(title) do
    pad = 38
    top = "╔" <> String.duplicate("═", pad) <> "╗"
    centered = title
               |> String.pad_leading(div(pad + String.length(title), 2))
               |> String.pad_trailing(pad)
    mid = "║" <> centered <> "║"
    bot = "╚" <> String.duplicate("═", pad) <> "╝"
    Enum.join([top, mid, bot], "\n")
  end

  defp print_map(map, indent) do
    pad = String.duplicate(" ", indent)

    Enum.each(map, fn {k, v} ->
      key_str = k |> to_string() |> String.pad_trailing(24)

      case v do
        v when is_map(v) ->
          IO.puts("#{pad}#{key_str}:")
          print_map(v, indent + 4)

        v when is_list(v) and length(v) > 0 and is_map(hd(v)) ->
          IO.puts("#{pad}#{key_str}:")
          v |> Enum.with_index(1) |> Enum.each(fn {item, i} ->
            IO.puts("#{pad}  [#{i}]")
            print_map(item, indent + 4)
          end)

        _ ->
          IO.puts("#{pad}#{key_str}: #{inspect(v)}")
      end
    end)
  end
end