in_memory.ex

defmodule Commanded.EventStore.Adapters.InMemory do
  @moduledoc """
  An in-memory event store adapter implemented as a `GenServer` process which
  stores events in memory only.

  This is only designed for testing purposes.
  """

  @behaviour Commanded.EventStore.Adapter

  use GenServer

  defmodule State do
    @moduledoc false

    defstruct [
      :name,
      :serializer,
      persisted_events: [],
      streams: %{},
      transient_subscribers: %{},
      persistent_subscriptions: %{},
      snapshots: %{},
      next_event_number: 1
    ]
  end

  alias Commanded.EventStore.Adapters.InMemory.{PersistentSubscription, State, Subscription}
  alias Commanded.EventStore.{EventData, RecordedEvent, SnapshotData}
  alias Commanded.UUID

  def start_link(opts \\ []) do
    {start_opts, in_memory_opts} =
      Keyword.split(opts, [:debug, :name, :timeout, :spawn_opt, :hibernate_after])

    state = %State{
      name: Keyword.fetch!(opts, :name),
      serializer: Keyword.get(in_memory_opts, :serializer)
    }

    GenServer.start_link(__MODULE__, state, start_opts)
  end

  @impl GenServer
  def init(%State{} = state) do
    {:ok, state}
  end

  @impl Commanded.EventStore.Adapter
  def child_spec(application, config) do
    {event_store_name, config} = parse_config(application, config)

    supervisor_name = subscriptions_supervisor_name(event_store_name)

    child_spec = [
      {DynamicSupervisor, strategy: :one_for_one, name: supervisor_name},
      %{
        id: event_store_name,
        start: {__MODULE__, :start_link, [config]}
      }
    ]

    {:ok, child_spec, %{name: event_store_name}}
  end

  @impl Commanded.EventStore.Adapter
  def append_to_stream(adapter_meta, stream_uuid, expected_version, events, _opts \\ []) do
    event_store = event_store_name(adapter_meta)

    GenServer.call(event_store, {:append, stream_uuid, expected_version, events})
  end

  @impl Commanded.EventStore.Adapter
  def stream_forward(adapter_meta, stream_uuid, start_version \\ 0, read_batch_size \\ 1_000)

  def stream_forward(adapter_meta, stream_uuid, start_version, _read_batch_size) do
    event_store = event_store_name(adapter_meta)

    GenServer.call(event_store, {:stream_forward, stream_uuid, start_version})
  end

  @impl Commanded.EventStore.Adapter
  def subscribe(adapter_meta, stream_uuid) do
    event_store = event_store_name(adapter_meta)

    GenServer.call(event_store, {:subscribe, stream_uuid, self()})
  end

  @impl Commanded.EventStore.Adapter
  def subscribe_to(adapter_meta, stream_uuid, subscription_name, subscriber, start_from, opts) do
    event_store = event_store_name(adapter_meta)

    subscription = %PersistentSubscription{
      concurrency_limit: Keyword.get(opts, :concurrency_limit),
      name: subscription_name,
      partition_by: Keyword.get(opts, :partition_by),
      start_from: start_from,
      stream_uuid: stream_uuid
    }

    GenServer.call(event_store, {:subscribe_to, subscription, subscriber})
  end

  @impl Commanded.EventStore.Adapter
  def ack_event(adapter_meta, subscription, %RecordedEvent{} = event) when is_pid(subscription) do
    event_store = event_store_name(adapter_meta)

    GenServer.call(event_store, {:ack_event, event, subscription})
  end

  @impl Commanded.EventStore.Adapter
  def unsubscribe(adapter_meta, subscription) when is_pid(subscription) do
    event_store = event_store_name(adapter_meta)

    GenServer.call(event_store, {:unsubscribe, subscription})
  end

  @impl Commanded.EventStore.Adapter
  def delete_subscription(adapter_meta, stream_uuid, subscription_name) do
    event_store = event_store_name(adapter_meta)

    GenServer.call(event_store, {:delete_subscription, stream_uuid, subscription_name})
  end

  @impl Commanded.EventStore.Adapter
  def read_snapshot(adapter_meta, source_uuid) do
    event_store = event_store_name(adapter_meta)

    GenServer.call(event_store, {:read_snapshot, source_uuid})
  end

  @impl Commanded.EventStore.Adapter
  def record_snapshot(adapter_meta, snapshot) do
    event_store = event_store_name(adapter_meta)

    GenServer.call(event_store, {:record_snapshot, snapshot})
  end

  @impl Commanded.EventStore.Adapter
  def delete_snapshot(adapter_meta, source_uuid) do
    event_store = event_store_name(adapter_meta)

    GenServer.call(event_store, {:delete_snapshot, source_uuid})
  end

  def reset!(application, config \\ []) do
    {event_store, _config} = parse_config(application, config)

    GenServer.call(event_store, :reset!)
  end

  @impl GenServer
  def handle_call({:append, stream_uuid, expected_version, events}, _from, %State{} = state) do
    %State{streams: streams} = state

    stream_events = Map.get(streams, stream_uuid)

    {reply, state} =
      case {expected_version, stream_events} do
        {:any_version, nil} ->
          persist_events(state, stream_uuid, [], events)

        {:any_version, stream_events} ->
          persist_events(state, stream_uuid, stream_events, events)

        {:no_stream, stream_events} when is_list(stream_events) ->
          {{:error, :stream_exists}, state}

        {:no_stream, nil} ->
          persist_events(state, stream_uuid, [], events)

        {:stream_exists, nil} ->
          {{:error, :stream_not_found}, state}

        {:stream_exists, stream_events} ->
          persist_events(state, stream_uuid, stream_events, events)

        {0, nil} ->
          persist_events(state, stream_uuid, [], events)

        {expected_version, nil} when is_integer(expected_version) ->
          {{:error, :wrong_expected_version}, state}

        {expected_version, stream_events}
        when is_integer(expected_version) and length(stream_events) != expected_version ->
          {{:error, :wrong_expected_version}, state}

        {expected_version, stream_events}
        when is_integer(expected_version) and length(stream_events) == expected_version ->
          persist_events(state, stream_uuid, stream_events, events)
      end

    {:reply, reply, state}
  end

  @impl GenServer
  def handle_call({:stream_forward, stream_uuid, start_version}, _from, %State{} = state) do
    %State{streams: streams} = state

    reply =
      case Map.get(streams, stream_uuid) do
        nil ->
          {:error, :stream_not_found}

        events ->
          events
          |> Enum.reverse()
          |> Stream.drop(max(0, start_version - 1))
          |> Stream.map(&deserialize(state, &1))
          |> Enum.map(&set_event_number_from_version(&1, stream_uuid))
      end

    {:reply, reply, state}
  end

  @impl GenServer
  def handle_call({:subscribe, stream_uuid, subscriber}, _from, %State{} = state) do
    %State{transient_subscribers: transient_subscribers} = state

    Process.monitor(subscriber)

    transient_subscribers =
      Map.update(transient_subscribers, stream_uuid, [subscriber], fn subscribers ->
        [subscriber | subscribers]
      end)

    state = %State{state | transient_subscribers: transient_subscribers}

    {:reply, :ok, state}
  end

  @impl GenServer
  def handle_call({:subscribe_to, subscription, subscriber}, _from, %State{} = state) do
    %PersistentSubscription{name: subscription_name} = subscription
    %State{persistent_subscriptions: persistent_subscriptions} = state

    {reply, state} =
      case Map.get(persistent_subscriptions, subscription_name) do
        nil ->
          start_persistent_subscription(state, subscription, subscriber)

        %PersistentSubscription{subscribers: []} = subscription ->
          start_persistent_subscription(state, subscription, subscriber)

        %PersistentSubscription{concurrency_limit: nil} ->
          {{:error, :subscription_already_exists}, state}

        %PersistentSubscription{} = subscription ->
          %PersistentSubscription{concurrency_limit: concurrency_limit, subscribers: subscribers} =
            subscription

          if length(subscribers) < concurrency_limit do
            start_persistent_subscription(state, subscription, subscriber)
          else
            {{:error, :too_many_subscribers}, state}
          end
      end

    {:reply, reply, state}
  end

  @impl GenServer
  def handle_call({:unsubscribe, pid}, _from, %State{} = state) do
    state =
      update_persistent_subscription(state, pid, fn %PersistentSubscription{} = subscription ->
        :ok = stop_subscription(state, pid)

        PersistentSubscription.unsubscribe(subscription, pid)
      end)

    {:reply, :ok, state}
  end

  @impl GenServer
  def handle_call({:delete_subscription, stream_uuid, subscription_name}, _from, %State{} = state) do
    %State{persistent_subscriptions: persistent_subscriptions} = state

    {reply, state} =
      case Map.get(persistent_subscriptions, subscription_name) do
        %PersistentSubscription{stream_uuid: ^stream_uuid, subscribers: []} ->
          state = %State{
            state
            | persistent_subscriptions: Map.delete(persistent_subscriptions, subscription_name)
          }

          {:ok, state}

        nil ->
          {{:error, :subscription_not_found}, state}
      end

    {:reply, reply, state}
  end

  @impl GenServer
  def handle_call({:read_snapshot, source_uuid}, _from, %State{} = state) do
    %State{snapshots: snapshots} = state

    reply =
      case Map.get(snapshots, source_uuid, nil) do
        nil -> {:error, :snapshot_not_found}
        snapshot -> {:ok, deserialize(state, snapshot)}
      end

    {:reply, reply, state}
  end

  @impl GenServer
  def handle_call({:record_snapshot, %SnapshotData{} = snapshot}, _from, %State{} = state) do
    %SnapshotData{source_uuid: source_uuid} = snapshot
    %State{snapshots: snapshots} = state

    state = %State{state | snapshots: Map.put(snapshots, source_uuid, serialize(state, snapshot))}

    {:reply, :ok, state}
  end

  @impl GenServer
  def handle_call({:delete_snapshot, source_uuid}, _from, %State{} = state) do
    %State{snapshots: snapshots} = state

    state = %State{state | snapshots: Map.delete(snapshots, source_uuid)}

    {:reply, :ok, state}
  end

  @impl GenServer
  def handle_call(:reset!, _from, %State{} = state) do
    %State{name: name, serializer: serializer, persistent_subscriptions: persistent_subscriptions} =
      state

    for {_name, %PersistentSubscription{subscribers: subscribers}} <- persistent_subscriptions do
      for %{pid: pid} <- subscribers, is_pid(pid) do
        stop_subscription(state, pid)
      end
    end

    initial_state = %State{name: name, serializer: serializer}

    {:reply, :ok, initial_state}
  end

  @impl GenServer
  def handle_call({:ack_event, event, subscriber}, _from, %State{} = state) do
    state = ack_persistent_subscription_by_pid(state, event, subscriber)

    {:reply, :ok, state}
  end

  @impl GenServer
  def handle_info({:DOWN, _ref, :process, pid, _reason}, %State{} = state) do
    state =
      state
      |> remove_persistent_subscription_by_pid(pid)
      |> remove_transient_subscriber_by_pid(pid)

    {:noreply, state}
  end

  defp persist_events(%State{} = state, stream_uuid, existing_events, new_events) do
    %State{
      next_event_number: next_event_number,
      persisted_events: persisted_events,
      persistent_subscriptions: persistent_subscriptions,
      streams: streams
    } = state

    initial_stream_version = length(existing_events) + 1
    now = DateTime.utc_now()

    new_events =
      new_events
      |> Enum.with_index(0)
      |> Enum.map(fn {recorded_event, index} ->
        event_number = next_event_number + index
        stream_version = initial_stream_version + index

        map_to_recorded_event(event_number, stream_uuid, stream_version, now, recorded_event)
      end)
      |> Enum.map(&serialize(state, &1))

    stream_events = prepend(existing_events, new_events)
    next_event_number = List.last(new_events).event_number + 1

    state = %State{
      state
      | streams: Map.put(streams, stream_uuid, stream_events),
        persisted_events: prepend(persisted_events, new_events),
        next_event_number: next_event_number
    }

    publish_all_events = Enum.map(new_events, &deserialize(state, &1))

    publish_stream_events =
      Enum.map(publish_all_events, &set_event_number_from_version(&1, stream_uuid))

    state = publish_to_transient_subscribers(state, :all, publish_all_events)
    state = publish_to_transient_subscribers(state, stream_uuid, publish_stream_events)

    persistent_subscriptions =
      Enum.into(persistent_subscriptions, %{}, fn {subscription_name, subscription} ->
        {subscription_name, publish_events(state, subscription)}
      end)

    state = %State{state | persistent_subscriptions: persistent_subscriptions}

    {:ok, state}
  end

  defp set_event_number_from_version(%RecordedEvent{} = event, :all), do: event

  # Event number should equal stream version for stream events.
  defp set_event_number_from_version(%RecordedEvent{} = event, _stream_uuid) do
    %RecordedEvent{stream_version: stream_version} = event

    %RecordedEvent{event | event_number: stream_version}
  end

  defp prepend(list, []), do: list
  defp prepend(list, [item | remainder]), do: prepend([item | list], remainder)

  defp map_to_recorded_event(event_number, stream_uuid, stream_version, now, %EventData{} = event) do
    %EventData{
      causation_id: causation_id,
      correlation_id: correlation_id,
      event_type: event_type,
      data: data,
      metadata: metadata
    } = event

    %RecordedEvent{
      event_id: UUID.uuid4(),
      event_number: event_number,
      stream_id: stream_uuid,
      stream_version: stream_version,
      causation_id: causation_id,
      correlation_id: correlation_id,
      event_type: event_type,
      data: data,
      metadata: metadata,
      created_at: now
    }
  end

  defp start_persistent_subscription(%State{} = state, subscription, subscriber) do
    %State{name: event_store_name, persistent_subscriptions: persistent_subscriptions} = state
    %PersistentSubscription{name: subscription_name, checkpoint: checkpoint} = subscription

    supervisor_name = subscriptions_supervisor_name(event_store_name)
    subscription_spec = Subscription.child_spec(subscriber) |> Map.put(:restart, :temporary)

    {:ok, pid} = DynamicSupervisor.start_child(supervisor_name, subscription_spec)

    Process.monitor(pid)

    checkpoint = if is_nil(checkpoint), do: start_from(state, subscription), else: checkpoint

    subscription = PersistentSubscription.subscribe(subscription, pid, checkpoint)
    subscription = publish_events(state, subscription)

    persistent_subscriptions = Map.put(persistent_subscriptions, subscription_name, subscription)

    state = %State{state | persistent_subscriptions: persistent_subscriptions}

    {{:ok, pid}, state}
  end

  defp publish_events(%State{} = state, %PersistentSubscription{} = subscription) do
    %State{persisted_events: persisted_events, streams: streams} = state
    %PersistentSubscription{checkpoint: checkpoint, stream_uuid: stream_uuid} = subscription

    events =
      case stream_uuid do
        :all -> persisted_events
        stream_uuid -> Map.get(streams, stream_uuid, [])
      end

    position = if checkpoint == 0, do: -1, else: -(checkpoint + 1)

    case Enum.at(events, position) do
      %RecordedEvent{} = unseen_event ->
        unseen_event =
          deserialize(state, unseen_event) |> set_event_number_from_version(stream_uuid)

        case PersistentSubscription.publish(subscription, unseen_event) do
          {:ok, subscription} -> publish_events(state, subscription)
          {:error, :no_subscriber_available} -> subscription
        end

      nil ->
        subscription
    end
  end

  defp stop_subscription(%State{} = state, subscription) do
    %State{name: name} = state

    supervisor_name = subscriptions_supervisor_name(name)

    DynamicSupervisor.terminate_child(supervisor_name, subscription)
  end

  defp ack_persistent_subscription_by_pid(%State{} = state, %RecordedEvent{} = event, pid) do
    %RecordedEvent{event_number: event_number} = event

    update_persistent_subscription(state, pid, fn %PersistentSubscription{} = subscription ->
      subscription = PersistentSubscription.ack(subscription, event_number)

      publish_events(state, subscription)
    end)
  end

  defp remove_persistent_subscription_by_pid(%State{} = state, pid) do
    update_persistent_subscription(state, pid, fn %PersistentSubscription{} = subscription ->
      subscription = PersistentSubscription.unsubscribe(subscription, pid)

      publish_events(state, subscription)
    end)
  end

  defp update_persistent_subscription(%State{} = state, pid, updater)
       when is_function(updater, 1) do
    %State{persistent_subscriptions: persistent_subscriptions} = state

    case find_persistent_subscription(persistent_subscriptions, pid) do
      {subscription_name, %PersistentSubscription{} = subscription} ->
        updated_subscription = updater.(subscription)

        persistent_subscriptions =
          Map.put(persistent_subscriptions, subscription_name, updated_subscription)

        %State{state | persistent_subscriptions: persistent_subscriptions}

      _ ->
        state
    end
  end

  defp find_persistent_subscription(persistent_subscriptions, pid) do
    Enum.find(persistent_subscriptions, fn {_name, %PersistentSubscription{} = subscription} ->
      PersistentSubscription.has_subscriber?(subscription, pid)
    end)
  end

  defp remove_transient_subscriber_by_pid(%State{} = state, pid) do
    %State{transient_subscribers: transient_subscribers} = state

    transient_subscribers =
      Enum.reduce(transient_subscribers, transient_subscribers, fn
        {stream_uuid, subscribers}, acc ->
          Map.put(acc, stream_uuid, List.delete(subscribers, pid))
      end)

    %State{state | transient_subscribers: transient_subscribers}
  end

  defp start_from(%State{} = state, %PersistentSubscription{} = subscription) do
    %State{persisted_events: persisted_events, streams: streams} = state
    %PersistentSubscription{start_from: start_from, stream_uuid: stream_uuid} = subscription

    case {start_from, stream_uuid} do
      {:current, :all} -> length(persisted_events)
      {:current, stream_uuid} -> Map.get(streams, stream_uuid, []) |> length()
      {:origin, _stream_uuid} -> 0
      {position, _stream_uuid} when is_integer(position) -> position
    end
  end

  defp publish_to_transient_subscribers(%State{} = state, stream_uuid, events) do
    %State{transient_subscribers: transient_subscribers} = state

    subscribers = Map.get(transient_subscribers, stream_uuid, [])

    for subscriber <- subscribers, &is_pid/1 do
      send(subscriber, {:events, events})
    end

    state
  end

  defp serialize(%State{serializer: nil}, data), do: data

  defp serialize(%State{} = state, %RecordedEvent{} = recorded_event) do
    %State{serializer: serializer} = state
    %RecordedEvent{data: data, metadata: metadata} = recorded_event

    %RecordedEvent{
      recorded_event
      | data: serializer.serialize(data),
        metadata: serializer.serialize(metadata)
    }
  end

  defp serialize(%State{} = state, %SnapshotData{} = snapshot) do
    %State{serializer: serializer} = state
    %SnapshotData{data: data, metadata: metadata} = snapshot

    %SnapshotData{
      snapshot
      | data: serializer.serialize(data),
        metadata: serializer.serialize(metadata)
    }
  end

  defp deserialize(%State{serializer: nil}, data), do: data

  defp deserialize(%State{} = state, %RecordedEvent{} = recorded_event) do
    %State{serializer: serializer} = state
    %RecordedEvent{data: data, metadata: metadata, event_type: event_type} = recorded_event

    %RecordedEvent{
      recorded_event
      | data: serializer.deserialize(data, type: event_type),
        metadata: serializer.deserialize(metadata)
    }
  end

  defp deserialize(%State{} = state, %SnapshotData{} = snapshot) do
    %State{serializer: serializer} = state
    %SnapshotData{data: data, metadata: metadata, source_type: source_type} = snapshot

    %SnapshotData{
      snapshot
      | data: serializer.deserialize(data, type: source_type),
        metadata: serializer.deserialize(metadata)
    }
  end

  defp parse_config(application, config) do
    case Keyword.get(config, :name) do
      nil ->
        name = Module.concat([application, EventStore])

        {name, Keyword.put(config, :name, name)}

      name when is_atom(name) ->
        {name, config}

      invalid ->
        raise ArgumentError,
          message:
            "expected :name option to be an atom but got: " <>
              inspect(invalid)
    end
  end

  defp event_store_name(adapter_meta) when is_map(adapter_meta),
    do: Map.get(adapter_meta, :name)

  defp subscriptions_supervisor_name(event_store),
    do: Module.concat([event_store, SubscriptionsSupervisor])
end