defmodule Gralkor.Python do
  @moduledoc """
  PythonX runtime owner for the embedded Gralkor stack.

  Responsibilities, all in `init/1`:

    1. **Reap redislite orphans.** `falkordblite` (loaded into PythonX in this
       BEAM) spawns a `redis-server` grandchild. A hard BEAM SIGKILL leaves
       it orphaned. SIGKILL anything matching `redislite/bin/redis-server`
       before we boot — safe because this runs *before* our own Python init,
       so anything matching is by definition not ours-yet, and the path is
       unique to falkordblite.

    2. **Materialise the venv + initialise the interpreter** via
       `Pythonx.uv_init/2` from the jido_gralkor-owned `@pyproject_toml`. This
       is what makes the embedded Python stack self-contained: the consumer
       (e.g. susu) configures nothing about Python — the graphiti-core pin is
       jido_gralkor's private detail. Guarded against re-init (the Pythonx NIF
       throws "already been initialized" on a second call) so multiple boots
       in one VM — as functional-test modules do — are safe.

    3. **Smoke-import `graphiti_core`** through PythonX so any venv / import
       failure surfaces at boot rather than on the first real call.

  See `ex-python-runtime` in `gralkor/TEST_TREES.md`.
  """

  use GenServer

  require Logger

  @pyproject_path Path.join([__DIR__, "..", "..", "priv", "python", "pyproject.toml"])
  @external_resource @pyproject_path
  @pyproject_toml File.read!(@pyproject_path)

  def start_link(opts \\ []) do
    name = Keyword.get(opts, :name, __MODULE__)
    GenServer.start_link(__MODULE__, opts, if(name, do: [name: name], else: []))
  end

  @impl true
  def init(opts) do
    list_orphans = Keyword.get(opts, :list_orphans, &list_redislite_orphans/0)
    kill_pid = Keyword.get(opts, :kill_pid, &sigkill/1)
    uv_init = Keyword.get(opts, :uv_init, &ensure_initialised/0)
    smoke_import = Keyword.get(opts, :smoke_import, &smoke_import_graphiti/0)
    install_loop? = Keyword.get(opts, :install_loop, true)
    reap_orphans? = Keyword.get(opts, :reap_orphans, true)

    with :ok <- maybe_reap(reap_orphans?, list_orphans, kill_pid),
         :ok <- uv_init.(),
         :ok <- smoke_import.(),
         :ok <- maybe_install_loop(install_loop?) do
      {:ok, %{}}
    else
      {:error, reason} -> {:stop, {:boot_failed, reason}}
    end
  end

  defp maybe_reap(false, _list, _kill), do: :ok
  defp maybe_reap(true, list, kill), do: reap_redislite_orphans(list, kill)

  @doc """
  Materialise the uv-managed venv and initialise the PythonX interpreter from
  the jido_gralkor-owned `@pyproject_toml`. Idempotent within a VM: a flag in
  `:persistent_term` short-circuits the second call so we never hit the NIF's
  "already been initialized" guard when more than one `Gralkor.Python` boots.
  """
  @spec ensure_initialised() :: :ok | {:error, term()}
  def ensure_initialised do
    if :persistent_term.get({__MODULE__, :uv_inited}, false) do
      :ok
    else
      Pythonx.uv_init(@pyproject_toml)
      :persistent_term.put({__MODULE__, :uv_inited}, true)
      :ok
    end
  rescue
    e -> {:error, {:uv_init, Exception.message(e)}}
  end

  defp maybe_install_loop(false), do: :ok
  defp maybe_install_loop(true), do: install_async_runtime()

  @doc """
  Spin up a daemon-thread asyncio event loop and stash it on `asyncio` as
  `_gralkor_loop` plus a `_gralkor_run(coro)` helper that submits onto it.

  Must run once per Pythonx interpreter, before any code that calls into
  graphiti via `asyncio._gralkor_run`. Idempotent — the second call is a
  no-op.

  Why: Pythonx.eval creates a fresh event loop per `asyncio.run` call.
  AsyncFalkorDB (and any redis-async connection) binds its connections to
  the loop they were created on; reusing them on a different loop raises
  "Future attached to a different loop". The spike measured the alternative
  pattern (Step 6 in `pythonx-spike/spike.exs`) at ~56µs per call vs ~112µs
  for `asyncio.run` — and, crucially, it shares one loop across all calls
  so connection reuse works.
  """
  @spec install_async_runtime() :: :ok | {:error, term()}
  def install_async_runtime do
    Pythonx.eval(
      """
      import asyncio, threading
      if not hasattr(asyncio, '_gralkor_loop'):
          loop = asyncio.new_event_loop()
          started = threading.Event()
          def _run():
              asyncio.set_event_loop(loop)
              started.set()
              loop.run_forever()
          threading.Thread(target=_run, daemon=True).start()
          started.wait()
          asyncio._gralkor_loop = loop
          asyncio._gralkor_run = lambda coro: asyncio.run_coroutine_threadsafe(coro, loop).result()
      """,
      %{}
    )

    :ok
  rescue
    e in Pythonx.Error -> {:error, {:install_async_runtime, Exception.message(e)}}
  end

  @doc """
  SIGKILL every pid the listing function returns. Pure plumbing — accepts
  injected list/kill functions so the unit test doesn't have to spawn real
  redis processes.
  """
  @spec reap_redislite_orphans((-> [integer()]), (integer() -> any())) :: :ok | {:error, term()}
  def reap_redislite_orphans(list_orphans, kill_pid)
      when is_function(list_orphans, 0) and is_function(kill_pid, 1) do
    Enum.each(list_orphans.(), fn pid ->
      Logger.warning("[gralkor] reaping redislite orphan pid #{pid}")
      kill_pid.(pid)
    end)

    :ok
  end

  @doc """
  Try to `import graphiti_core` via Pythonx; surface any failure as
  `{:error, _}`.
  """
  @spec smoke_import_graphiti() :: :ok | {:error, term()}
  def smoke_import_graphiti do
    Pythonx.eval("import graphiti_core", %{})
    :ok
  rescue
    e -> {:error, {:graphiti_import, Exception.message(e)}}
  end

  # ── default OS plumbing ────────────────────────────────────

  defp list_redislite_orphans do
    case System.cmd("pgrep", ["-af", "redislite/bin/redis-server"], stderr_to_stdout: true) do
      {output, 0} ->
        output
        |> String.split("\n", trim: true)
        |> Enum.map(fn line ->
          [pid_str | _] = String.split(line, " ", parts: 2)
          String.to_integer(pid_str)
        end)

      {_, _} ->
        # pgrep exits non-zero when nothing matches — that's fine.
        []
    end
  end

  defp sigkill(pid) do
    System.cmd("kill", ["-KILL", to_string(pid)], stderr_to_stdout: true)
    :ok
  end
end