cmdc_gateway

defmodule CMDCGateway.SSEHandler do
  @moduledoc """
  SSE 事件流处理器。

  订阅 CMDC EventBus → 翻译为对外 JSON → 按 SSE 格式逐行 chunk 推送。
  连接断开时自动 unsubscribe。心跳 keepalive 间隔 30 秒。
  """

  import Plug.Conn

  alias CMDCGateway.EventTranslator

  @heartbeat_interval 30_000

  @doc """
  接管 Plug.Conn，开启 SSE chunked 流。

  此函数会阻塞当前进程直到客户端断开或 session 结束。
  """
  @spec stream(Plug.Conn.t(), String.t()) :: Plug.Conn.t()
  def stream(conn, session_id) do
    conn =
      conn
      |> put_resp_content_type("text/event-stream")
      |> put_resp_header("cache-control", "no-cache")
      |> put_resp_header("x-accel-buffering", "no")
      |> send_chunked(200)

    CMDC.EventBus.subscribe(session_id)
    schedule_heartbeat()

    try do
      loop(conn, session_id)
    after
      CMDC.EventBus.unsubscribe(session_id)
    end
  end

  # ==========================================================================
  # Private — Event Loop
  # ==========================================================================

  defp loop(conn, session_id) do
    receive do
      {:cmdc_event, ^session_id, event} ->
        case EventTranslator.translate(event) do
          {:ok, event_type, data} ->
            case chunk(conn, EventTranslator.to_sse(event_type, data)) do
              {:ok, conn} ->
                if terminal_event?(event) do
                  conn
                else
                  loop(conn, session_id)
                end

              {:error, _reason} ->
                conn
            end

          :skip ->
            loop(conn, session_id)
        end

      :heartbeat ->
        case chunk(conn, EventTranslator.heartbeat()) do
          {:ok, conn} ->
            schedule_heartbeat()
            loop(conn, session_id)

          {:error, _reason} ->
            conn
        end
    after
      60_000 ->
        conn
    end
  end

  defp terminal_event?({:agent_end, _messages, _usage}), do: true
  defp terminal_event?({:agent_abort, _reason}), do: true
  defp terminal_event?(:agent_abort), do: true
  defp terminal_event?(_), do: false

  defp schedule_heartbeat do
    Process.send_after(self(), :heartbeat, @heartbeat_interval)
  end
end