cmdc_gateway

defmodule CMDCGateway.TaskStore do
  @moduledoc """
  A2A Task 状态短期缓存。

  A2A 协议「兜底查询」机制 — 当 webhook 派发失败
  / 客户端漏接收时，可通过 `GET /v1/a2a/tasks/:id` 主动拉取最近一次
  task 状态。

  ## 设计

  - ETS 表 `:cmdc_gateway_task_store` 按 `task_id` 索引，存放最近 10 分钟
    内的 task 状态快照
  - 每次 webhook 派发时（无论成功失败）由 `A2A.start_with_webhook/3`
    内部 `dispatch_webhook/5` 同步 update 一次状态
  - GenServer 后台定时 GC，默认每 60s 扫一遍清理过期条目
  - 不持久化（重启即丢，仅做"短期兜底"，与 12M.1 dead-letter 长期
    持久化职责正交）

  ## 公共 API

      TaskStore.put(task_id, status_atom, payload, opts \\ [])
      TaskStore.get(task_id) :: {:ok, map} | {:error, :not_found}
      TaskStore.delete(task_id)
      TaskStore.list_recent(limit \\ 50)

  ## ETS 行格式

      {task_id, %{
        task_id: String.t(),
        status: :accepted | :working | :completed | :failed,
        last_event: :accepted | :status_update | :artifact_update
                  | :completed | :failed,
        payload: map(),
        updated_at_ms: integer(),
        ttl_until_ms: integer()
      }}
  """

  use GenServer
  require Logger

  @table :cmdc_gateway_task_store
  @default_ttl_ms 600_000
  @default_gc_interval_ms 60_000

  @type task_id :: String.t()
  @type status :: :accepted | :working | :completed | :failed
  @type entry :: %{
          task_id: task_id(),
          status: status(),
          last_event: atom(),
          payload: map(),
          updated_at_ms: integer(),
          ttl_until_ms: integer()
        }

  # ==========================================================================
  # Public API
  # ==========================================================================

  @doc """
  写入 / 更新 task 状态。

  ## 选项
  - `:status` — 显式覆盖；默认从 `event` 推导（accepted/status_update→working/
    artifact_update→working/completed/failed）
  - `:ttl_ms` — 自定义 TTL；默认 600_000 (10 分钟)
  """
  @spec put(task_id(), atom(), map(), keyword()) :: :ok
  def put(task_id, event, payload, opts \\ [])
      when is_binary(task_id) and is_atom(event) and is_map(payload) do
    now = System.system_time(:millisecond)
    ttl = Keyword.get(opts, :ttl_ms, @default_ttl_ms)
    status = Keyword.get(opts, :status, status_for_event(event))

    entry = %{
      task_id: task_id,
      status: status,
      last_event: event,
      payload: payload,
      updated_at_ms: now,
      ttl_until_ms: now + ttl
    }

    case :ets.whereis(@table) do
      :undefined ->
        :ok

      _ ->
        :ets.insert(@table, {task_id, entry})
        :ok
    end
  end

  @doc """
  查询 task 状态。

  返回 `{:ok, entry}`，或在 task 未知 / 已过期时返回 `{:error, :not_found}`。
  """
  @spec get(task_id()) :: {:ok, entry()} | {:error, :not_found}
  def get(task_id) when is_binary(task_id) do
    case :ets.whereis(@table) do
      :undefined -> {:error, :not_found}
      _ -> do_get(task_id)
    end
  end

  defp do_get(task_id) do
    case :ets.lookup(@table, task_id) do
      [{^task_id, entry}] -> check_ttl(entry)
      [] -> {:error, :not_found}
    end
  end

  defp check_ttl(%{ttl_until_ms: ttl} = entry) do
    if ttl >= System.system_time(:millisecond) do
      {:ok, entry}
    else
      {:error, :not_found}
    end
  end

  @doc "删除指定 task_id 的条目，主要用于测试 / 手动清理。"
  @spec delete(task_id()) :: :ok
  def delete(task_id) when is_binary(task_id) do
    case :ets.whereis(@table) do
      :undefined -> :ok
      _ -> :ets.delete(@table, task_id) && :ok
    end
  end

  @doc "列出最近 N 个 task（按 `updated_at_ms` 倒序），用于运维查看。"
  @spec list_recent(pos_integer()) :: [entry()]
  def list_recent(limit \\ 50) when is_integer(limit) and limit > 0 do
    case :ets.whereis(@table) do
      :undefined ->
        []

      _ ->
        @table
        |> :ets.tab2list()
        |> Enum.map(fn {_id, entry} -> entry end)
        |> Enum.sort_by(& &1.updated_at_ms, :desc)
        |> Enum.take(limit)
    end
  end

  @doc "强制运行一次 GC（清理所有过期条目），主要用于测试。"
  @spec gc_now() :: non_neg_integer()
  def gc_now do
    GenServer.call(__MODULE__, :gc_now)
  end

  # ==========================================================================
  # GenServer
  # ==========================================================================

  @doc """
  启动 TaskStore GenServer + ETS 表。

  ## 选项
  - `:gc_interval_ms` — GC 触发间隔；默认 60_000 (1 分钟)
  - `:default_ttl_ms` — 单条 TTL；默认 600_000 (10 分钟)
  - `:name` — 进程名；默认 `__MODULE__`
  """
  def start_link(opts \\ []) do
    GenServer.start_link(__MODULE__, opts, name: Keyword.get(opts, :name, __MODULE__))
  end

  @impl true
  def init(opts) do
    table = ensure_table()
    gc_interval = Keyword.get(opts, :gc_interval_ms, @default_gc_interval_ms)
    schedule_gc(gc_interval)

    {:ok,
     %{
       table: table,
       gc_interval_ms: gc_interval,
       default_ttl_ms: Keyword.get(opts, :default_ttl_ms, @default_ttl_ms)
     }}
  end

  @impl true
  def handle_call(:gc_now, _from, state) do
    removed = run_gc()
    {:reply, removed, state}
  end

  @impl true
  def handle_info(:gc_tick, state) do
    removed = run_gc()

    if removed > 0 do
      Logger.debug("[TaskStore] gc removed #{removed} expired entries")
    end

    schedule_gc(state.gc_interval_ms)
    {:noreply, state}
  end

  def handle_info(_msg, state), do: {:noreply, state}

  # ==========================================================================
  # Private
  # ==========================================================================

  defp ensure_table do
    case :ets.whereis(@table) do
      :undefined ->
        :ets.new(@table, [
          :set,
          :public,
          :named_table,
          read_concurrency: true,
          write_concurrency: true
        ])

      ref ->
        ref
    end
  end

  defp schedule_gc(interval_ms) do
    Process.send_after(self(), :gc_tick, interval_ms)
  end

  defp run_gc do
    case :ets.whereis(@table) do
      :undefined ->
        0

      _ ->
        now = System.system_time(:millisecond)

        # match_spec：删除所有 ttl_until_ms < now 的条目
        # ETS match spec for nested map field is tricky；用 select_delete + guard
        spec = [
          {{:_, %{ttl_until_ms: :"$1"}}, [{:<, :"$1", now}], [true]}
        ]

        :ets.select_delete(@table, spec)
    end
  end

  defp status_for_event(:accepted), do: :accepted
  defp status_for_event(:status_update), do: :working
  defp status_for_event(:artifact_update), do: :working
  defp status_for_event(:completed), do: :completed
  defp status_for_event(:failed), do: :failed
  defp status_for_event(_), do: :working
end