defmodule CMDCGateway.SessionStore do @moduledoc """ ETS-based Session 注册表。 维护 `session_id → Agent pid + metadata` 映射关系。 通过 monitor 机制在 Agent 进程退出时自动清理对应条目。 ## metadata 字段 - `:pid` — Agent 进程 pid - `:api_key` — 创建此 session 的 API Key - `:tenant_id` — 租户标识(Phase 10 多租户隔离预留,当前默认 `"default"`) - `:monitor_ref` — 进程监控引用 - `:created_at` — 创建时间戳(毫秒) - `:blueprint` — 使用的 Blueprint 模块名(可选) - `:model` — LLM 模型标识 """ use GenServer @table __MODULE__ @monitor_table :"#{__MODULE__}.Monitors" @type session_metadata :: %{ pid: pid(), api_key: String.t(), tenant_id: String.t(), monitor_ref: reference(), created_at: integer(), blueprint: module() | nil, model: String.t() | nil } # ========================================================================== # Public API # ========================================================================== @doc "启动 SessionStore GenServer。" @spec start_link(keyword()) :: GenServer.on_start() def start_link(opts \\ []) do GenServer.start_link(__MODULE__, opts, name: __MODULE__) end @doc """ 注册一个新 session。 ## 参数 - `session_id` — 唯一会话标识 - `pid` — Agent 进程 pid - `attrs` — 额外属性 keyword,支持 `:api_key`、`:tenant_id`、`:blueprint`、`:model` ## 返回 - `{:ok, session_id}` — 注册成功 - `{:error, :already_exists}` — session_id 已存在 """ @spec create(String.t(), pid(), keyword()) :: {:ok, String.t()} | {:error, :already_exists} def create(session_id, pid, attrs \\ []) do GenServer.call(__MODULE__, {:create, session_id, pid, attrs}) end @doc """ 获取 session 元数据。 返回 `{:ok, metadata}` 或 `{:error, :not_found}`。 """ @spec get(String.t()) :: {:ok, session_metadata()} | {:error, :not_found} def get(session_id) do case :ets.lookup(@table, session_id) do [{^session_id, metadata}] -> {:ok, metadata} [] -> {:error, :not_found} end end @doc """ 删除一个 session,取消进程监控。 返回 `:ok`(无论 session 是否存在)。 """ @spec delete(String.t()) :: :ok def delete(session_id) do GenServer.call(__MODULE__, {:delete, session_id}) end @doc "列出所有活跃 session(返回 `[{session_id, metadata}]`)。" @spec list() :: [{String.t(), session_metadata()}] def list do :ets.tab2list(@table) end @doc """ 按 tenant_id 过滤 session。 Phase 10 多租户隔离核心查询路径。 """ @spec list_by_tenant(String.t()) :: [{String.t(), session_metadata()}] def list_by_tenant(target_tenant_id) do @table |> :ets.tab2list() |> Enum.filter(fn {_id, %{tenant_id: tid}} -> tid == target_tenant_id end) end @doc "按 api_key 过滤 session。" @spec list_by_api_key(String.t()) :: [{String.t(), session_metadata()}] def list_by_api_key(target_api_key) do @table |> :ets.tab2list() |> Enum.filter(fn {_id, %{api_key: ak}} -> ak == target_api_key end) end @doc "返回当前活跃 session 数量。" @spec count() :: non_neg_integer() def count do :ets.info(@table, :size) end # ========================================================================== # GenServer Callbacks # ========================================================================== @impl true def init(_opts) do table = :ets.new(@table, [:named_table, :set, :public, read_concurrency: true]) _mon_table = :ets.new(@monitor_table, [:named_table, :set, :protected]) {:ok, %{table: table}} end @impl true def handle_call({:create, session_id, pid, attrs}, _from, state) do case :ets.lookup(@table, session_id) do [{^session_id, _}] -> {:reply, {:error, :already_exists}, state} [] -> ref = Process.monitor(pid) :ets.insert(@monitor_table, {ref, session_id}) metadata = %{ pid: pid, api_key: Keyword.get(attrs, :api_key, "default"), tenant_id: Keyword.get(attrs, :tenant_id, "default"), monitor_ref: ref, created_at: System.system_time(:millisecond), blueprint: Keyword.get(attrs, :blueprint), model: Keyword.get(attrs, :model) } :ets.insert(@table, {session_id, metadata}) {:reply, {:ok, session_id}, state} end end def handle_call({:delete, session_id}, _from, state) do case :ets.lookup(@table, session_id) do [{^session_id, %{monitor_ref: ref}}] -> Process.demonitor(ref, [:flush]) :ets.delete(@monitor_table, ref) :ets.delete(@table, session_id) [] -> :ok end {:reply, :ok, state} end @impl true def handle_info({:DOWN, ref, :process, _pid, _reason}, state) do case :ets.lookup(@monitor_table, ref) do [{^ref, session_id}] -> :ets.delete(@table, session_id) :ets.delete(@monitor_table, ref) [] -> :ok end {:noreply, state} end def handle_info(_msg, state), do: {:noreply, state} end