cmdc

defmodule CMDC.EventBus do
  @moduledoc """
  基于 Registry 的会话事件发布/订阅系统，附带可选的 ring buffer 用于断线重连补帧。

  ## 基础订阅

      CMDC.EventBus.subscribe("session-abc123")

      receive do
        {:cmdc_event, "session-abc123", event} -> handle_event(event)
      end

      CMDC.EventBus.broadcast("session-abc123", {:stream_chunk, "session-abc123", "hello"})

      CMDC.EventBus.subscribe_all()

  ## Ring Buffer

  适用场景：WebSocket / Channel 短暂断线重连时不丢事件。

  Agent 启动时通过 `CMDC.Options.new!(event_buffer_size: 100)` 开启该会话的内存
  ring buffer。Buffer 满后自动丢弃最早事件（FIFO）。默认 `0`（关闭，零内存开销）。

  重连后用 `:since` 选项 replay：

      {:ok, _pid} = CMDC.EventBus.subscribe("session-abc123", since: last_index)

      receive do
        {:cmdc_event, _sid, event} -> # ... 包含 replay + 新事件
      end

  典型 buffer size 建议 `50 ~ 200`，对应 1-3 秒内的 stream_chunk 风暴。
  Buffer 由 ETS 实现，session 终止时通过 `disable_buffer/1` 清理（CMDC.Agent
  在 `terminate/3` 自动调用）。
  """

  @registry CMDC.EventBus.Registry
  @buffer_table :cmdc_event_buffer
  @meta_table :cmdc_event_buffer_meta

  # ==========================================================================
  # Public API — Subscribe / Broadcast
  # ==========================================================================

  @doc """
  订阅当前进程对指定会话 ID 的事件。

  接收 `{:cmdc_event, session_id, event}` 消息。

  ## 选项

  - `:since :: non_neg_integer()` — 从该 index 之后的事件开始 replay。要求该
    会话已开启 ring buffer（`Options.event_buffer_size > 0`），否则该选项被忽略。
    如 `since` 早于 buffer 起点（已被丢弃），仅 replay 还在 buffer 内的部分。
  - `:types :: [atom()]` — 只 replay 这些 type 的事件;
    适用于 UI 重连补帧只想要 `:stream_chunk` + `:agent_end` 不想要 `:tool_started`
    等内部事件的场景。事件 type 来自事件第一个元素（如 `{:agent_end, _, _}` 的 type
    是 `:agent_end`，裸 atom 事件如 `:agent_start` type 就是它自己）。
    **此选项仅影响 replay 阶段**；实时订阅不过滤，避免漏事件。

  返回 `{:ok, pid}`，replay 事件在订阅注册后按时序异步投递到当前进程。
  """
  @spec subscribe(String.t(), keyword()) :: {:ok, pid()} | {:error, term()}
  def subscribe(session_id, opts \\ []) when is_binary(session_id) and is_list(opts) do
    case Registry.register(@registry, session_id, []) do
      {:ok, _} = ok ->
        case Keyword.get(opts, :since) do
          nil ->
            :noop

          since when is_integer(since) and since >= 0 ->
            type_filter = build_type_filter(Keyword.get(opts, :types))
            replay_since(session_id, since, type_filter)
        end

        ok

      {:error, _} = error ->
        error
    end
  end

  defp build_type_filter(nil), do: nil
  defp build_type_filter([]), do: nil

  defp build_type_filter(types) when is_list(types) do
    set = MapSet.new(types)
    fn event -> MapSet.member?(set, event_type(event)) end
  end

  defp event_type(event) when is_atom(event), do: event
  defp event_type(event) when is_tuple(event) and tuple_size(event) >= 1, do: elem(event, 0)
  defp event_type(_), do: nil

  @doc "订阅当前进程对**所有**会话的事件（监控/调试用）。"
  @spec subscribe_all() :: {:ok, pid()} | {:error, {:already_registered, pid()}}
  def subscribe_all do
    Registry.register(@registry, :all, [])
  end

  @doc """
  广播事件到指定会话所有订阅者，并写入 ring buffer（若已开启）。

  返回事件分配到的单调递增 `index`（开启 buffer 时）或 `:ok`（未开启）。
  Registry 未启动时（如部分测试环境）静默返回 `:ok`。
  """
  @spec broadcast(String.t(), term()) :: :ok | {:ok, pos_integer()}
  def broadcast(session_id, event) when is_binary(session_id) do
    do_broadcast(session_id, event)
  rescue
    ArgumentError -> :ok
  end

  @doc "取消当前进程对指定会话 ID 的订阅。"
  @spec unsubscribe(String.t()) :: :ok
  def unsubscribe(session_id) when is_binary(session_id) do
    Registry.unregister(@registry, session_id)
  end

  @doc "取消当前进程的通配符订阅。"
  @spec unsubscribe_all() :: :ok
  def unsubscribe_all do
    Registry.unregister(@registry, :all)
  end

  # ==========================================================================
  # Public API — Ring Buffer 生命周期
  # ==========================================================================

  @doc false
  @spec init_buffer_tables() :: :ok
  def init_buffer_tables do
    create_table(@buffer_table, [:ordered_set, :public, :named_table, read_concurrency: true])
    create_table(@meta_table, [:set, :public, :named_table, write_concurrency: true])
    :ok
  end

  @doc """
  开启指定会话的 ring buffer（CMDC.Agent.init 自动调用）。

  `buffer_size` 必须为正整数；`<= 0` 等同 no-op。
  """
  @spec enable_buffer(String.t(), non_neg_integer()) :: :ok
  def enable_buffer(_session_id, size) when size <= 0, do: :ok

  def enable_buffer(session_id, buffer_size)
      when is_binary(session_id) and is_integer(buffer_size) do
    ensure_tables()
    :ets.insert(@meta_table, {session_id, buffer_size, 0})
    :ok
  end

  @doc """
  关闭 ring buffer 并清理该会话所有缓存事件（CMDC.Agent.terminate 自动调用）。
  """
  @spec disable_buffer(String.t()) :: :ok
  def disable_buffer(session_id) when is_binary(session_id) do
    if buffer_tables_alive?() do
      :ets.delete(@meta_table, session_id)

      ms = [{{{session_id, :"$1"}, :_}, [], [true]}]
      :ets.select_delete(@buffer_table, ms)
    end

    :ok
  end

  @doc """
  返回该会话最新事件 index（未开启 buffer 时为 `nil`）。

  常用于断线重连前记录"上次看到的 index"，重连时传 `subscribe(sid, since: idx)`。
  """
  @spec last_index(String.t()) :: non_neg_integer() | nil
  def last_index(session_id) when is_binary(session_id) do
    if buffer_tables_alive?() do
      case :ets.lookup(@meta_table, session_id) do
        [{_sid, _size, last_idx}] -> last_idx
        [] -> nil
      end
    end
  end

  @doc false
  @spec buffer_size(String.t()) :: non_neg_integer()
  def buffer_size(session_id) when is_binary(session_id) do
    if buffer_tables_alive?() do
      ms = [{{{session_id, :_}, :_}, [], [true]}]
      :ets.select_count(@buffer_table, ms)
    else
      0
    end
  end

  # ==========================================================================
  # 私有辅助
  # ==========================================================================

  defp do_broadcast(session_id, event) do
    msg = {:cmdc_event, session_id, event}

    Registry.dispatch(@registry, session_id, fn entries ->
      for {pid, _value} <- entries, do: send(pid, msg)
    end)

    Registry.dispatch(@registry, :all, fn entries ->
      for {pid, _value} <- entries, do: send(pid, msg)
    end)

    case maybe_buffer_event(session_id, event) do
      nil -> :ok
      idx -> {:ok, idx}
    end
  end

  defp maybe_buffer_event(session_id, event) do
    if buffer_tables_alive?() do
      case :ets.lookup(@meta_table, session_id) do
        [{_sid, buffer_size, _last_idx}] ->
          new_idx = :ets.update_counter(@meta_table, session_id, {3, 1})
          :ets.insert(@buffer_table, {{session_id, new_idx}, event})
          trim_buffer(session_id, new_idx, buffer_size)
          new_idx

        [] ->
          nil
      end
    end
  end

  defp trim_buffer(_sid, new_idx, buffer_size) when new_idx <= buffer_size, do: :ok

  defp trim_buffer(session_id, new_idx, buffer_size) do
    cutoff = new_idx - buffer_size

    ms = [
      {
        {{session_id, :"$1"}, :_},
        [{:"=<", :"$1", cutoff}],
        [true]
      }
    ]

    :ets.select_delete(@buffer_table, ms)
    :ok
  end

  defp replay_since(session_id, since_idx, type_filter) do
    if buffer_tables_alive?() do
      do_replay(session_id, since_idx, type_filter)
    end

    :ok
  end

  defp do_replay(session_id, since_idx, type_filter) do
    ms = [
      {
        {{session_id, :"$1"}, :"$2"},
        [{:>, :"$1", since_idx}],
        [{{:"$1", :"$2"}}]
      }
    ]

    events = :ets.select(@buffer_table, ms)
    pid = self()

    events
    |> Enum.sort_by(fn {idx, _ev} -> idx end)
    |> Enum.filter(fn {_idx, ev} -> filter_match?(ev, type_filter) end)
    |> Enum.each(fn {_idx, event} -> send(pid, {:cmdc_event, session_id, event}) end)
  end

  defp filter_match?(_event, nil), do: true
  defp filter_match?(event, type_filter) when is_function(type_filter, 1), do: type_filter.(event)

  defp create_table(name, opts) do
    if :ets.info(name) == :undefined do
      :ets.new(name, opts)
    end
  end

  defp ensure_tables do
    if not buffer_tables_alive?() do
      init_buffer_tables()
    end
  end

  defp buffer_tables_alive? do
    :ets.info(@buffer_table) != :undefined and :ets.info(@meta_table) != :undefined
  end
end