sql

# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: 2025 DBVisor

defmodule SQL.Pool do
  @moduledoc false
  use GenServer

  def start_link(opts) do
    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
  end

  @impl true
  def init(state) do
    {:ok, initialize(state)}
  end

  def checkout(%{id: _id}=sql, pool \\ :default) do
    # start_time = System.monotonic_time()
    scheduler_id = :erlang.system_info(:scheduler_id)
    state = :persistent_term.get({__MODULE__, pool})
    {_size, connections, activations, recent_activations, sockets, schedulers, _indexes, _health} = state
    {n, workers} = Map.get(schedulers, scheduler_id)
    case checkout(state, n, workers) do
      :none=error ->
        # :telemetry.execute([:sql, :checkout], %{pool: pool, duration: System.monotonic_time()-start_time}, %{id: id})
        error
      {idx, _load} ->
        case :atomics.compare_exchange(connections, idx, 0, 1) do
          :ok ->
            :counters.add(activations, idx, 1)
            :counters.add(recent_activations, idx, 1)
            result = {idx, elem(sockets, idx-1)}
            # :telemetry.execute([:sql, :checkout], %{pool: pool, duration: System.monotonic_time()-start_time}, %{id: id})
            result
          _ ->
            checkout(sql, pool)
        end
    end
  end

  defp checkout(_state, _n, []), do: :none
  defp checkout({size, _connections, activations, recent_activations, _sockets, _schedulers, indexes, health}=state, n, workers) do
    # Power-of-three-choices with load weighting
    workers
    |> Enum.take_random(min(3, n))
    |> Enum.map(fn id ->
      active = :counters.get(activations, id)
      recent = :counters.get(recent_activations, id)
      health = :atomics.get(health, id)
      {id, active + recent / 1000, health} # Weight recent activations
    end)
    |> Enum.filter(fn {_, _, health} -> health == 1 end)
    |> Enum.min_by(fn {_, load, _} -> load end, fn -> nil end)
    |> case do
      {id, load, _} -> {id, trunc(load)}
      _ -> checkout(state, size-n, indexes--workers)
    end
  end

  def checkin(idx, pool) do
    {_size, connections, activations, recent_activations, _sockets, _schedulers, _indexes, _health} = :persistent_term.get({__MODULE__, pool})
    :ok = :atomics.put(connections, idx, 0)
    :ok = :counters.sub(activations, idx, 1)
    :ok = :counters.sub(recent_activations, idx, 1)
  end

  defp initialize(state) do
    pool = build_pool(state)
    :persistent_term.put({__MODULE__, state.name}, pool)
    Map.put(state, :pool, pool)
  end

  defp build_pool(state) do
    schedulers = :erlang.system_info(:schedulers_online)
    size = state.size
    connections = :atomics.new(size, signed: false)
    health = :atomics.new(size, signed: false)
    indexes = Enum.to_list(1..size)
    schedulers = indexes
    |> Enum.reduce(%{}, fn id, acc ->
      Map.update(acc, rem(id, schedulers)+1, [id], &([id | &1]))
    end)
    |> Map.new(fn {k, v} -> {k, {length(v), v}} end)
    {size, connections, :counters.new(size, [:write_concurrency]), :counters.new(size, [:write_concurrency]), init_sockets(size, connections, health, state.protocol), schedulers, indexes, health}
  end

  defp init_sockets(size, connections, health, _protocol) do
    # queries = :persistent_term.get({__MODULE__, :queries})
    for idx <- 1..size do
      :atomics.put(connections, idx, 0)
      :atomics.put(health, idx, 1)
      # :socket.open(:inet, :stream, protocol)
      self()
    end
    |> List.to_tuple()
  end
end