super_cache

defmodule SuperCache.Distributed.Struct do
  @moduledoc """
  Cluster-aware struct store.

  Writes (`add`, `remove`, `remove_all`) are routed to the primary node for
  the partition that owns the struct type. Reads (`get`, `get_all`) are
  served from the local node (eventual consistency).

  ## Partition strategy
  Every operation derives its partition by hashing `{:struct_storage, StructName}`.
  This is the namespace key — consistent across puts, deletes, and reads.

  API is identical to `SuperCache.Struct`.

  ## Example

      alias SuperCache.Distributed.Struct, as: DStruct

      SuperCache.Cluster.Bootstrap.start!(...)

      defmodule Person, do: defstruct [:id, :name, :age]

      DStruct.init(%Person{}, :id)
      DStruct.add(%Person{id: 1, name: "Alice", age: 30})
      DStruct.get(%Person{id: 1})      # => {:ok, %Person{...}}
      DStruct.remove(%Person{id: 1})
      DStruct.remove_all(%Person{})
  """

  alias SuperCache.{Storage, Partition}
  alias SuperCache.Cluster.{Manager, Replicator}
  require Logger

  ## Public API ────────────────────────────────────────────────────────────────

  @doc "Register the key field for a struct type. Routed to primary."
  @spec init(map, atom) :: true | {:error, any}
  def init(%{__struct__: _} = struct, key \\ :id) when is_atom(key) do
    with true <- Map.has_key?(struct, key),
         {:error, :key_not_found} <- get_key_field(struct) do
      route_write(struct, :local_init, [struct, key])
    else
      false    -> {:error, "key does not exist on struct"}
      {:ok, _} -> {:error, "struct already initialised"}
    end
  end

  @doc "Store a struct. Routed to primary."
  @spec add(map) :: {:ok, map} | {:error, any}
  def add(%{__struct__: _} = struct) do
    with {:ok, _key} <- get_key_field(struct) do
      route_write(struct, :local_add, [struct])
    end
  end

  @doc "Retrieve a struct by key. Read from local node."
  @spec get(map) :: {:ok, map} | {:error, :not_found | any}
  def get(%{__struct__: struct_name} = struct) do
    with {:ok, key} <- get_key_field(struct) do
      ns        = namespace(struct)
      partition = Partition.get_partition(ns)
      ets_key   = {{:struct_storage, :struct, struct_name}, Map.get(struct, key)}

      case Storage.get(ets_key, partition) do
        []            -> {:error, :not_found}
        [{_, result}] -> {:ok, result}
      end
    end
  end

  @doc "Return all structs of this type from the local node."
  @spec get_all(map) :: {:ok, list} | {:error, any}
  def get_all(%{__struct__: struct_name} = struct) do
    with {:ok, _key} <- get_key_field(struct) do
      ns        = namespace(struct)
      partition = Partition.get_partition(ns)

      results =
        Storage.get_by_match_object(
          {{{:struct_storage, :struct, struct_name}, :_}, :_},
          partition
        )
        |> Enum.map(fn {_, value} -> value end)

      {:ok, results}
    end
  end

  @doc "Remove a struct by key. Routed to primary."
  @spec remove(map) :: {:ok, map} | {:error, any}
  def remove(%{__struct__: _} = struct) do
    with {:ok, _key} <- get_key_field(struct) do
      # Read locally first to return the old value.
      case get(struct) do
        {:error, :not_found} = err -> err
        {:ok, existing}            -> route_write(struct, :local_remove, [struct])
                                      {:ok, existing}
      end
    end
  end

  @doc "Remove all structs of this type. Routed to primary."
  @spec remove_all(map) :: {:ok, :removed} | {:error, any}
  def remove_all(%{__struct__: struct_name} = struct) do
    with {:ok, _key} <- get_key_field(struct) do
      Logger.debug(fn -> "super_cache, dist.struct, remove_all #{inspect(struct_name)}" end)
      route_write(struct, :local_remove_all, [struct])
    end
  end

  ## Remote entry points (called via :erpc — do NOT call directly) ─────────────

  @doc false
  def local_init(%{__struct__: struct_name} = struct, key) do
    ns        = namespace(struct)
    partition = Partition.get_partition(ns)
    idx       = Partition.get_partition_order(ns)
    record    = {{:struct_storage, :key, struct_name}, key}

    Storage.put(record, partition)
    Replicator.replicate(idx, :put, record)
    true
  end

  @doc false
  def local_add(%{__struct__: struct_name} = struct) do
    with {:ok, key} <- get_key_field(struct) do
      ns        = namespace(struct)
      partition = Partition.get_partition(ns)
      idx       = Partition.get_partition_order(ns)
      key_data  = Map.get(struct, key)
      ets_key   = {{:struct_storage, :struct, struct_name}, key_data}

      Logger.debug(fn ->
        "super_cache, dist.struct, add #{inspect(struct_name)} key=#{inspect(key_data)}"
      end)

      # Overwrite — delete old then insert new.
      Storage.delete(ets_key, partition)
      Storage.put({ets_key, struct}, partition)
      Replicator.replicate(idx, :delete, ets_key)
      Replicator.replicate(idx, :put, {ets_key, struct})
      {:ok, struct}
    end
  end

  @doc false
  def local_remove(%{__struct__: struct_name} = struct) do
    with {:ok, key} <- get_key_field(struct) do
      ns        = namespace(struct)
      partition = Partition.get_partition(ns)
      idx       = Partition.get_partition_order(ns)
      ets_key   = {{:struct_storage, :struct, struct_name}, Map.get(struct, key)}

      Storage.delete(ets_key, partition)
      Replicator.replicate(idx, :delete, ets_key)
      :ok
    end
  end

  @doc false
  def local_remove_all(%{__struct__: struct_name} = struct) do
    ns        = namespace(struct)
    partition = Partition.get_partition(ns)
    idx       = Partition.get_partition_order(ns)
    pattern   = {{{:struct_storage, :struct, struct_name}, :_}, :_}

    Storage.delete_match(pattern, partition)
    Replicator.replicate(idx, :delete_match, pattern)
    {:ok, :removed}
  end

  ## Private ───────────────────────────────────────────────────────────────────

  # Consistent namespace key used for ALL partition hashing in this module.
  # Every read, write, and delete hashes this same value.
  defp namespace(%{__struct__: struct_name}), do: {:struct_storage, struct_name}

  defp get_key_field(%{__struct__: struct_name} = struct) do
    ns        = namespace(struct)
    partition = Partition.get_partition(ns)

    case Storage.get({:struct_storage, :key, struct_name}, partition) do
      []         -> {:error, :key_not_found}
      [{_, key}] -> {:ok, key}
    end
  end

  defp primary_for(struct) do
    idx = Partition.get_partition_order(namespace(struct))
    {primary, _} = Manager.get_replicas(idx)
    primary
  end

  defp route_write(struct, fun, args) do
    primary = primary_for(struct)

    if primary == node() do
      apply(__MODULE__, fun, args)
    else
      Logger.debug(fn ->
        "super_cache, dist.struct, fwd #{fun} → #{inspect(primary)}"
      end)
      :erpc.call(primary, __MODULE__, fun, args, 5_000)
    end
  end
end