brook

defmodule Brook.Storage.Redis do
  @moduledoc """
  Implements the `Brook.Storage` behaviour for the Redis
  key/value storage system, saving the application view state
  as binary encodings of the direct Elixir terms to be saved with
  maximum compression.
  """
  use GenServer
  require Logger
  import Brook.Config, only: [registry: 1, put: 3, get: 2]
  @behaviour Brook.Storage

  @type config :: [
          redix_args: keyword(),
          namespace: String.t()
        ]

  @impl Brook.Storage
  def persist(instance, event, collection, key, value) do
    %{redix: redix, namespace: namespace, event_limits: event_limits} = state(instance)
    Logger.debug(fn -> "#{__MODULE__}: persisting #{collection}:#{key}:#{inspect(value)} to redis" end)

    with {:ok, serialized_event} <- Brook.Serializer.serialize(event),
         gzipped_serialized_event <- :zlib.gzip(serialized_event),
         event_limit <- Map.get(event_limits, event.type, :no_limit),
         {:ok, serialized_value} <- Brook.Serializer.serialize(value),
         {:ok, "OK"} <-
           redis_set(
             redix,
             key(namespace, collection, key),
             Jason.encode!(%{"key" => key, "value" => serialized_value})
           ),
         {:ok, _count} <-
           redis_append(
             redix,
             events_key(namespace, collection, key, event.type),
             gzipped_serialized_event,
             event_limit
           ) do
      :ok
    end
  rescue
    ArgumentError -> {:error, not_initialized_exception()}
  end

  @impl Brook.Storage
  def delete(instance, collection, key) do
    %{redix: redix, namespace: namespace} = state(instance)

    with {:ok, event_keys} <- redis_keys(redix, events_key(namespace, collection, key, "*")),
         {:ok, _count} <- redis_delete(redix, [key(namespace, collection, key) | event_keys]) do
      :ok
    end
  end

  @impl Brook.Storage
  def get(instance, collection, key) do
    %{redix: redix, namespace: namespace} = state(instance)

    case redis_get(redix, key(namespace, collection, key)) do
      {:ok, nil} ->
        {:ok, nil}

      {:ok, value} ->
        value
        |> Jason.decode!()
        |> Map.get("value")
        |> Brook.Deserializer.deserialize()

      error_result ->
        error_result
    end
  end

  @impl Brook.Storage
  def get_all(instance, collection) do
    %{redix: redix, namespace: namespace} = state(instance)

    with {:ok, keys} <- redis_keys(redix, key(namespace, collection, "*")),
         {:ok, encoded_values} <- redis_multiget(redix, keys),
         {:ok, decoded_values} <- safe_map(encoded_values, &Jason.decode/1) do
      decoded_values
      |> Enum.map(&deserialize_data/1)
      |> Enum.into(%{})
      |> ok()
    end
  end

  @impl Brook.Storage
  def get_events(instance, collection, key) do
    %{redix: redix, namespace: namespace} = state(instance)

    with {:ok, event_keys} <- redis_keys(redix, events_key(namespace, collection, key, "*")),
         {:ok, nested_events} <- safe_map(event_keys, &redis_get_all(redix, &1)),
         compressed_events <- List.flatten(nested_events),
         serialized_events <- Enum.map(compressed_events, &:zlib.gunzip/1),
         {:ok, events} <- safe_map(serialized_events, &Brook.Deserializer.deserialize/1) do
      events |> sort_events() |> ok()
    end
  end

  @impl Brook.Storage
  def get_events(instance, collection, key, type) do
    %{redix: redix, namespace: namespace} = state(instance)

    with {:ok, compressed_events} <- redis_get_all(redix, events_key(namespace, collection, key, type)),
         serialized_events <- Enum.map(compressed_events, &:zlib.gunzip/1),
         {:ok, events} <- safe_map(serialized_events, &Brook.Deserializer.deserialize/1) do
      events |> sort_events() |> ok()
    end
  end

  @impl Brook.Storage
  def start_link(args) do
    instance = Keyword.fetch!(args, :instance)
    GenServer.start_link(__MODULE__, args, name: via(registry(instance)))
  end

  @impl GenServer
  def init(args) do
    instance = Keyword.fetch!(args, :instance)
    redix_args = Keyword.fetch!(args, :redix_args)
    namespace = Keyword.fetch!(args, :namespace)
    event_limits = Keyword.get(args, :event_limits, %{})

    {:ok, redix} = Redix.start_link(redix_args)

    put(instance, __MODULE__, %{namespace: namespace, redix: redix, event_limits: event_limits})

    Brook.Storage.Redis.Migration.migrate(instance, redix, namespace, event_limits)

    {:ok, %{namespace: namespace, redix: redix, event_limits: event_limits}}
  end

  defp state(instance) do
    case get(instance, __MODULE__) do
      {:ok, value} -> value
      :error -> raise not_initialized_exception()
    end
  end

  defp safe_map(list, function) do
    Enum.reduce_while(list, {:ok, []}, fn value, {:ok, list} ->
      case function.(value) do
        {:ok, result} -> {:cont, {:ok, [result | list]}}
        {:error, reason} -> {:halt, {:error, reason}}
      end
    end)
  end

  defp not_initialized_exception() do
    Brook.Uninitialized.exception(message: "#{__MODULE__} is not initialized yet!")
  end

  defp redis_get(redix, key), do: Redix.command(redix, ["GET", key])
  defp redis_get_all(redix, key), do: Redix.command(redix, ["LRANGE", key, 0, -1])
  defp redis_set(redix, key, value), do: Redix.command(redix, ["SET", key, value])

  defp redis_append(redix, key, value, limit) do
    with {:ok, count} <- Redix.command(redix, ["RPUSH", key, value]),
         :ok <- redis_trim(redix, key, limit) do
      {:ok, count}
    end
  end

  defp redis_trim(_redix, _key, :no_limit), do: :ok

  defp redis_trim(redix, key, limit) do
    with {:ok, "OK"} <- Redix.command(redix, ["LTRIM", key, -limit, -1]) do
      :ok
    end
  end

  defp redis_keys(redix, key), do: Redix.command(redix, ["KEYS", key])
  defp redis_delete(redix, keys), do: Redix.command(redix, ["DEL" | keys])

  defp redis_multiget(_redix, []), do: {:ok, []}
  defp redis_multiget(redix, keys), do: Redix.command(redix, ["MGET" | keys])

  defp ok(value), do: {:ok, value}
  defp key(namespace, collection, key), do: "#{namespace}:state:#{collection}:#{key}"
  defp events_key(namespace, collection, key, event_type), do: "#{namespace}:events:#{collection}:#{key}:#{event_type}"
  defp via(registry), do: {:via, Registry, {registry, __MODULE__}}

  defp deserialize_data(%{"key" => key, "value" => value}) do
    {:ok, deserialized_value} = Brook.Deserializer.deserialize(value)
    {key, deserialized_value}
  end

  defp sort_events(events) do
    Enum.sort_by(events, fn event -> event.create_ts end)
  end
end