defmodule Vela.Backend.Redis do
  @moduledoc """
  Redis-backed cache backend. Requires the `:redix` dependency.

  Entries are serialized using a configurable serializer (defaults to
  `Vela.Serializer.Native`). Keys are prefixed with the cache name
  to avoid collisions when multiple caches share a Redis instance.

  ## Backend Options

      backend_opts: [
        url: "redis://localhost:6379",
        key_prefix: "myapp:",           # optional, defaults to "vela:{cache_name}:"
        serializer: Vela.Serializer.Native  # optional
      ]
  """

  @behaviour Vela.Backend

  alias Vela.Cache.Entry

  defstruct [:conn, :key_prefix, :serializer]

  @impl true
  def init(config) do
    opts = config.backend_opts
    url = Keyword.get(opts, :url, "redis://localhost:6379")

    case Redix.start_link(url) do
      {:ok, conn} ->
        state = %__MODULE__{
          conn: conn,
          key_prefix: Keyword.get(opts, :key_prefix, "vela:#{config.name}:"),
          serializer: Keyword.get(opts, :serializer, Vela.Serializer.Native)
        }

        {:ok, state}

      {:error, reason} ->
        {:error, {:redis_connect_failed, reason}}
    end
  end

  @impl true
  def get(%__MODULE__{conn: conn, key_prefix: prefix, serializer: ser}, key) do
    redis_key = prefix <> encode_key(key)

    case Redix.command(conn, ["GET", redis_key]) do
      {:ok, nil} ->
        {:error, :not_found}

      {:ok, bytes} ->
        {:ok, entry} = ser.decode(bytes, [])
        {:ok, entry}

      {:error, reason} ->
        {:error, reason}
    end
  end

  @impl true
  def put(%__MODULE__{conn: conn, key_prefix: prefix, serializer: ser} = state, %Entry{} = entry) do
    redis_key = prefix <> encode_key(entry.key)
    {:ok, bytes} = ser.encode(entry, [])

    cmd =
      case entry.expires_at do
        :infinity ->
          ["SET", redis_key, bytes]

        expires_at ->
          ttl_ms = max(1, expires_at - System.monotonic_time(:millisecond))
          ["SET", redis_key, bytes, "PX", to_string(ttl_ms)]
      end

    case Redix.command(conn, cmd) do
      {:ok, "OK"} -> {:ok, state}
      {:error, reason} -> {:error, reason}
    end
  end

  @impl true
  def delete(%__MODULE__{conn: conn, key_prefix: prefix} = state, key) do
    Redix.command(conn, ["DEL", prefix <> encode_key(key)])
    {:ok, state}
  end

  @impl true
  def get_many(state, keys) do
    result =
      Enum.reduce(keys, %{}, fn key, acc ->
        case get(state, key) do
          {:ok, entry} -> Map.put(acc, key, entry)
          _ -> acc
        end
      end)

    {:ok, result}
  end

  @impl true
  def put_many(state, entries) do
    Enum.reduce_while(entries, {:ok, state}, fn entry, {:ok, acc} ->
      case put(acc, entry) do
        {:ok, new_state} -> {:cont, {:ok, new_state}}
        error -> {:halt, error}
      end
    end)
  end

  @impl true
  def flush(%__MODULE__{conn: conn, key_prefix: prefix} = state) do
    case Redix.command(conn, ["KEYS", prefix <> "*"]) do
      {:ok, []} ->
        {:ok, state}

      {:ok, keys} ->
        Redix.command(conn, ["DEL" | keys])
        {:ok, state}

      {:error, _} ->
        {:ok, state}
    end
  end

  @impl true
  def flush_expired(state, _now) do
    # Redis handles TTL expiry natively
    {:ok, 0, state}
  end

  @impl true
  def size(%__MODULE__{conn: conn, key_prefix: prefix}) do
    case Redix.command(conn, ["KEYS", prefix <> "*"]) do
      {:ok, keys} -> length(keys)
      {:error, _} -> 0
    end
  end

  @impl true
  def delete_by_tag(%__MODULE__{} = state, tag) do
    # Redis doesn't have native tag support — scan all keys and filter.
    # For production use with large datasets, consider maintaining a
    # tag index in a Redis SET.
    case Redix.command(state.conn, ["KEYS", state.key_prefix <> "*"]) do
      {:ok, []} ->
        {:ok, 0, state}

      {:ok, redis_keys} ->
        deleted = Enum.count(redis_keys, &delete_if_tagged(state, &1, tag))
        {:ok, deleted, state}

      {:error, _} ->
        {:ok, 0, state}
    end
  end

  defp delete_if_tagged(state, redis_key, tag) do
    with {:ok, bytes} when not is_nil(bytes) <- Redix.command(state.conn, ["GET", redis_key]),
         {:ok, entry} <- state.serializer.decode(bytes, []),
         true <- tag in entry.tags do
      Redix.command(state.conn, ["DEL", redis_key])
      true
    else
      _ -> false
    end
  end

  defp encode_key(key), do: :erlang.term_to_binary(key) |> Base.encode16()
end