Copy & Paste von jehovakel ex

Author: arfl

.formatter.exs

@@ -0,0 +1,4 @@
+# Used by "mix format"
+[
+  inputs: ["{mix,.formatter}.exs", "{config,lib,test}/**/*.{ex,exs}"]
+]

.gitignore

@@ -0,0 +1,24 @@
+# The directory Mix will write compiled artifacts to.
+/_build/
+
+# If you run "mix test --cover", coverage assets end up here.
+/cover/
+
+# The directory Mix downloads your dependencies sources to.
+/deps/
+
+# Where 3rd-party dependencies like ExDoc output generated docs.
+/doc/
+
+# Ignore .fetch files in case you like to edit your project deps locally.
+/.fetch
+
+# If the VM crashes, it generates a dump, let's ignore it too.
+erl_crash.dump
+
+# Also ignore archive artifacts (built via "mix archive.build").
+*.ez
+
+# Ignore package tarball (built via "mix hex.build").
+jehovakel_ex_event_store-*.tar
+

README.md

@@ -0,0 +1,21 @@
+# Shared
+
+**Jehovakel Library used with [Eventstore](https://github.com/commanded/eventstore)**
+
+## Installation
+
+If [available in Hex](https://hex.pm/docs/publish), the package can be installed
+by adding `jehovakel_ex_event_store` to your list of dependencies in `mix.exs`:
+
+```elixir
+def deps do
+  [
+    {:jehovakel_ex_event_store, "~> 1.0.0", git: "https://github.com/STUDITEMPS/jehovakel_ex_event_store.git"}
+  ]
+end
+```
+
+Documentation can be generated with [ExDoc](https://github.com/elixir-lang/ex_doc)
+and published on [HexDocs](https://hexdocs.pm). Once published, the docs can
+be found at [https://hexdocs.pm/jehovakel_ex_event_store](https://hexdocs.pm/jehovakel_ex_event_store).
+

config/config.exs

@@ -0,0 +1,53 @@
+# This file is responsible for configuring your application
+# and its dependencies with the aid of the Mix.Config module.
+use Mix.Config
+
+# This configuration is loaded before any dependency and is restricted
+# to this project. If another project depends on this project, this
+# file won't be loaded nor affect the parent project. For this reason,
+# if you want to provide default values for your application for
+# 3rd-party users, it should be done in your "mix.exs" file.
+
+# You can configure your application as:
+#
+#     config :jehovakel_ex_event_store, key: :value
+#
+# and access this configuration in your application as:
+#
+#     Application.get_env(:jehovakel_ex_event_store, :key)
+#
+# You can also configure a 3rd-party app:
+#
+#     config :logger, level: :info
+#
+
+# It is also possible to import configuration files, relative to this
+# directory. For example, you can emulate configuration per environment
+# by uncommenting the line below and defining dev.exs, test.exs and such.
+# Configuration from the imported file will override the ones defined
+# here (which is why it is important to import them last).
+#
+
+config :jehovakel_ex_event_store,
+  ecto_repos: [Support.JehovakelExRepo]
+
+# General Repository configuration
+config :jehovakel_ex_event_store, Support.JehovakelExRepo,
+  username: System.get_env("PG_USER") || System.get_env()["USER"],
+  password: System.get_env("PG_PASSWORD") || "",
+  port: System.get_env("PG_PORT") || "5432",
+  hostname: System.get_env("PG_HOST") || "localhost",
+  database: System.get_env("PG_NAME") || "jehovakel_ex_event_store_#{Mix.env()}",
+  pool_size: String.to_integer(System.get_env("POOL_SIZE") || "10"),
+  migration_source: "readstore_schema_migrations"
+
+config :eventstore, EventStore.Storage,
+  serializer: EventStore.TermSerializer,
+  username: System.get_env("PG_USER") || System.get_env()["USER"],
+  password: System.get_env("PG_PASSWORD") || "",
+  port: System.get_env("PG_PORT") || "5432",
+  hostname: System.get_env("PG_HOST") || "localhost",
+  database: System.get_env("PG_NAME") || "jehovakel_ex_event_store_#{Mix.env()}",
+  pool_size: String.to_integer(System.get_env("POOL_SIZE") || "10")
+
+import_config "#{Mix.env()}.exs"

config/dev.exs

@@ -0,0 +1 @@
+use Mix.Config

config/test.exs

@@ -0,0 +1,35 @@
+use Mix.Config
+
+config :jehovakel_ex_event_store,
+  ecto_repos: [Support.Repo],
+  event_stores: [JehovakelEx.EventStore]
+
+# General Repository configuration
+config :jehovakel_ex_event_store, Support.Repo,
+  adapter: Ecto.Adapters.Postgres,
+  username: System.get_env("PG_USER") || System.get_env()["USER"],
+  password: System.get_env("PG_PASSWORD") || "",
+  port: System.get_env("PG_PORT") || "5432",
+  hostname: System.get_env("PG_HOST") || "localhost",
+  database: System.get_env("PG_NAME") || "jehovakel_ex_#{Mix.env()}",
+  # Avoid collisions with eventstore `schema_migrations` relation
+  migration_source: "readstore_schema_migrations",
+  pool_size: String.to_integer(System.get_env("POOL_SIZE") || "10"),
+  # erhöht die DB-Pool checkout timeouts, so dass die Tests mehr Zeit haben und durchlaufen können
+  queue_target: 1_000,
+  queue_interval: 5_000
+
+config :jehovakel_ex_event_store, JehovakelEx.EventStore,
+  serializer: EventStore.TermSerializer,
+  username: System.get_env("PG_USER") || System.get_env()["USER"],
+  password: System.get_env("PG_PASSWORD") || "",
+  port: System.get_env("PG_PORT") || "5432",
+  hostname: System.get_env("PG_HOST") || "localhost",
+  database: System.get_env("PG_NAME") || "jehovakel_ex_#{Mix.env()}",
+  pool_size: String.to_integer(System.get_env("POOL_SIZE") || "10"),
+  # erhöht die DB-Pool checkout timeouts, so dass die Tests mehr Zeit haben und durchlaufen können
+  queue_target: 1_000,
+  queue_interval: 5_000
+
+# TODO: set level :error, which breaks lib/event_store/event_store_listener_test.exs
+config :logger, level: :error

lib/event_store/appendable_event.ex

@@ -0,0 +1,61 @@
+defprotocol Shared.AppendableEvent do
+  @doc "Returns the stream ID to which the event could be appended."
+  def stream_id(event)
+
+  @doc "Returns the list of fields to link the event to fields values inside the event"
+  def streams_to_link(event)
+end
+
+defimpl Shared.AppendableEvent, for: Any do
+  defmacro __deriving__(module, _struct, options) do
+    quote do
+      defimpl Shared.AppendableEvent, for: unquote(module) do
+        def stream_id(event) do
+          stream_id_field = Keyword.fetch!(unquote(options), :stream_id)
+          stream_id = Map.fetch!(event, stream_id_field)
+
+          unless is_binary(stream_id) do
+            raise ArgumentError, "Stream ID has to be a string, got '#{inspect(stream_id)}'."
+          end
+
+          stream_id
+        end
+
+        def streams_to_link(event) do
+          fields_to_link = Keyword.get(unquote(options), :streams_to_link, []) |> List.wrap()
+
+          invalid_links =
+            Enum.reduce(fields_to_link, %{}, fn field, errors ->
+              field_value = Map.get(event, field)
+
+              unless is_binary(field_value) do
+                Map.put(errors, field, field_value)
+              else
+                errors
+              end
+            end)
+
+          if map_size(invalid_links) > 0 do
+            invalid_links =
+              invalid_links
+              |> Map.to_list()
+              |> Enum.map(fn {field, value} ->
+                "#{field} -> #{inspect(value)}"
+              end)
+              |> Enum.join(", ")
+
+            raise ArgumentError,
+                  "Streams ids to link need to be a string, got '#{invalid_links}'."
+          end
+
+          Map.take(event, fields_to_link) |> Map.values()
+        end
+      end
+    end
+  end
+
+  # No default available
+  def stream_id(_event), do: raise(ArgumentError, "Implement the Appendable Protocol")
+
+  def streams_to_link(_event), do: []
+end

lib/event_store/appendable_event_test.exs

@@ -0,0 +1,78 @@
+defmodule Shared.AppendableEventTest do
+  use ExUnit.Case, async: true
+  require Protocol
+
+  defmodule Event do
+    defstruct [:a, :b, :c, :d]
+  end
+
+  describe "derive" do
+    setup do
+      {:ok, %{event: %Event{a: "a", b: "b", c: nil, d: %{foo: :bar}}}}
+    end
+
+    test "event id is a single field, streams_to_link optional", %{event: event} do
+      Protocol.derive(Shared.AppendableEvent, Event, stream_id: :a)
+
+      assert Shared.AppendableEvent.stream_id(event) == "a"
+      assert Shared.AppendableEvent.streams_to_link(event) == []
+    end
+
+    test "fvent id is a single field, streams_to_link empty", %{event: event} do
+      Protocol.derive(Shared.AppendableEvent, Event, stream_id: :a, streams_to_link: [])
+
+      assert Shared.AppendableEvent.stream_id(event) == "a"
+      assert Shared.AppendableEvent.streams_to_link(event) == []
+    end
+
+    test "use a single field as streams_to_link", %{event: event} do
+      Protocol.derive(Shared.AppendableEvent, Event, stream_id: :a, streams_to_link: :b)
+
+      assert Shared.AppendableEvent.stream_id(event) == "a"
+      assert Shared.AppendableEvent.streams_to_link(event) == ["b"]
+    end
+
+    test "use a list as streams_to_link", %{event: event} do
+      Protocol.derive(Shared.AppendableEvent, Event, stream_id: :a, streams_to_link: [:b])
+
+      assert Shared.AppendableEvent.stream_id(event) == "a"
+      assert Shared.AppendableEvent.streams_to_link(event) == ["b"]
+    end
+
+    test "Stream id needs to be present", %{event: event} do
+      Protocol.derive(Shared.AppendableEvent, Event, stream_id: :c)
+
+      assert_raise ArgumentError, "Stream ID has to be a string, got 'nil'.", fn ->
+        Shared.AppendableEvent.stream_id(event)
+      end
+    end
+
+    test "Stream id needs to be a string", %{event: event} do
+      Protocol.derive(Shared.AppendableEvent, Event, stream_id: :d)
+
+      assert_raise ArgumentError, "Stream ID has to be a string, got '%{foo: :bar}'.", fn ->
+        Shared.AppendableEvent.stream_id(event)
+      end
+    end
+
+    test "all Links need to be present", %{event: event} do
+      Protocol.derive(Shared.AppendableEvent, Event, stream_id: :a, streams_to_link: :c)
+
+      assert_raise ArgumentError,
+                   "Streams ids to link need to be a string, got 'c -> nil'.",
+                   fn ->
+                     Shared.AppendableEvent.streams_to_link(event)
+                   end
+    end
+
+    test "all links need to be strings", %{event: event} do
+      Protocol.derive(Shared.AppendableEvent, Event, stream_id: :a, streams_to_link: [:c, :d])
+
+      assert_raise ArgumentError,
+                   "Streams ids to link need to be a string, got 'c -> nil, d -> %{foo: :bar}'.",
+                   fn ->
+                     Shared.AppendableEvent.streams_to_link(event)
+                   end
+    end
+  end
+end

lib/event_store/event_store.ex

@@ -0,0 +1,77 @@
+defmodule Shared.EventStore do
+  defmacro __using__(_opts \\ []) do
+    quote location: :keep, generated: true do
+      @event_store_backend __MODULE__
+      alias Shared.EventStoreEvent
+      require Logger
+
+      def append_event(domain_events, metadata) do
+        domain_events = List.wrap(domain_events)
+
+        appended_events =
+          Enum.flat_map(domain_events, fn domain_event ->
+            stream_id = Shared.AppendableEvent.stream_id(domain_event)
+            {:ok, appended_event} = append_event(stream_id, domain_event, metadata)
+            appended_event
+          end)
+
+        {:ok, appended_events}
+      end
+
+      def append_event(
+            stream_uuid,
+            domain_events,
+            metadata
+          ) do
+        persisted_events = domain_events |> EventStoreEvent.wrap_for_persistence(metadata)
+
+        case @event_store_backend.append_to_stream(stream_uuid, :any_version, persisted_events) do
+          :ok ->
+            log(stream_uuid, domain_events, metadata)
+            {:ok, persisted_events}
+
+          error ->
+            error
+        end
+      end
+
+      def append_events(stream_uuid, domain_events, metadata),
+        do: append_event(stream_uuid, domain_events, metadata)
+
+      def all_events(stream_id \\ nil, opts \\ []) do
+        {:ok, events} =
+          case stream_id do
+            nil -> @event_store_backend.read_all_streams_forward()
+            stream_id when is_binary(stream_id) -> events_for_stream(stream_id)
+          end
+
+        if Keyword.get(opts, :unwrap, true) do
+          Enum.map(events, &Shared.EventStoreEvent.unwrap/1)
+        else
+          events
+        end
+      end
+
+      defp events_for_stream(stream_id) do
+        case @event_store_backend.read_stream_forward(stream_id) do
+          {:error, :stream_not_found} -> {:ok, []}
+          {:ok, events} -> {:ok, events}
+        end
+      end
+
+      defp log(stream_uuid, events, metadata) do
+        events = List.wrap(events)
+
+        Enum.each(events, fn event ->
+          logged_event = Shared.LoggableEvent.to_log(event)
+
+          Logger.info(fn ->
+            "Appended event stream_uuid=#{stream_uuid} event=[#{logged_event}] metadata=#{
+              metadata |> inspect()
+            }"
+          end)
+        end)
+      end
+    end
+  end
+end

lib/event_store/event_store_event.ex

@@ -0,0 +1,30 @@
+defmodule Shared.EventStoreEvent do
+  def wrap_for_persistence(events, metadata) do
+    events = List.wrap(events)
+    metadata = Enum.into(metadata, %{})
+
+    Enum.map(events, fn event ->
+      %EventStore.EventData{
+        data: event,
+        metadata: metadata
+      }
+    end)
+  end
+
+  def unwrap(%EventStore.RecordedEvent{data: domain_event, metadata: metadata} = event) do
+    metadata = Enum.into(metadata, %{})
+
+    recorded_event_metadata =
+      Map.take(event, [
+        :event_number,
+        :event_id,
+        :stream_uuid,
+        :stream_version,
+        :correlation_id,
+        :causation_id,
+        :created_at
+      ])
+
+    {domain_event, Map.merge(recorded_event_metadata, metadata)}
+  end
+end

lib/event_store/event_store_event_test.exs

@@ -0,0 +1,24 @@
+defmodule Shared.EventStoreEventTest do
+  use ExUnit.Case, async: true
+  alias Shared.EventStoreEvent, as: Event
+
+  defmodule TestEvent do
+    defstruct foo: "bar"
+  end
+
+  test "wrappe in EventStore.EventData" do
+    event = %TestEvent{}
+
+    assert [%EventStore.EventData{} = event_data] =
+             Event.wrap_for_persistence([event], %{
+               my: "metadata",
+               causation_id: "causation_id",
+               correlation_id: "correlation_id"
+             })
+
+    assert event_data.data == event
+    assert event_data.metadata.correlation_id == "correlation_id"
+    assert event_data.metadata.my == "metadata"
+    assert event_data.metadata.causation_id == "causation_id"
+  end
+end

lib/event_store/event_store_listener.ex

@@ -0,0 +1,337 @@
+defmodule Shared.EventStoreListener do
+  use GenServer
+  require Logger
+  alias EventStore.RecordedEvent
+
+  defmodule ErrorContext do
+    defstruct error_count: 0, max_retries: 3, delay_factor: 10
+
+    @type t :: %__MODULE__{
+            error_count: integer,
+            max_retries: integer,
+            delay_factor: integer
+          }
+
+    def new do
+      %__MODULE__{error_count: 0, max_retries: 3}
+    end
+
+    def record_error(%__MODULE__{} = context) do
+      Map.update(context, :error_count, 1, fn error_count -> error_count + 1 end)
+    end
+
+    def retry?(%__MODULE__{error_count: error_count, max_retries: max_retries}) do
+      error_count <= max_retries
+    end
+
+    def retry_count(%__MODULE__{error_count: error_count}) do
+      error_count - 1
+    end
+
+    def delay(%__MODULE__{
+          error_count: error_count,
+          max_retries: max_retries,
+          delay_factor: delay_factor
+        })
+        when error_count <= max_retries do
+      # Exponential backoff
+      sleep_duration = (:math.pow(2, error_count) * delay_factor) |> round()
+
+      Process.sleep(sleep_duration)
+    end
+  end
+
+  @type domain_event :: struct()
+  @type metadata :: map()
+  @type error_context :: ErrorContext.t()
+  @type state :: map() | list()
+  @type handle_result :: :ok | {:error, reason :: any()}
+
+  @callback handle(domain_event(), metadata()) :: handle_result()
+  @callback handle(domain_event(), metadata(), state()) :: handle_result()
+  @callback on_error(
+              error :: term(),
+              failed_event :: domain_event(),
+              metadata :: metadata(),
+              error_context :: error_context()
+            ) ::
+              {:retry, error_context :: error_context()}
+              | {:retry, delay :: non_neg_integer(), error_context :: error_context()}
+              | :skip
+              | {:stop, reason :: term()}
+
+  @callback on_error(
+              error :: term(),
+              stacktrace :: list(),
+              failed_event :: domain_event(),
+              metadata :: metadata(),
+              error_context :: error_context()
+            ) ::
+              {:retry, error_context :: error_context()}
+              | {:retry, delay :: non_neg_integer(), error_context :: error_context()}
+              | :skip
+              | {:stop, reason :: term()}
+
+  defmacro __using__(opts) do
+    opts = opts || []
+
+    quote location: :keep do
+      @opts unquote(opts) || []
+      @name @opts[:name] || __MODULE__
+
+      @behaviour Shared.EventStoreListener
+
+      # Verhindere, dass`@subscription_key` eine Warnung produzieren, falls nicht gesetzt.
+      unless Module.get_attribute(__MODULE__, :subscription_key) do
+        Module.put_attribute(__MODULE__, :subscription_key, nil)
+      end
+
+      # Adds default handle method
+      @before_compile unquote(__MODULE__)
+
+      def start_link(opts \\ []) do
+        opts = Keyword.merge(@opts, opts)
+        Shared.EventStoreListener.start_link(@name, __MODULE__, opts)
+      end
+
+      def child_spec(opts) do
+        default = %{
+          id: @name,
+          start: {__MODULE__, :start_link, [opts]},
+          restart: :permanent,
+          type: :worker
+        }
+
+        Supervisor.child_spec(default, [])
+      end
+    end
+  end
+
+  def start_link(name, handler_module, opts) do
+    default_opts = %{
+      name: nil,
+      handler_module: nil,
+      subscription_key: nil,
+      subscription: nil,
+      start_from: :origin,
+      event_store: nil
+    }
+
+    opts = Enum.into(opts, default_opts)
+    opts[:event_store] || raise "Event Store(event_store: My.EventStore) configuration is missing"
+
+    state = %{opts | handler_module: handler_module, name: name}
+
+    GenServer.start_link(__MODULE__, state, name: name)
+  end
+
+  defmacro __before_compile__(_env) do
+    quote generated: true do
+      def init(state) do
+        state =
+          case @subscription_key do
+            subscription_key when is_binary(subscription_key) and subscription_key != "" ->
+              %{state | subscription_key: subscription_key}
+
+            _ ->
+              state
+          end
+
+        {:ok, state}
+      end
+
+      defoverridable init: 1
+
+      def handle(_event, _metadata), do: :ok
+      defoverridable handle: 2
+
+      def handle(event, metadata, _state), do: handle(event, metadata)
+      defoverridable handle: 3
+
+      def on_error({:error, reason}, _event, _metadata, error_context),
+        do: {:retry, error_context}
+
+      defoverridable on_error: 4
+
+      def on_error(error, _stacktrace, event, metadata, error_context),
+        do: on_error(error, event, metadata, error_context)
+
+      defoverridable on_error: 5
+    end
+  end
+
+  @impl true
+  def init(
+        %{name: handler_name, handler_module: handler_module, event_store: event_store} = state
+      ) do
+    with {:ok, new_state} <- handler_module.init(state),
+         subscription_key = new_state[:subscription_key] || subscription_key_for(handler_name),
+         start_from = new_state[:start_from] || :origin,
+         {:ok, subscription} <-
+           event_store.subscribe_to_all_streams(
+             subscription_key,
+             self(),
+             start_from: start_from
+           ) do
+      {:ok, %{new_state | subscription: subscription}}
+    end
+  end
+
+  @impl true
+  def handle_info({:subscribed, _subscription}, %{name: name} = state) do
+    Logger.debug(fn ->
+      "#{name} sucessfully subscribed to event store."
+    end)
+
+    {:noreply, state}
+  end
+
+  @impl true
+  def handle_info({:events, events}, %{name: name} = state) do
+    Logger.debug(fn -> "#{name} received events: #{inspect(events)}" end)
+
+    try do
+      Enum.each(events, fn event -> handle_event(event, state, ErrorContext.new()) end)
+      {:noreply, state}
+    catch
+      {:error, reason} ->
+        {:stop, reason, state}
+    end
+  end
+
+  @impl true
+  def handle_call(:get_state, _from, state) do
+    {:reply, state, state}
+  end
+
+  defp handle_event(
+         %RecordedEvent{} = event,
+         %{name: name} = state,
+         %ErrorContext{} = error_context
+       ) do
+    case delegate_event_to_handler(event, state) do
+      :ok ->
+        ack_event(event, state)
+
+      {:ok, _} ->
+        ack_event(event, state)
+
+      {:error, reason} ->
+        Logger.error(fn ->
+          "#{name} failed to handle event #{inspect(event)} due to #{inspect(reason)}"
+        end)
+
+        handle_error({:error, reason}, current_stacktrace(), event, state, error_context)
+
+      {:error, reason, stacktrace} ->
+        Logger.error(fn ->
+          "#{name} failed to handle event #{inspect(event)} due to #{inspect(reason)}"
+        end)
+
+        handle_error({:error, reason}, stacktrace, event, state, error_context)
+    end
+  end
+
+  defp delegate_event_to_handler(
+         %RecordedEvent{} = event,
+         %{
+           handler_module: handler_module
+         } = state
+       ) do
+    try do
+      {domain_event, metadata} = Shared.EventStoreEvent.unwrap(event)
+      handler_module.handle(domain_event, metadata, state)
+    rescue
+      error ->
+        {:error, error, __STACKTRACE__}
+    end
+  end
+
+  defp handle_error(
+         error,
+         stacktrace,
+         event,
+         %{handler_module: handler_module, name: name} = state,
+         context
+       ) do
+    %RecordedEvent{data: domain_event, metadata: metadata} = event
+
+    case handler_module.on_error(error, stacktrace, domain_event, metadata, context) do
+      {:retry, %ErrorContext{} = context} ->
+        context = ErrorContext.record_error(context)
+
+        if ErrorContext.retry?(context) do
+          ErrorContext.delay(context)
+
+          Logger.warn(fn ->
+            "#{name} is retrying (#{context.error_count}/#{context.max_retries}) failed event #{
+              inspect(event)
+            }"
+          end)
+
+          handle_event(event, state, context)
+        else
+          reason =
+            "#{name} is dying due to bad event after #{ErrorContext.retry_count(context)} retries #{
+              inspect(error)
+            }, Stacktrace: #{inspect(stacktrace)}"
+
+          Logger.warn(reason)
+
+          throw({:error, reason})
+        end
+
+      :skip ->
+        Logger.debug(fn ->
+          "#{name} is skipping event #{inspect(event)}"
+        end)
+
+        ack_event(event, state)
+
+      {:stop, reason} ->
+        reason = "#{name} has requested to stop in on_error/5 callback with #{inspect(reason)}"
+
+        Logger.warn(reason)
+        throw({:error, reason})
+
+      error ->
+        Logger.warn(fn ->
+          "#{name} on_error/5 returned an invalid response #{inspect(error)}"
+        end)
+
+        throw(error)
+    end
+  end
+
+  defp current_stacktrace do
+    case Process.info(self(), :current_stacktrace) do
+      {:current_stacktrace, stacktrace} -> stacktrace
+      nil -> "Process is not alive. No stacktrace available"
+    end
+  end
+
+  defp ack_event(event, %{subscription: subscription, event_store: event_store}) do
+    :ok = event_store.ack(subscription, event)
+  end
+
+  @deprecated """
+  Set `subscription_key` on initialization. Otherwise a change of the file name would break the subscription and all the events get processed again.
+  """
+  defp subscription_key_for(handler) do
+    subscription_key =
+      handler
+      |> Atom.to_string()
+      |> String.split(".")
+      |> Enum.at(-1)
+      |> Macro.underscore()
+      |> Kernel.<>("_event_listener")
+
+    Logger.warn(
+      "Please specify a `subscription_key` on initialization for `#{handler}`. Otherwise a change of the file name would break the subscription and all the events get processed again. Default was: \"#{
+        subscription_key
+      }\""
+    )
+
+    subscription_key
+  end
+end

lib/event_store/event_store_listener_test.exs

@@ -0,0 +1,114 @@
+defmodule Shared.EventStoreListenerTest do
+  use Support.EventStoreCase, async: false
+  import ExUnit.CaptureLog
+
+  @event %Shared.EventTest.FakeEvent{}
+
+  defmodule EventHandlingError do
+    defexception [:message]
+  end
+
+  defmodule Counter do
+    use Agent
+
+    def start_link(initial_value) do
+      Agent.start_link(fn -> initial_value end, name: __MODULE__)
+    end
+
+    def increment do
+      Agent.get_and_update(__MODULE__, &{&1, &1 + 1})
+    end
+  end
+
+  defmodule ExampleConsumer do
+    use Shared.EventStoreListener,
+      subscription_key: "example_consumer",
+      event_store: JehovakelEx.EventStore
+
+    def handle(_event, %{test_pid: test_pid, raise_until: raise_until}) do
+      case Counter.increment() do
+        count when count <= raise_until ->
+          send(test_pid, :exception_during_event_handling)
+          raise EventHandlingError, "BAM BAM BAM"
+
+        _ ->
+          send(test_pid, :event_handled_successfully)
+      end
+
+      :ok
+    end
+  end
+
+  setup do
+    old_log_level = Logger.level()
+    Logger.configure(level: :warn)
+
+    start_supervised!(ExampleConsumer)
+    {:ok, _pid} = Counter.start_link(0)
+
+    on_exit fn ->
+      Logger.configure(level: old_log_level)
+    end
+    :ok
+  end
+
+  describe "Retry" do
+    test "automatically on Exception during event handling without GenServer restart" do
+      capture_log([level: :warn], fn ->
+        {:ok, _events} =
+          JehovakelEx.EventStore.append_event(@event, %{test_pid: self(), raise_until: 0})
+
+        assert_receive :exception_during_event_handling, 500
+        assert_receive :event_handled_successfully, 500
+      end)
+    end
+
+    test "does not restart Listener process" do
+      capture_log([level: :warn], fn ->
+        listener_pid = Process.whereis(ExampleConsumer)
+
+        {:ok, _events} =
+          JehovakelEx.EventStore.append_event(@event, %{test_pid: self(), raise_until: 0})
+
+        assert_receive :event_handled_successfully, 500
+        assert listener_pid == Process.whereis(ExampleConsumer)
+      end)
+    end
+
+    test "stops EventStoreListener GenServer after 3 attempts" do
+      logs =
+        capture_log([level: :warn], fn ->
+          listener_pid = Process.whereis(ExampleConsumer)
+
+          {:ok, _events} =
+            JehovakelEx.EventStore.append_event(@event, %{test_pid: self(), raise_until: 3})
+
+          assert_receive :exception_during_event_handling
+          assert_receive :event_handled_successfully, 500
+
+          assert listener_pid != Process.whereis(ExampleConsumer)
+        end)
+
+      assert logs =~ "ExampleConsumer is retrying (1/3)"
+      assert logs =~ "ExampleConsumer is retrying (2/3)"
+      assert logs =~ "ExampleConsumer is retrying (3/3)"
+      assert logs =~ "is dying due to bad event after 3 retries"
+    end
+  end
+
+  test "Log Stacktrace on failing to handle exception during event handling" do
+    logs =
+      capture_log([level: :warn], fn ->
+        {:ok, _events} =
+          JehovakelEx.EventStore.append_event(@event, %{test_pid: self(), raise_until: 4})
+
+        assert_receive :exception_during_event_handling
+        assert_receive :event_handled_successfully, 500
+      end)
+
+    assert logs =~ "Stacktrace"
+    assert logs =~ "BAM BAM BAM"
+    assert logs =~ "Shared.EventStoreListenerTest.ExampleConsumer"
+    assert logs =~ "lib/event_store/event_store_listener_test.exs"
+  end
+end

lib/event_store/event_store_test.exs

@@ -0,0 +1,34 @@
+defmodule Shared.EventTest do
+  use Support.EventStoreCase, async: false
+
+  @event %Shared.EventTest.FakeEvent{}
+  @metadata %{meta: "data"}
+
+  test "append event to stream" do
+    assert {:ok, [%{data: @event}]} = append_event(@event, @metadata)
+
+    assert [%EventStore.RecordedEvent{data: @event, metadata: @metadata}] =
+             all_events(nil, unwrap: false)
+
+    assert [{@event, @metadata}] = all_events()
+  end
+
+  test "append event to stream with stream_uuid" do
+    assert {:ok, [%{data: @event}]} = append_event("stream_uuid", @event, @metadata)
+
+    assert [%EventStore.RecordedEvent{data: @event, metadata: @metadata}] =
+             all_events(nil, unwrap: false)
+
+    assert [{@event, @metadata}] = all_events()
+    assert [{@event, @metadata}] = all_events("stream_uuid")
+  end
+
+  test "append list of events" do
+    assert {:ok, [%{data: @event}]} = append_event([@event], @metadata)
+
+    assert [%EventStore.RecordedEvent{data: @event, metadata: @metadata}] =
+             all_events(nil, unwrap: false)
+
+    assert [{@event, @metadata}] = all_events()
+  end
+end

lib/event_store/link_appendable_events.ex

@@ -0,0 +1,78 @@
+defmodule Shared.LinkAppendableEvents do
+  use GenServer
+
+  def child_spec(opts) do
+    start_options = %{
+      id: __MODULE__,
+      start: {__MODULE__, :start_link, [opts]},
+      restart: :permanent,
+      type: :worker
+    }
+
+    Supervisor.child_spec(start_options, [])
+  end
+
+  def start_link(opts \\ []) do
+    GenServer.start_link(__MODULE__, opts)
+  end
+
+  # Opts are all options available for EventStore.subscribe_to_all_streams/3
+  def init(opts) do
+    event_store = Keyword.fetch!(opts, :event_store)
+
+    {subscription_name, opts} = Keyword.get_and_update(opts, :subscription_name, fn _ -> :pop end)
+    subscription_name = subscription_name || "__jehovakel_ex_event_store_link_appendable_events__"
+    # Subscribe to events from all streams
+    {:ok, subscription} = event_store.subscribe_to_all_streams(subscription_name, self(), opts)
+
+    {:ok, %{event_store: event_store, subscription: subscription}}
+  end
+
+  # Successfully subscribed to all streams
+  def handle_info({:subscribed, subscription}, %{subscription: subscription} = state) do
+    {:noreply, state}
+  end
+
+  # Event notification
+  def handle_info(
+        {:events, events},
+        %{subscription: subscription, event_store: event_store} = state
+      ) do
+    link_appendable_events(event_store, subscription, events)
+
+    {:noreply, state}
+  end
+
+  defp link_appendable_events(event_store, subscription, events) do
+    for event <- events do
+      link_appendable_event(event_store, subscription, event)
+    end
+  end
+
+  defp link_appendable_event(
+         event_store,
+         subscription,
+         %EventStore.RecordedEvent{} = eventstore_event
+       ) do
+    event = eventstore_event.data
+
+    if Shared.AppendableEvent.impl_for(event) do
+      streams_to_link = Shared.AppendableEvent.streams_to_link(event)
+
+      for stream <- streams_to_link do
+        :ok = link_appendable_event(event_store, eventstore_event, stream)
+      end
+    end
+
+    # Confirm receipt of received events
+    :ok = event_store.ack(subscription, eventstore_event)
+  end
+
+  defp link_appendable_event(event_store, %EventStore.RecordedEvent{} = event, stream_id) do
+    case event_store.link_to_stream(stream_id, :any_version, [event]) do
+      :ok -> :ok
+      {:error, :duplicate_event} -> :ok
+      error -> error
+    end
+  end
+end

lib/event_store/link_appendable_events_test.exs

@@ -0,0 +1,55 @@
+defmodule Shared.LinkAppendableEventsTest do
+  use Support.EventStoreCase, async: false
+
+  defmodule TestEvent do
+    @derive {Shared.AppendableEvent, stream_id: :a, streams_to_link: :b}
+    defstruct [:a, :b, :c, :d]
+  end
+
+  setup do
+    event = %TestEvent{a: "id_a", b: "id_b", c: "some_value_for_c", d: :something}
+
+    start_supervised!({Shared.LinkAppendableEvents, event_store: JehovakelEx.EventStore})
+
+    {:ok, %{event: event}}
+  end
+
+  test "link an event to the streams defined in AppendableEvent", %{event: event} do
+    assert {:ok, _event_data} = JehovakelEx.EventStore.append_event(event, _metadata = %{})
+
+    wait_until(fn ->
+      assert [{^event, metadata}] = all_events("id_a")
+      assert [{^event, metadata}] = all_events("id_b")
+    end)
+  end
+
+  test "link an event to the stream twice does not produce errors", %{
+    event: event,
+    postgrex_connection: postgrex
+  } do
+    assert {:ok, _event_data} = JehovakelEx.EventStore.append_event(event, _metadata = %{})
+
+    wait_until(fn ->
+      assert [{^event, metadata}] = all_events("id_a")
+      assert [{^event, metadata}] = all_events("id_b")
+    end)
+
+    # Simulate linking running twice
+    assert :ok = stop_supervised(Shared.LinkAppendableEvents)
+
+    Postgrex.query!(postgrex, "DELETE FROM subscriptions WHERE true", [])
+
+    start_supervised!(
+      {Shared.LinkAppendableEvents, event_store: JehovakelEx.EventStore, subscription_name: "FOO"}
+    )
+
+    wait_until(fn ->
+      assert %{num_rows: 1, rows: [[1]]} =
+               Postgrex.query!(
+                 postgrex,
+                 "SELECT last_seen FROM subscriptions WHERE subscription_name='FOO'",
+                 []
+               )
+    end)
+  end
+end

lib/event_store/loggable_event.ex

@@ -0,0 +1,22 @@
+defprotocol Shared.LoggableEvent do
+  @doc "Converts an event to a loggable String"
+  @fallback_to_any true
+  def to_log(event)
+end
+
+defimpl Shared.LoggableEvent, for: Any do
+  def to_log(%event_type{} = event) do
+    event_type = event_type |> Atom.to_string() |> String.split(".") |> Enum.at(-1)
+
+    event_data =
+      event
+      |> Map.from_struct()
+      |> Enum.reduce("", fn {key, value}, event_as_string ->
+        event_as_string <> " #{key}=" <> inspect(value)
+      end)
+
+    ~s(#{event_type}:#{event_data})
+  end
+
+  def to_log(_event), do: raise(ArgumentError, "Implement the Shared.LoggableEvent Protocol")
+end

lib/event_store/loggable_event_test.exs

@@ -0,0 +1,19 @@
+defmodule Shared.LoggableEventTest do
+  use ExUnit.Case, async: true
+  require Protocol
+
+  defmodule Event do
+    defstruct [:a, :b, :c, :d]
+  end
+
+  defmodule Strct do
+    defstruct attr: :brr
+  end
+
+  test "to_log/1" do
+    event = %Event{a: "foo", b: 23, c: nil, d: %Strct{}}
+
+    assert "Event: a=\"foo\" b=23 c=nil d=%Shared.LoggableEventTest.Strct{attr: :brr}" ==
+             Shared.LoggableEvent.to_log(event)
+  end
+end

lib/event_store/migration/event.ex

@@ -0,0 +1,84 @@
+if Code.ensure_loaded?(Ecto) && Code.ensure_loaded?(Shared.Ecto.Term) do
+  defmodule Shared.EventStore.Migration.Event do
+    @moduledoc """
+    Warnung: Diese Datei ist ungetestet. Vergewissere dich, dass du deine Skripte, die darauf basieren gut lokal und
+    auf Staging getestet sind, bevor du diese auf Production loslässt!!!
+    """
+    use Ecto.Schema
+    import Ecto.Query
+    import Ecto.Changeset
+    require Logger
+
+    @primary_key {:event_id, :binary_id, autogenerate: false}
+    schema "events" do
+      field(:event_type, :string)
+      field(:data, Shared.Ecto.Term)
+      field(:metadata, Shared.Ecto.Term)
+      field(:created_at, :utc_datetime)
+    end
+
+    def migrate_event(event_type_to_migrate, migration, repository)
+        when is_atom(event_type_to_migrate) and is_function(migration) do
+      event_type = Atom.to_string(event_type_to_migrate)
+
+      query =
+        from(
+          e in Shared.EventStore.Migration.Event,
+          where: e.event_type == ^event_type
+        )
+
+      anzahl_events = repository.aggregate(query, :count)
+
+      events = repository.stream(query)
+
+      if anzahl_events > 0 do
+        Logger.info(
+          "Migrating " <>
+            to_string(anzahl_events) <> " Events vom Typ " <> to_string(event_type_to_migrate)
+        )
+
+        Ecto.Adapters.SQL.query!(repository, "DROP RULE no_update_events ON events")
+
+        repository.transaction(
+          fn ->
+            Enum.each(events, fn event ->
+              {new_data, new_metadata} = run_migration(migration, event)
+              %event_module{} = new_data
+              event_type = Atom.to_string(event_module)
+
+              {:ok, migrated_at} = DateTime.now("Europe/Berlin")
+
+              new_metadata =
+                new_metadata
+                |> Enum.into(%{})
+                |> Map.merge(%{migrated_at: migrated_at, original_event: event.data})
+
+              changeset =
+                change(event, event_type: event_type, data: new_data, metadata: new_metadata)
+
+              repository.update!(changeset)
+            end)
+          end,
+          timeout: 600_000_000
+        )
+      end
+    after
+      Ecto.Adapters.SQL.query!(
+        repository,
+        "CREATE RULE no_update_events AS ON UPDATE TO events DO INSTEAD NOTHING"
+      )
+    end
+
+    defp run_migration(migration, event) when is_function(migration, 2) do
+      migration.(event.data, event.metadata)
+    end
+
+    defp run_migration(migration, event) when is_function(migration, 3) do
+      migration.(event.data, event.metadata, %{
+        id: event.event_id,
+        type: event.event_type,
+        created_at: event.created_at
+      })
+    end
+  end
+end

lib/test_helper.exs

@@ -0,0 +1 @@
+ExUnit.start(capture_log: true)

mix.exs

@@ -0,0 +1,69 @@
+defmodule Shared.MixProject do
+  use Mix.Project
+
+  def project do
+    [
+      app: :jehovakel_ex_event_store,
+      version: "1.0.0",
+      elixir: "~> 1.7",
+      elixirc_paths: elixirc_paths(Mix.env()),
+      start_permanent: Mix.env() == :prod,
+      aliases: aliases(),
+      test_paths: ["lib"],
+      test_coverage: [tool: ExCoveralls],
+      deps: deps(),
+      aliases: aliases(),
+      consolidate_protocols: Mix.env() != :test,
+      name: "Jehovakel EX EventStore",
+      source_url: "https://github.com/STUDITEMPS/jehovakel_ex_event_store",
+      description: description(),
+      package: package()
+    ]
+  end
+
+  # Run "mix help compile.app" to learn about applications.
+  def application do
+    [
+      extra_applications: [:logger]
+    ]
+  end
+
+  # Specifies which paths to compile per environment.
+  defp elixirc_paths(:test), do: ["lib", "test/support"]
+  defp elixirc_paths(_), do: ["lib"]
+
+  defp aliases do
+    [
+      test: ["ecto.create --quiet", "event_store.init", "ecto.migrate", "test"]
+    ]
+  end
+
+  # Run "mix help deps" to learn about dependencies.
+  defp deps do
+    [
+      # CQRS event store using PostgreSQL for persistence
+      {:eventstore, "~> 1.0"},
+      {:ecto, "~> 3.0", optional: true},
+      {:ecto_sql, "~> 3.0", optional: true},
+      # {:jehovakel_ex_ecto, ">= 0.0.0", optional: true, in_umbrella: true},
+      {:excoveralls, ">= 0.10.5", only: :test}
+    ]
+  end
+
+  defp description do
+    "TODO: describe this package"
+  end
+
+  defp package do
+    [
+      # This option is only needed when you don't want to use the OTP application name
+      name: "jehovakel_ex_event_store",
+      # These are the default files included in the package
+      licenses: ["MIT License"],
+      links: %{
+        "GitHub" => "https://github.com/STUDITEMPS/jehovakel_ex_event_store",
+        "Studitemps" => "https://tech.studitemps.de"
+      }
+    ]
+  end
+end

mix.lock

@@ -0,0 +1,24 @@
+%{
+  "certifi": {:hex, :certifi, "2.5.2", "b7cfeae9d2ed395695dd8201c57a2d019c0c43ecaf8b8bcb9320b40d6662f340", [:rebar3], [{:parse_trans, "~>3.3", [hex: :parse_trans, repo: "hexpm", optional: false]}], "hexpm", "3b3b5f36493004ac3455966991eaf6e768ce9884693d9968055aeeeb1e575040"},
+  "connection": {:hex, :connection, "1.0.4", "a1cae72211f0eef17705aaededacac3eb30e6625b04a6117c1b2db6ace7d5976", [:mix], [], "hexpm", "4a0850c9be22a43af9920a71ab17c051f5f7d45c209e40269a1938832510e4d9"},
+  "db_connection": {:hex, :db_connection, "2.3.0", "d56ef906956a37959bcb385704fc04035f4f43c0f560dd23e00740daf8028c49", [:mix], [{:connection, "~> 1.0.2", [hex: :connection, repo: "hexpm", optional: false]}], "hexpm", "dcc082b8f723de9a630451b49fdbd7a59b065c4b38176fb147aaf773574d4520"},
+  "decimal": {:hex, :decimal, "2.0.0", "a78296e617b0f5dd4c6caf57c714431347912ffb1d0842e998e9792b5642d697", [:mix], [], "hexpm", "34666e9c55dea81013e77d9d87370fe6cb6291d1ef32f46a1600230b1d44f577"},
+  "ecto": {:hex, :ecto, "3.5.4", "73ee115deb10769c73fd2d27e19e36bc4af7c56711ad063616a86aec44f80f6f", [:mix], [{:decimal, "~> 1.6 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "7f13f9c9c071bd2ca04652373ff3edd1d686364de573255096872a4abc471807"},
+  "ecto_sql": {:hex, :ecto_sql, "3.5.3", "1964df0305538364b97cc4661a2bd2b6c89d803e66e5655e4e55ff1571943efd", [:mix], [{:db_connection, "~> 2.2", [hex: :db_connection, repo: "hexpm", optional: false]}, {:ecto, "~> 3.5.0", [hex: :ecto, repo: "hexpm", optional: false]}, {:myxql, "~> 0.3.0 or ~> 0.4.0", [hex: :myxql, repo: "hexpm", optional: true]}, {:postgrex, "~> 0.15.0", [hex: :postgrex, repo: "hexpm", optional: true]}, {:tds, "~> 2.1.1", [hex: :tds, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "d2f53592432ce17d3978feb8f43e8dc0705e288b0890caf06d449785f018061c"},
+  "elixir_uuid": {:hex, :elixir_uuid, "1.2.1", "dce506597acb7e6b0daeaff52ff6a9043f5919a4c3315abb4143f0b00378c097", [:mix], [], "hexpm", "f7eba2ea6c3555cea09706492716b0d87397b88946e6380898c2889d68585752"},
+  "eventstore": {:hex, :eventstore, "1.1.0", "1f5f20feb343c78cf8a38e8d288f6c7a860056c341dbe1520e6faa728c5e633f", [:mix], [{:elixir_uuid, "~> 1.2", [hex: :elixir_uuid, repo: "hexpm", optional: false]}, {:fsm, "~> 0.3", [hex: :fsm, repo: "hexpm", optional: false]}, {:gen_stage, "~> 1.0", [hex: :gen_stage, repo: "hexpm", optional: false]}, {:highlander, "~> 0.2", [hex: :highlander, repo: "hexpm", optional: false]}, {:jason, "~> 1.2", [hex: :jason, repo: "hexpm", optional: true]}, {:poolboy, "~> 1.5", [hex: :poolboy, repo: "hexpm", optional: true]}, {:postgrex, "~> 0.15", [hex: :postgrex, repo: "hexpm", optional: false]}], "hexpm", "0b61fa4972a1521efa9027c3fd3955afac2df396783357ab2295c9d424124046"},
+  "excoveralls": {:hex, :excoveralls, "0.13.3", "edc5f69218f84c2bf61b3609a22ddf1cec0fbf7d1ba79e59f4c16d42ea4347ed", [:mix], [{:hackney, "~> 1.16", [hex: :hackney, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "cc26f48d2f68666380b83d8aafda0fffc65dafcc8d8650358e0b61f6a99b1154"},
+  "fsm": {:hex, :fsm, "0.3.1", "087aa9b02779a84320dc7a2d8464452b5308e29877921b2bde81cdba32a12390", [:mix], [], "hexpm", "fbf0d53f89e9082b326b0b5828b94b4c549ff9d1452bbfd00b4d1ac082208e96"},
+  "gen_stage": {:hex, :gen_stage, "1.0.0", "51c8ae56ff54f9a2a604ca583798c210ad245f415115453b773b621c49776df5", [:mix], [], "hexpm", "1d9fc978db5305ac54e6f5fec7adf80cd893b1000cf78271564c516aa2af7706"},
+  "hackney": {:hex, :hackney, "1.16.0", "5096ac8e823e3a441477b2d187e30dd3fff1a82991a806b2003845ce72ce2d84", [:rebar3], [{:certifi, "2.5.2", [hex: :certifi, repo: "hexpm", optional: false]}, {:idna, "6.0.1", [hex: :idna, repo: "hexpm", optional: false]}, {:metrics, "1.0.1", [hex: :metrics, repo: "hexpm", optional: false]}, {:mimerl, "~>1.1", [hex: :mimerl, repo: "hexpm", optional: false]}, {:parse_trans, "3.3.0", [hex: :parse_trans, repo: "hexpm", optional: false]}, {:ssl_verify_fun, "1.1.6", [hex: :ssl_verify_fun, repo: "hexpm", optional: false]}], "hexpm", "3bf0bebbd5d3092a3543b783bf065165fa5d3ad4b899b836810e513064134e18"},
+  "highlander": {:hex, :highlander, "0.2.1", "e59b459f857e89daf73f2598bf2b2c0479a435481e6101ea389fd3625919b052", [:mix], [], "hexpm", "5ba19a18358803d82a923511acec8ee85fac30731c5ca056f2f934bc3d3afd9a"},
+  "idna": {:hex, :idna, "6.0.1", "1d038fb2e7668ce41fbf681d2c45902e52b3cb9e9c77b55334353b222c2ee50c", [:rebar3], [{:unicode_util_compat, "0.5.0", [hex: :unicode_util_compat, repo: "hexpm", optional: false]}], "hexpm", "a02c8a1c4fd601215bb0b0324c8a6986749f807ce35f25449ec9e69758708122"},
+  "jason": {:hex, :jason, "1.2.2", "ba43e3f2709fd1aa1dce90aaabfd039d000469c05c56f0b8e31978e03fa39052", [:mix], [{:decimal, "~> 1.0 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "18a228f5f0058ee183f29f9eae0805c6e59d61c3b006760668d8d18ff0d12179"},
+  "metrics": {:hex, :metrics, "1.0.1", "25f094dea2cda98213cecc3aeff09e940299d950904393b2a29d191c346a8486", [:rebar3], [], "hexpm", "69b09adddc4f74a40716ae54d140f93beb0fb8978d8636eaded0c31b6f099f16"},
+  "mimerl": {:hex, :mimerl, "1.2.0", "67e2d3f571088d5cfd3e550c383094b47159f3eee8ffa08e64106cdf5e981be3", [:rebar3], [], "hexpm", "f278585650aa581986264638ebf698f8bb19df297f66ad91b18910dfc6e19323"},
+  "parse_trans": {:hex, :parse_trans, "3.3.0", "09765507a3c7590a784615cfd421d101aec25098d50b89d7aa1d66646bc571c1", [:rebar3], [], "hexpm", "17ef63abde837ad30680ea7f857dd9e7ced9476cdd7b0394432af4bfc241b960"},
+  "postgrex": {:hex, :postgrex, "0.15.7", "724410acd48abac529d0faa6c2a379fb8ae2088e31247687b16cacc0e0883372", [:mix], [{:connection, "~> 1.0", [hex: :connection, repo: "hexpm", optional: false]}, {:db_connection, "~> 2.1", [hex: :db_connection, repo: "hexpm", optional: false]}, {:decimal, "~> 1.5 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}], "hexpm", "88310c010ff047cecd73d5ceca1d99205e4b1ab1b9abfdab7e00f5c9d20ef8f9"},
+  "ssl_verify_fun": {:hex, :ssl_verify_fun, "1.1.6", "cf344f5692c82d2cd7554f5ec8fd961548d4fd09e7d22f5b62482e5aeaebd4b0", [:make, :mix, :rebar3], [], "hexpm", "bdb0d2471f453c88ff3908e7686f86f9be327d065cc1ec16fa4540197ea04680"},
+  "telemetry": {:hex, :telemetry, "0.4.2", "2808c992455e08d6177322f14d3bdb6b625fbcfd233a73505870d8738a2f4599", [:rebar3], [], "hexpm", "2d1419bd9dda6a206d7b5852179511722e2b18812310d304620c7bd92a13fcef"},
+  "unicode_util_compat": {:hex, :unicode_util_compat, "0.5.0", "8516502659002cec19e244ebd90d312183064be95025a319a6c7e89f4bccd65b", [:rebar3], [], "hexpm", "d48d002e15f5cc105a696cf2f1bbb3fc72b4b770a184d8420c8db20da2674b38"},
+}

priv/repo/migrations/.keep

@@ -0,0 +1,4 @@
+defmodule JehovakelEx.EventStore do
+  use EventStore, otp_app: :jehovakel_ex_event_store
+  use Shared.EventStore
+end

test/support/event_store.ex

@@ -0,0 +1,48 @@
+defmodule Support.EventStoreCase do
+  @moduledoc """
+  Used for Tests using event store
+  """
+
+  use ExUnit.CaseTemplate
+
+  using do
+    quote do
+      import JehovakelEx.EventStore,
+        only: [append_event: 2, append_event: 3, all_events: 2, all_events: 1, all_events: 0],
+        warn: false
+
+      def wait_until(fun), do: wait_until(500, fun)
+
+      def wait_until(0, fun), do: fun.()
+
+      def wait_until(timeout, fun) do
+        try do
+          fun.()
+        rescue
+          ExUnit.AssertionError ->
+            :timer.sleep(100)
+            wait_until(max(0, timeout - 100), fun)
+        end
+      end
+    end
+  end
+
+  setup _tags do
+    # reset eventstore
+    config = EventStore.Config.parsed(JehovakelEx.EventStore, :jehovakel_ex_event_store)
+    postgrex_config = EventStore.Config.default_postgrex_opts(config)
+    {:ok, eventstore_connection} = Postgrex.start_link(postgrex_config)
+    EventStore.Storage.Initializer.reset!(eventstore_connection)
+    {:ok, _} = Application.ensure_all_started(:eventstore)
+
+    start_supervised!(JehovakelEx.EventStore)
+
+    on_exit(fn ->
+      # stop eventstore application
+      Application.stop(:eventstore)
+      Process.exit(eventstore_connection, :shutdown)
+    end)
+
+    {:ok, %{postgrex_connection: eventstore_connection}}
+  end
+end

test/support/event_store_case.ex

@@ -0,0 +1,21 @@
+# Because of consolidation the protocol implementation needs to live in a file
+# that is compiled before the tests are executed. Check the "Consolidation"
+# section in the documentation for Kernel.defprotocol/2
+
+defmodule Shared.EventTest.FakeEvent do
+  defstruct some: :default
+end
+
+defimpl Shared.LoggableEvent, for: Shared.EventTest.FakeEvent do
+  def to_log(_event) do
+    "FakeEvent: logging"
+  end
+end
+
+defimpl Shared.AppendableEvent, for: Shared.EventTest.FakeEvent do
+  def stream_id(event) do
+    event.some |> Atom.to_string()
+  end
+
+  def streams_to_link(_event), do: []
+end

test/support/fake_event.ex

@@ -0,0 +1,5 @@
+defmodule Support.Repo do
+  use Ecto.Repo,
+    otp_app: :jehovakel_ex_event_store,
+    adapter: Ecto.Adapters.Postgres
+end