From 6ebd67651f19e507265d7b59b590674a2b6e531e Mon Sep 17 00:00:00 2001 From: Thierry Bomandouki Date: Mon, 29 Dec 2025 22:52:00 +0100 Subject: [PATCH 01/17] Init events_root table and events partitioned table --- lib/event_store/sql/init.ex | 57 +++++++++++++++++++++++++++++++------ 1 file changed, 48 insertions(+), 9 deletions(-) diff --git a/lib/event_store/sql/init.ex b/lib/event_store/sql/init.ex index ea529782..374b9bec 100644 --- a/lib/event_store/sql/init.ex +++ b/lib/event_store/sql/init.ex @@ -58,21 +58,50 @@ defmodule EventStore.Sql.Init do """ end - defp create_events_table(column_data_type) do + # Create `events_root` table + defp create_events_root_table do """ - CREATE TABLE events + CREATE TABLE events_root ( - event_id uuid PRIMARY KEY NOT NULL, - event_type text NOT NULL, - causation_id uuid NULL, - correlation_id uuid NULL, - data #{column_data_type} NOT NULL, - metadata #{column_data_type} NULL, - created_at timestamp with time zone DEFAULT NOW() NOT NULL + event_id uuid PRIMARY KEY NOT NULL ); """ end + # Create `events` parent table + defp create_events_table(column_data_type) do + """ + CREATE TABLE events ( + event_id UUID NOT NULL, + event_type TEXT NOT NULL, + causation_id UUID NULL, + correlation_id UUID NULL, + "data" #{column_data_type} NOT NULL, + metadata #{column_data_type} NULL, + created_at TIMESTAMPTZ DEFAULT now() NOT NULL, + CONSTRAINT event_store_events_pkey PRIMARY KEY (event_id, created_at), + CONSTRAINT event_store_events_root_fk + FOREIGN KEY (event_id) + REFERENCES events_root (event_id) + ) PARTITION BY RANGE (created_at); + """ + end + + # Create `events` indexes + defp create_events_indexes(column_data_type) do + idx_query = """ + CREATE INDEX event_store_events_created_at_idx ON events(created_at); + CREATE INDEX event_store_events_event_type_idx ON events(event_type, created_at); + """ + + # Adding an index if data is a jsonb + if String.downcase(column_data_type) == "jsonb" do + idx_query <> """CREATE INDEX ON events USING GIN ("data" jsonb_path_ops);""" + else + idx_query + end + end + defp create_event_store_exception_function do """ CREATE OR REPLACE FUNCTION event_store_exception() @@ -115,6 +144,11 @@ defmodule EventStore.Sql.Init do # prevent updates to `events` table defp prevent_event_update do """ + CREATE TRIGGER no_update_events_root + BEFORE UPDATE ON events_root + FOR EACH STATEMENT + EXECUTE PROCEDURE event_store_exception('Cannot update events_root'); + CREATE TRIGGER no_update_events BEFORE UPDATE ON events FOR EACH STATEMENT @@ -129,6 +163,11 @@ defmodule EventStore.Sql.Init do BEFORE DELETE ON events FOR EACH STATEMENT EXECUTE PROCEDURE event_store_delete('Cannot delete events'); + + CREATE TRIGGER no_delete_events_root + BEFORE DELETE ON events_root + FOR EACH STATEMENT + EXECUTE PROCEDURE event_store_delete('Cannot delete events_root'); """ end From 3a22c8fb45276809a201d9993d84c558d2bef3da Mon Sep 17 00:00:00 2001 From: Thierry Bomandouki Date: Mon, 29 Dec 2025 23:02:52 +0100 Subject: [PATCH 02/17] Adding tables creation functions in init --- lib/event_store/sql/init.ex | 2 ++ 1 file changed, 2 insertions(+) diff --git a/lib/event_store/sql/init.ex b/lib/event_store/sql/init.ex index 374b9bec..b16cb78c 100644 --- a/lib/event_store/sql/init.ex +++ b/lib/event_store/sql/init.ex @@ -11,7 +11,9 @@ defmodule EventStore.Sql.Init do ~s(SET LOCAL search_path TO "#{schema}";), create_streams_table(), create_stream_uuid_index(), + create_events_root_table(), create_events_table(column_data_type), + create_events_indexes(column_data_type) create_stream_events_table(), create_stream_events_index(), create_event_store_exception_function(), From a1ce23cf32777c1a2858358716e96a3c6afcdecd Mon Sep 17 00:00:00 2001 From: Thierry Bomandouki Date: Mon, 29 Dec 2025 23:14:10 +0100 Subject: [PATCH 03/17] Inserting in events_root table before inserting in events to ensure event_id unicity --- .../sql/statements/insert_events.sql.eex | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/lib/event_store/sql/statements/insert_events.sql.eex b/lib/event_store/sql/statements/insert_events.sql.eex index 27141d76..6aaac046 100644 --- a/lib/event_store/sql/statements/insert_events.sql.eex +++ b/lib/event_store/sql/statements/insert_events.sql.eex @@ -34,6 +34,22 @@ WITH ($<%= i*9+3 %>::uuid, $<%= i*9+10 %>::int, $<%= i*9+11 %>::bigint) <% end %> ), + events_root AS ( + <% + # insert the new events into the events_root table + # using the 7 bind variables from 3 to 9 inclusive + # n.b.: the bind for the event_id is re-generated here + %> + INSERT INTO "<%= schema %>".events_root + ( + event_id + ) + VALUES + <%= for i <- 0..(number_of_events - 1) do %> + <%= unless i == 0 do %>,<% end %> + ($<%= i*9+3 %>) + <% end %> + ), events AS ( <% # insert the new events into the events table From 0822cfabffc8aa730885bfb9f8ceda76779aa536 Mon Sep 17 00:00:00 2001 From: Thierry BOMANDOUKI Date: Mon, 5 Jan 2026 12:14:26 +0100 Subject: [PATCH 04/17] Add new config parameter to EventStore to enable using a partitioned events table. --- config/config.exs | 4 + lib/event_store/config.ex | 8 ++ lib/event_store/sql/init.ex | 99 +++++++++++++------ lib/event_store/sql/statements.ex | 6 +- .../sql/statements/insert_events.sql.eex | 5 + 5 files changed, 91 insertions(+), 31 deletions(-) diff --git a/config/config.exs b/config/config.exs index 871a3d1e..8a9c51a0 100644 --- a/config/config.exs +++ b/config/config.exs @@ -1,3 +1,7 @@ import Config +# Configuration globale de EventStore +config :eventstore, EventStore, + partitioned_events: true # Default false, set to true if you want a partioned events table + import_config "#{Mix.env()}.exs" diff --git a/lib/event_store/config.ex b/lib/event_store/config.ex index 92ebaacf..b920cf23 100644 --- a/lib/event_store/config.ex +++ b/lib/event_store/config.ex @@ -154,4 +154,12 @@ defmodule EventStore.Config do config |> Keyword.merge(Keyword.get(config, :session_mode_pool, [])) end + + # Retrieve the value of the optional partitioned_events parameter, + # which indicates whether the Postgres event table is partitioned. + defp partitioned_events?(config) do + config + |> Keyword.merge(Keyword.get(config, :partitioned_events, false)) + end + end diff --git a/lib/event_store/sql/init.ex b/lib/event_store/sql/init.ex index b16cb78c..b1262183 100644 --- a/lib/event_store/sql/init.ex +++ b/lib/event_store/sql/init.ex @@ -3,24 +3,32 @@ defmodule EventStore.Sql.Init do # PostgreSQL statements to intialize an event store schema. + def create_partitioned_or_normal_events_table(partitioned, column_data_type) do + if partitioned do + create_events_root_table() + create_partitioned_events_table(column_data_type) + else + create_events_table(column_data_type) + end + def statements(config) do column_data_type = Keyword.fetch!(config, :column_data_type) schema = Keyword.fetch!(config, :schema) + partitioned = Keyword.fetch!(config, :partitioned_events, false) [ ~s(SET LOCAL search_path TO "#{schema}";), create_streams_table(), create_stream_uuid_index(), - create_events_root_table(), - create_events_table(column_data_type), + create_partitioned_or_normal_events_table(partitioned, column_data_type), create_events_indexes(column_data_type) create_stream_events_table(), create_stream_events_index(), create_event_store_exception_function(), create_event_store_delete_function(), prevent_streams_delete(), - prevent_event_delete(), - prevent_event_update(), + prevent_event_delete(partitioned), + prevent_event_update(partitioned), prevent_stream_events_delete(), prevent_stream_events_update(), create_notify_events_function(), @@ -70,8 +78,8 @@ defmodule EventStore.Sql.Init do """ end - # Create `events` parent table - defp create_events_table(column_data_type) do + # Create partitioned `events` parent table + defp create_partitioned_events_table(column_data_type) do """ CREATE TABLE events ( event_id UUID NOT NULL, @@ -89,6 +97,22 @@ defmodule EventStore.Sql.Init do """ end + # Create `events` table + defp create_events_table(column_data_type) do + """ + CREATE TABLE events + ( + event_id uuid PRIMARY KEY NOT NULL, + event_type text NOT NULL, + causation_id uuid NULL, + correlation_id uuid NULL, + data #{column_data_type} NOT NULL, + metadata #{column_data_type} NULL, + created_at timestamp with time zone DEFAULT NOW() NOT NULL + ); + """ + end + # Create `events` indexes defp create_events_indexes(column_data_type) do idx_query = """ @@ -98,7 +122,7 @@ defmodule EventStore.Sql.Init do # Adding an index if data is a jsonb if String.downcase(column_data_type) == "jsonb" do - idx_query <> """CREATE INDEX ON events USING GIN ("data" jsonb_path_ops);""" + idx_query <> "CREATE INDEX ON events USING GIN (\"data\" jsonb_path_ops);" else idx_query end @@ -144,33 +168,50 @@ defmodule EventStore.Sql.Init do end # prevent updates to `events` table - defp prevent_event_update do - """ - CREATE TRIGGER no_update_events_root - BEFORE UPDATE ON events_root - FOR EACH STATEMENT - EXECUTE PROCEDURE event_store_exception('Cannot update events_root'); + defp prevent_event_update(partitioned) do + events_trigger = """ + CREATE TRIGGER no_update_events + BEFORE UPDATE ON events + FOR EACH STATEMENT + EXECUTE PROCEDURE event_store_exception('Cannot update events'); + """ + events_root_trigger = + if partitioned do + """ + CREATE TRIGGER no_update_events_root + BEFORE UPDATE ON events_root + FOR EACH STATEMENT + EXECUTE PROCEDURE event_store_exception('Cannot update events_root'); + """ + else + "" + end - CREATE TRIGGER no_update_events - BEFORE UPDATE ON events - FOR EACH STATEMENT - EXECUTE PROCEDURE event_store_exception('Cannot update events'); - """ + events_root_trigger <> events_trigger end # prevent deletion from `events` table - defp prevent_event_delete do - """ - CREATE TRIGGER no_delete_events - BEFORE DELETE ON events - FOR EACH STATEMENT - EXECUTE PROCEDURE event_store_delete('Cannot delete events'); + defp prevent_event_delete(partitioned) do + events_trigger = """ + CREATE TRIGGER no_delete_events + BEFORE DELETE ON events + FOR EACH STATEMENT + EXECUTE PROCEDURE event_store_delete('Cannot delete events'); + """ + + events_root_trigger = + if partitioned do + """ + CREATE TRIGGER no_delete_events_root + BEFORE DELETE ON events_root + FOR EACH STATEMENT + EXECUTE PROCEDURE event_store_delete('Cannot delete events_root'); + """ + else + "" + end - CREATE TRIGGER no_delete_events_root - BEFORE DELETE ON events_root - FOR EACH STATEMENT - EXECUTE PROCEDURE event_store_delete('Cannot delete events_root'); - """ + events_root_trigger <> events_trigger end defp create_stream_events_table do diff --git a/lib/event_store/sql/statements.ex b/lib/event_store/sql/statements.ex index c9801731..33d23927 100644 --- a/lib/event_store/sql/statements.ex +++ b/lib/event_store/sql/statements.ex @@ -3,15 +3,17 @@ defmodule EventStore.Sql.Statements do require EEx - alias EventStore.Sql.{Init, Reset} + alias EventStore.Sql.{Config, Init, Reset} defdelegate initializers(config), to: Init, as: :statements defdelegate reset(config), to: Reset, as: :statements + partitioned_events = Config.partitioned_events(config) + for {fun, args} <- [ {:count_streams, [:schema]}, {:create_stream, [:schema]}, - {:insert_events, [:schema, :stream_id, :number_of_events, :created_at]}, + {:insert_events, [:schema, :stream_id, :number_of_events, :created_at, :partitioned_events]}, {:insert_events_any_version, [:schema, :stream_id, :number_of_events, :created_at]}, {:insert_link_events, [:schema, :number_of_events]}, {:soft_delete_stream, [:schema]}, diff --git a/lib/event_store/sql/statements/insert_events.sql.eex b/lib/event_store/sql/statements/insert_events.sql.eex index 6aaac046..a88c70f7 100644 --- a/lib/event_store/sql/statements/insert_events.sql.eex +++ b/lib/event_store/sql/statements/insert_events.sql.eex @@ -34,6 +34,10 @@ WITH ($<%= i*9+3 %>::uuid, $<%= i*9+10 %>::int, $<%= i*9+11 %>::bigint) <% end %> ), + <% + # if events table is a partioned table then we need to begin by inserting in events_root table + %> + <%= if @partitioned_events do %> events_root AS ( <% # insert the new events into the events_root table @@ -50,6 +54,7 @@ WITH ($<%= i*9+3 %>) <% end %> ), + <% end %> events AS ( <% # insert the new events into the events table From e2577fec5d53eb3f5f5e4fb0bfe448f59b3d9fd1 Mon Sep 17 00:00:00 2001 From: Thierry BOMANDOUKI Date: Mon, 5 Jan 2026 19:03:24 +0100 Subject: [PATCH 05/17] Add :partitioned parameter to all functions that insert events and ensure there is only one SQL statement per init function. --- lib/event_store.ex | 17 ++-- lib/event_store/sql/init.ex | 97 ++++++++++++------- lib/event_store/sql/statements.ex | 6 +- .../insert_events_any_version.sql.eex | 18 ++++ lib/event_store/storage.ex | 4 +- lib/event_store/storage/appender.ex | 10 +- lib/event_store/streams/stream.ex | 29 +++--- 7 files changed, 116 insertions(+), 65 deletions(-) diff --git a/lib/event_store.ex b/lib/event_store.ex index a02560a8..f1f2bcbd 100644 --- a/lib/event_store.ex +++ b/lib/event_store.ex @@ -118,7 +118,7 @@ defmodule EventStore do Use a dynamic event store by providing its name as an option to each function: - :ok = EventStore.append_to_stream(stream_uuid, expected_version, events, name: :eventstore1) + :ok = EventStore.append_to_stream(stream_uuid, expected_version, events, name: :eventstore1, partitioned) {:ok, events} = EventStore.read_stream_forward(stream_uuid, 0, 1_000, name: :eventstore1) @@ -183,7 +183,7 @@ defmodule EventStore do {:ok, pid} = Postgrex.start_link(config) Postgrex.transaction(pid, fn conn -> - :ok = EventStore.append_to_stream(stream_uuid, expected_version, events, conn: conn) + :ok = EventStore.append_to_stream(stream_uuid, expected_version, events, conn: conn, partitioned) end) This can also be used with an Ecto `Repo` which is configured to use the @@ -194,7 +194,7 @@ defmodule EventStore do conn = Process.get({Ecto.Adapters.SQL, pool}) - :ok = EventStore.append_to_stream(stream_uuid, expected_version, events, conn: conn) + :ok = EventStore.append_to_stream(stream_uuid, expected_version, events, conn: conn, partitioned) end) --- @@ -296,17 +296,17 @@ defmodule EventStore do @accepted_overrides_append_to_stream [:created_at_override] - def append_to_stream(stream_uuid, expected_version, events, opts \\ []) + def append_to_stream(stream_uuid, expected_version, events, opts \\ [], partitioned) - def append_to_stream(@all_stream, _expected_version, _events, _opts), + def append_to_stream(@all_stream, _expected_version, _events, _opts, _partitioned), do: {:error, :cannot_append_to_all_stream} - def append_to_stream(stream_uuid, expected_version, events, opts) do + def append_to_stream(stream_uuid, expected_version, events, opts, partitioned) do overrides = Keyword.take(opts, @accepted_overrides_append_to_stream) {conn, opts} = parse_opts(opts) opts = Keyword.merge(opts, overrides) - Stream.append_to_stream(conn, stream_uuid, expected_version, events, opts) + Stream.append_to_stream(conn, stream_uuid, expected_version, events, opts, partitioned) end def link_to_stream( @@ -658,7 +658,8 @@ defmodule EventStore do stream_uuid :: String.t(), expected_version, events :: list(EventData.t()), - opts :: options + opts :: options, + partitioned :: Boolean ) :: :ok | {:error, :cannot_append_to_all_stream} diff --git a/lib/event_store/sql/init.ex b/lib/event_store/sql/init.ex index b1262183..6a37da9c 100644 --- a/lib/event_store/sql/init.ex +++ b/lib/event_store/sql/init.ex @@ -3,32 +3,37 @@ defmodule EventStore.Sql.Init do # PostgreSQL statements to intialize an event store schema. - def create_partitioned_or_normal_events_table(partitioned, column_data_type) do + def create_partitioned_or_not_events_table(partitioned, column_data_type) do if partitioned do - create_events_root_table() create_partitioned_events_table(column_data_type) else create_events_table(column_data_type) + end end def statements(config) do column_data_type = Keyword.fetch!(config, :column_data_type) - schema = Keyword.fetch!(config, :schema) - partitioned = Keyword.fetch!(config, :partitioned_events, false) + schema = Keyword.fetch!(config, :schema) || 'eventi_store' + partitioned = Keyword.fetch!(config, :partitioned_events) || false [ ~s(SET LOCAL search_path TO "#{schema}";), create_streams_table(), create_stream_uuid_index(), - create_partitioned_or_normal_events_table(partitioned, column_data_type), - create_events_indexes(column_data_type) - create_stream_events_table(), + create_events_root_table(partitioned), + create_partitioned_or_not_events_table(partitioned, column_data_type), + create_events_index_1(), + create_events_index_2(), + create_events_index_3(column_data_type), + create_stream_events_table(partitioned), create_stream_events_index(), create_event_store_exception_function(), create_event_store_delete_function(), prevent_streams_delete(), - prevent_event_delete(partitioned), - prevent_event_update(partitioned), + prevent_event_delete(), + prevent_event_root_delete(partitioned), + prevent_event_update(), + prevent_event_root_update(partitioned), prevent_stream_events_delete(), prevent_stream_events_update(), create_notify_events_function(), @@ -69,13 +74,17 @@ defmodule EventStore.Sql.Init do end # Create `events_root` table - defp create_events_root_table do - """ - CREATE TABLE events_root - ( - event_id uuid PRIMARY KEY NOT NULL - ); - """ + defp create_events_root_table(partitioned) do + if partitioned do + """ + CREATE TABLE events_root + ( + event_id uuid PRIMARY KEY NOT NULL + ); + """ + else + "" + end end # Create partitioned `events` parent table @@ -114,17 +123,26 @@ defmodule EventStore.Sql.Init do end # Create `events` indexes - defp create_events_indexes(column_data_type) do - idx_query = """ + defp create_events_index_1 do + """ CREATE INDEX event_store_events_created_at_idx ON events(created_at); + """ + end + + defp create_events_index_2 do + """ CREATE INDEX event_store_events_event_type_idx ON events(event_type, created_at); """ + end - # Adding an index if data is a jsonb + defp create_events_index_3(column_data_type) do + # Adding an index if data is a jsonb if String.downcase(column_data_type) == "jsonb" do - idx_query <> "CREATE INDEX ON events USING GIN (\"data\" jsonb_path_ops);" + """ + CREATE INDEX ON events USING GIN ("data" jsonb_path_ops); + """ else - idx_query + "SELECT 1;" end end @@ -168,14 +186,17 @@ defmodule EventStore.Sql.Init do end # prevent updates to `events` table - defp prevent_event_update(partitioned) do - events_trigger = """ + defp prevent_event_update() do + """ CREATE TRIGGER no_update_events BEFORE UPDATE ON events FOR EACH STATEMENT EXECUTE PROCEDURE event_store_exception('Cannot update events'); """ - events_root_trigger = + end + + # prevent updates to `events_root` table + defp prevent_event_root_update(partitioned) do if partitioned do """ CREATE TRIGGER no_update_events_root @@ -184,22 +205,23 @@ defmodule EventStore.Sql.Init do EXECUTE PROCEDURE event_store_exception('Cannot update events_root'); """ else - "" + "SELECT 1;" end - - events_root_trigger <> events_trigger end # prevent deletion from `events` table - defp prevent_event_delete(partitioned) do - events_trigger = """ + defp prevent_event_delete() do + """ CREATE TRIGGER no_delete_events BEFORE DELETE ON events FOR EACH STATEMENT EXECUTE PROCEDURE event_store_delete('Cannot delete events'); """ + end + - events_root_trigger = + # prevent deletion from `events_root` table + defp prevent_event_root_delete(partitioned) do if partitioned do """ CREATE TRIGGER no_delete_events_root @@ -208,17 +230,21 @@ defmodule EventStore.Sql.Init do EXECUTE PROCEDURE event_store_delete('Cannot delete events_root'); """ else - "" + "SELECT 1;" end - - events_root_trigger <> events_trigger end - defp create_stream_events_table do + defp create_stream_events_table(partitioned) do + events_table = + if partitioned do + "events_root" + else + "events" + end """ CREATE TABLE stream_events ( - event_id uuid NOT NULL REFERENCES events (event_id), + event_id uuid NOT NULL REFERENCES #{events_table} (event_id), stream_id bigint NOT NULL REFERENCES streams (stream_id), stream_version bigint NOT NULL, original_stream_id bigint REFERENCES streams (stream_id), @@ -359,4 +385,5 @@ defmodule EventStore.Sql.Init do VALUES (1, 3, 2); """ end + end diff --git a/lib/event_store/sql/statements.ex b/lib/event_store/sql/statements.ex index 33d23927..61288444 100644 --- a/lib/event_store/sql/statements.ex +++ b/lib/event_store/sql/statements.ex @@ -3,18 +3,18 @@ defmodule EventStore.Sql.Statements do require EEx - alias EventStore.Sql.{Config, Init, Reset} + alias EventStore.Sql.{Init, Reset} defdelegate initializers(config), to: Init, as: :statements defdelegate reset(config), to: Reset, as: :statements - partitioned_events = Config.partitioned_events(config) + partitioned_events = Application.get_env(:eventstore, EventStore)[:partitioned_events] for {fun, args} <- [ {:count_streams, [:schema]}, {:create_stream, [:schema]}, {:insert_events, [:schema, :stream_id, :number_of_events, :created_at, :partitioned_events]}, - {:insert_events_any_version, [:schema, :stream_id, :number_of_events, :created_at]}, + {:insert_events_any_version, [:schema, :stream_id, :number_of_events, :created_at, :partitioned_events]}, {:insert_link_events, [:schema, :number_of_events]}, {:soft_delete_stream, [:schema]}, {:hard_delete_stream, [:schema]}, diff --git a/lib/event_store/sql/statements/insert_events_any_version.sql.eex b/lib/event_store/sql/statements/insert_events_any_version.sql.eex index 23207265..17c0febe 100644 --- a/lib/event_store/sql/statements/insert_events_any_version.sql.eex +++ b/lib/event_store/sql/statements/insert_events_any_version.sql.eex @@ -23,6 +23,24 @@ WITH ($<%= i*9+3 %>::uuid, $<%= i*9+10 %>::int, $<%= i*9+11 %>::bigint) <% end %> ), + <%= if @partitioned_events do %> + events_root AS ( + <% + # insert the new events into the events_root table + # using the 7 bind variables from 3 to 9 inclusive + # n.b.: the bind for the event_id is re-generated here + %> + INSERT INTO "<%= schema %>".events_root + ( + event_id + ) + VALUES + <%= for i <- 0..(number_of_events - 1) do %> + <%= unless i == 0 do %>,<% end %> + ($<%= i*9+3 %>) + <% end %> + ), + <% end %> events AS ( INSERT INTO "<%= schema %>".events ( diff --git a/lib/event_store/storage.ex b/lib/event_store/storage.ex index 0f2af5ac..e8290b17 100644 --- a/lib/event_store/storage.ex +++ b/lib/event_store/storage.ex @@ -11,6 +11,8 @@ defmodule EventStore.Storage do Subscription } + partitioned = Application.get_env(:eventstore, EventStore)[:partitioned_events] || false + @doc """ Create a new event stream with the given unique identifier. """ @@ -19,7 +21,7 @@ defmodule EventStore.Storage do @doc """ Append the given list of recorded events to storage. """ - defdelegate append_to_stream(conn, stream_id, events, opts), to: Appender, as: :append + defdelegate append_to_stream(conn, stream_id, events, opts, partitioned), to: Appender, as: :append @doc """ Link the existing event ids already present in a stream to the given stream. diff --git a/lib/event_store/storage/appender.ex b/lib/event_store/storage/appender.ex index 13c2f40e..895ba20c 100644 --- a/lib/event_store/storage/appender.ex +++ b/lib/event_store/storage/appender.ex @@ -13,7 +13,7 @@ defmodule EventStore.Storage.Appender do Returns `:ok` on success, `{:error, reason}` on failure. """ - def append(conn, stream_id, events, opts) do + def append(conn, stream_id, events, opts, partitioned) do [%RecordedEvent{stream_uuid: stream_uuid} | _] = events try do @@ -24,7 +24,7 @@ defmodule EventStore.Storage.Appender do event_count = length(batch) with {:ok, new_stream_id} <- - insert_event_batch(conn, stream_id, stream_uuid, batch, event_count, opts) do + insert_event_batch(conn, stream_id, stream_uuid, batch, event_count, opts, partitioned) do Logger.debug("Appended #{event_count} event(s) to stream #{inspect(stream_uuid)}") new_stream_id else @@ -98,7 +98,7 @@ defmodule EventStore.Storage.Appender do defp encode_uuid(nil), do: nil defp encode_uuid(value), do: UUID.string_to_binary!(value) - defp insert_event_batch(conn, stream_id, stream_uuid, events, event_count, opts) do + defp insert_event_batch(conn, stream_id, stream_uuid, events, event_count, opts, partitioned) do {schema, opts} = Keyword.pop(opts, :schema) {expected_version, opts} = Keyword.pop(opts, :expected_version) {created_at, opts} = Keyword.pop(opts, :created_at_override) @@ -106,10 +106,10 @@ defmodule EventStore.Storage.Appender do statement = case expected_version do :any_version -> - Statements.insert_events_any_version(schema, stream_id, event_count, created_at) + Statements.insert_events_any_version(schema, stream_id, event_count, created_at, partitioned) _expected_version -> - Statements.insert_events(schema, stream_id, event_count, created_at) + Statements.insert_events(schema, stream_id, event_count, created_at, partitioned) end stream_id_or_uuid = stream_id || stream_uuid diff --git a/lib/event_store/streams/stream.ex b/lib/event_store/streams/stream.ex index 39088b62..d01561bf 100644 --- a/lib/event_store/streams/stream.ex +++ b/lib/event_store/streams/stream.ex @@ -4,18 +4,18 @@ defmodule EventStore.Streams.Stream do alias EventStore.{EventData, RecordedEvent, Storage, UUID} alias EventStore.Streams.StreamInfo - def append_to_stream(conn, stream_uuid, expected_version, events, opts) + def append_to_stream(conn, stream_uuid, expected_version, events, opts, partitioned) when length(events) < 1000 do {serializer, new_opts} = Keyword.pop(opts, :serializer) with {:ok, stream} <- stream_info(conn, stream_uuid, expected_version, new_opts), - :ok <- do_append_to_storage(conn, stream, events, expected_version, serializer, new_opts) do + :ok <- do_append_to_storage(conn, stream, events, expected_version, serializer, new_opts, partitioned) do :ok end - |> maybe_retry_once(conn, stream_uuid, expected_version, events, opts) + |> maybe_retry_once(conn, stream_uuid, expected_version, events, opts, partitioned) end - def append_to_stream(conn, stream_uuid, expected_version, events, opts) do + def append_to_stream(conn, stream_uuid, expected_version, events, opts, partitioned) do {serializer, new_opts} = Keyword.pop(opts, :serializer) transaction( @@ -29,7 +29,8 @@ defmodule EventStore.Streams.Stream do events, expected_version, serializer, - new_opts + new_opts, + partitioned ) do :ok else @@ -38,7 +39,7 @@ defmodule EventStore.Streams.Stream do end, new_opts ) - |> maybe_retry_once(conn, stream_uuid, expected_version, events, opts) + |> maybe_retry_once(conn, stream_uuid, expected_version, events, opts, partitioned) end def link_to_stream(conn, stream_uuid, expected_version, events_or_event_ids, opts) do @@ -143,11 +144,12 @@ defmodule EventStore.Streams.Stream do events, expected_version, serializer, - opts + opts, + partitioned ) do prepared_events = prepare_events(events, stream, serializer, opts) - write_to_stream(conn, prepared_events, stream, expected_version, opts) + write_to_stream(conn, prepared_events, stream, expected_version, opts, partitioned) end defp prepare_events(events, %StreamInfo{} = stream, serializer, opts) do @@ -210,12 +212,12 @@ defmodule EventStore.Streams.Stream do raise ArgumentError, message: "Invalid event id, expected a UUID but got: #{inspect(invalid)}" end - defp write_to_stream(conn, prepared_events, %StreamInfo{} = stream, expected_version, opts) do + defp write_to_stream(conn, prepared_events, %StreamInfo{} = stream, expected_version, opts, partitioned) do %StreamInfo{stream_id: stream_id} = stream opts = Keyword.put(opts, :expected_version, expected_version) - Storage.append_to_stream(conn, stream_id, prepared_events, opts) + Storage.append_to_stream(conn, stream_id, prepared_events, opts, partitioned) end defp read_storage_forward(conn, %StreamInfo{} = stream, start_version, count, opts) do @@ -337,18 +339,19 @@ defmodule EventStore.Streams.Stream do stream_uuid, expected_version, events, - opts + opts, + partitioned ) do unless Keyword.has_key?(opts, :retried_once) do opts = Keyword.put(opts, :retried_once, true) - append_to_stream(conn, stream_uuid, expected_version, events, opts) + append_to_stream(conn, stream_uuid, expected_version, events, opts, partitioned) else {:error, {:already_retried_once, :duplicate_stream_uuid}} end end - defp maybe_retry_once(error, _conn, _stream_uuid, _expected_version, _events, _opts), do: error + defp maybe_retry_once(error, _conn, _stream_uuid, _expected_version, _events, _opts, _partitioned), do: error defp transaction(conn, transaction_fun, opts) do case Postgrex.transaction(conn, transaction_fun, opts) do From bd1fae3f1666bd598e72927df7b580ec447f3240 Mon Sep 17 00:00:00 2001 From: Thierry BOMANDOUKI Date: Mon, 26 Jan 2026 15:11:45 +0100 Subject: [PATCH 06/17] Fix all tests --- config/config.exs | 6 +- config/dev.exs | 6 + config/test.exs | 8 +- lib/event_store.ex | 17 +- lib/event_store/sql/init.ex | 361 ++++++++++++------ lib/event_store/sql/reset.ex | 81 +++- lib/event_store/sql/statements.ex | 9 +- .../sql/statements/hard_delete_stream.sql.eex | 6 + .../sql/statements/insert_events.sql.eex | 2 +- .../insert_events_any_version.sql.eex | 6 +- lib/event_store/storage.ex | 4 +- lib/event_store/storage/appender.ex | 16 +- lib/event_store/storage/delete_stream.ex | 3 +- lib/event_store/streams/stream.ex | 34 +- test/multi_event_store_test.exs | 22 +- test/schema_test.exs | 10 +- test/shared_connection_pool_test.exs | 7 +- test/storage/append_events_test.exs | 62 +-- test/storage/link_events_test.exs | 4 +- test/storage/read_events_test.exs | 4 +- test/storage/stream_persistence_test.exs | 4 +- test/streams/all_stream_test.exs | 11 +- test/streams/hard_delete_stream_test.exs | 18 +- test/streams/single_stream_test.exs | 25 +- test/streams/soft_delete_stream_test.exs | 8 +- .../all_streams_subscription_test.exs | 6 +- .../concurrent_subscription_test.exs | 1 + .../linked_event_stream_subscription_test.exs | 6 +- .../monitor_subscription_test.exs | 10 +- .../single_stream_subscription_test.exs | 6 +- .../subscribe_to_stream_test.exs | 127 +++--- .../subscription_acknowledgement_test.exs | 6 +- .../subscription_back_pressure_test.exs | 6 +- .../subscription_catch_up_test.exs | 6 +- .../subscription_recovery_test.exs | 6 +- test/support/subscription_helpers.ex | 6 +- 36 files changed, 628 insertions(+), 292 deletions(-) diff --git a/config/config.exs b/config/config.exs index 8a9c51a0..80c91a0f 100644 --- a/config/config.exs +++ b/config/config.exs @@ -2,6 +2,10 @@ import Config # Configuration globale de EventStore config :eventstore, EventStore, - partitioned_events: true # Default false, set to true if you want a partioned events table + partitioned_events: false, # Default false, set to true if you want a partioned events table + use_pg_partman: false # Default false, set to true if you want to use postgresql extension pg_partman + +config :eventstore, + event_stores: [DevEventStore] import_config "#{Mix.env()}.exs" diff --git a/config/dev.exs b/config/dev.exs index a22c3221..24d7fe5d 100644 --- a/config/dev.exs +++ b/config/dev.exs @@ -4,3 +4,9 @@ import Config config :logger, :console, format: "[$level] $message\n" config :mix_test_watch, clear: true + +config :eventstore, DevEventStore, + schema: "event_store", + column_data_type: "jsonb", + partitioned_events: true, + use_pg_partman: true diff --git a/config/test.exs b/config/test.exs index c920f670..e2446dd9 100644 --- a/config/test.exs +++ b/config/test.exs @@ -13,13 +13,17 @@ default_config = [ password: "postgres", database: "eventstore_test", hostname: "localhost", + schema: "public", pool_size: 1, serializer: EventStore.JsonSerializer, - subscription_retry_interval: 1_000 + subscription_retry_interval: 1_000, + partitioned_events: false, # Default false, set to true if you want a partioned events table + use_pg_partman: false, + column_data_type: "jsonb" ] config :eventstore, TestEventStore, default_config -config :eventstore, SecondEventStore, Keyword.put(default_config, :database, "eventstore_test_2") +config :eventstore, SecondEventStore, Keyword.put(default_config, :database, "thierryb_eventstore_test_2") config :eventstore, SchemaEventStore, default_config config :eventstore, event_stores: [TestEventStore, SecondEventStore, SchemaEventStore] diff --git a/lib/event_store.ex b/lib/event_store.ex index f1f2bcbd..efaf1c12 100644 --- a/lib/event_store.ex +++ b/lib/event_store.ex @@ -118,7 +118,7 @@ defmodule EventStore do Use a dynamic event store by providing its name as an option to each function: - :ok = EventStore.append_to_stream(stream_uuid, expected_version, events, name: :eventstore1, partitioned) + :ok = EventStore.append_to_stream(stream_uuid, expected_version, events, name: :eventstore1, partitioned_events: :partitioned) {:ok, events} = EventStore.read_stream_forward(stream_uuid, 0, 1_000, name: :eventstore1) @@ -183,7 +183,7 @@ defmodule EventStore do {:ok, pid} = Postgrex.start_link(config) Postgrex.transaction(pid, fn conn -> - :ok = EventStore.append_to_stream(stream_uuid, expected_version, events, conn: conn, partitioned) + :ok = EventStore.append_to_stream(stream_uuid, expected_version, events, conn: conn, partitioned_events: :partitioned) end) This can also be used with an Ecto `Repo` which is configured to use the @@ -194,7 +194,7 @@ defmodule EventStore do conn = Process.get({Ecto.Adapters.SQL, pool}) - :ok = EventStore.append_to_stream(stream_uuid, expected_version, events, conn: conn, partitioned) + :ok = EventStore.append_to_stream(stream_uuid, expected_version, events, conn: conn, partitioned_events: :partitioned) end) --- @@ -296,17 +296,17 @@ defmodule EventStore do @accepted_overrides_append_to_stream [:created_at_override] - def append_to_stream(stream_uuid, expected_version, events, opts \\ [], partitioned) + def append_to_stream(stream_uuid, expected_version, events, opts \\ []) - def append_to_stream(@all_stream, _expected_version, _events, _opts, _partitioned), + def append_to_stream(@all_stream, _expected_version, _events, _opts), do: {:error, :cannot_append_to_all_stream} - def append_to_stream(stream_uuid, expected_version, events, opts, partitioned) do + def append_to_stream(stream_uuid, expected_version, events, opts) do overrides = Keyword.take(opts, @accepted_overrides_append_to_stream) {conn, opts} = parse_opts(opts) opts = Keyword.merge(opts, overrides) - Stream.append_to_stream(conn, stream_uuid, expected_version, events, opts, partitioned) + Stream.append_to_stream(conn, stream_uuid, expected_version, events, opts) end def link_to_stream( @@ -658,8 +658,7 @@ defmodule EventStore do stream_uuid :: String.t(), expected_version, events :: list(EventData.t()), - opts :: options, - partitioned :: Boolean + opts :: options ) :: :ok | {:error, :cannot_append_to_all_stream} diff --git a/lib/event_store/sql/init.ex b/lib/event_store/sql/init.ex index 6a37da9c..82e59f64 100644 --- a/lib/event_store/sql/init.ex +++ b/lib/event_store/sql/init.ex @@ -3,53 +3,54 @@ defmodule EventStore.Sql.Init do # PostgreSQL statements to intialize an event store schema. - def create_partitioned_or_not_events_table(partitioned, column_data_type) do + def create_partitioned_or_not_events_table(partitioned, schema, column_data_type) do if partitioned do - create_partitioned_events_table(column_data_type) + create_partitioned_events_table(schema, column_data_type) else - create_events_table(column_data_type) + create_events_table(schema, column_data_type) end end def statements(config) do column_data_type = Keyword.fetch!(config, :column_data_type) - schema = Keyword.fetch!(config, :schema) || 'eventi_store' + schema = Keyword.fetch!(config, :schema) || 'event_store' + database = Keyword.fetch!(config, :database) partitioned = Keyword.fetch!(config, :partitioned_events) || false + partman = Keyword.fetch!(config, :use_pg_partman) || false [ ~s(SET LOCAL search_path TO "#{schema}";), - create_streams_table(), - create_stream_uuid_index(), - create_events_root_table(partitioned), - create_partitioned_or_not_events_table(partitioned, column_data_type), - create_events_index_1(), - create_events_index_2(), - create_events_index_3(column_data_type), - create_stream_events_table(partitioned), - create_stream_events_index(), - create_event_store_exception_function(), - create_event_store_delete_function(), - prevent_streams_delete(), - prevent_event_delete(), - prevent_event_root_delete(partitioned), - prevent_event_update(), - prevent_event_root_update(partitioned), - prevent_stream_events_delete(), - prevent_stream_events_update(), - create_notify_events_function(), - seed_all_stream(), - create_event_notification_trigger(), - create_subscriptions_table(), - create_subscription_index(), - create_snapshots_table(column_data_type), - create_schema_migrations_table(), - record_event_store_schema_version() - ] + create_streams_table(schema), + create_stream_uuid_index(schema), + create_events_root_table(partitioned, schema), + create_partitioned_or_not_events_table(partitioned, schema, column_data_type) + ] ++ create_events_indexes(schema, column_data_type) ++ + [ + create_stream_events_table(partitioned, schema), + create_stream_events_index(schema), + create_event_store_exception_function(schema), + create_event_store_delete_function(schema), + prevent_streams_delete(schema), + prevent_event_delete(schema), + prevent_event_root_delete(partitioned, schema), + prevent_event_update(schema), + prevent_event_root_update(partitioned, schema), + prevent_stream_events_delete(schema), + prevent_stream_events_update(schema), + create_notify_events_function(schema), + seed_all_stream(schema), + create_event_notification_trigger(schema), + create_subscriptions_table(schema), + create_subscription_index(schema), + create_snapshots_table(schema, column_data_type), + create_schema_migrations_table(schema), + record_event_store_schema_version(schema) + ] ++ create_events_partitions(partitioned, database, schema, column_data_type, partman) end - defp create_streams_table do + defp create_streams_table(schema) do """ - CREATE TABLE streams + CREATE TABLE #{schema}.streams ( stream_id bigserial PRIMARY KEY NOT NULL, stream_uuid text NOT NULL, @@ -60,37 +61,37 @@ defmodule EventStore.Sql.Init do """ end - defp create_stream_uuid_index do + defp create_stream_uuid_index(schema) do """ - CREATE UNIQUE INDEX ix_streams_stream_uuid ON streams (stream_uuid); + CREATE UNIQUE INDEX ix_streams_stream_uuid ON #{schema}.streams (stream_uuid); """ end # Create `$all` stream - defp seed_all_stream do + defp seed_all_stream(schema) do """ - INSERT INTO streams (stream_id, stream_uuid, stream_version) VALUES (0, '$all', 0); + INSERT INTO #{schema}.streams (stream_id, stream_uuid, stream_version) VALUES (0, '$all', 0); """ end # Create `events_root` table - defp create_events_root_table(partitioned) do + defp create_events_root_table(partitioned, schema) do if partitioned do """ - CREATE TABLE events_root + CREATE TABLE IF NOT EXISTS #{schema}.events_root ( event_id uuid PRIMARY KEY NOT NULL ); """ else - "" + "SELECT 1;" end end # Create partitioned `events` parent table - defp create_partitioned_events_table(column_data_type) do + defp create_partitioned_events_table(schema, column_data_type) do """ - CREATE TABLE events ( + CREATE TABLE IF NOT EXISTS #{schema}.events ( event_id UUID NOT NULL, event_type TEXT NOT NULL, causation_id UUID NULL, @@ -101,15 +102,161 @@ defmodule EventStore.Sql.Init do CONSTRAINT event_store_events_pkey PRIMARY KEY (event_id, created_at), CONSTRAINT event_store_events_root_fk FOREIGN KEY (event_id) - REFERENCES events_root (event_id) + REFERENCES #{schema}.events_root (event_id) ) PARTITION BY RANGE (created_at); """ end + # Create partitioned `events` children tables + defp create_events_partitions(_partitioned=true, database, schema, _column_data_type, _partman=true) do + today = Date.utc_today() + year = to_string(today.year) + month = String.pad_leading(to_string(today.month), 2, "0") + [ + "CREATE SCHEMA IF NOT EXISTS partman;", + "CREATE EXTENSION IF NOT EXISTS pg_partman SCHEMA partman;" + ] ++ create_role(schema, 'partman_user') ++ + [ + "GRANT ALL ON SCHEMA partman TO partman_user;", + "GRANT ALL ON ALL TABLES IN SCHEMA partman TO partman_user;", + "GRANT EXECUTE ON ALL FUNCTIONS IN SCHEMA partman TO partman_user;", + "GRANT EXECUTE ON ALL PROCEDURES IN SCHEMA partman TO partman_user;", + "GRANT ALL ON SCHEMA #{schema} TO partman_user;", + "GRANT TEMPORARY ON DATABASE #{database} to partman_user;", + """ + CREATE OR REPLACE FUNCTION #{schema}.create_events_partitions( + schema_name TEXT, + year TEXT, + month TEXT + ) + RETURNS VOID AS $$ + DECLARE + r RECORD; + start_date TEXT; + found BOOLEAN; + BEGIN + start_date := year || '-' || month || '-01'; + SELECT EXISTS( + SELECT + child.relname AS partition_name + FROM pg_inherits + JOIN pg_class parent ON pg_inherits.inhparent = parent.oid + JOIN pg_class child ON pg_inherits.inhrelid = child.oid + JOIN pg_namespace nsp ON parent.relnamespace = nsp.oid + WHERE parent.relname = 'events' + AND nsp.nspname = schema_name + ) INTO found; + IF NOT found THEN + IF to_regclass('partman.part_config') IS NOT NULL THEN + DELETE FROM partman.part_config WHERE parent_table = '#{schema}.events'; + END IF; + EXECUTE format(' + SELECT partman.create_parent( + p_parent_table := ''%I.events'', + p_control := ''created_at'', + p_interval := ''1 month'', + p_start_partition := ''%I'' + )', schema_name, start_date); + END IF; + END;$$ + LANGUAGE plpgsql; + """, + "SELECT #{schema}.create_events_partitions('#{schema}', '#{year}', '#{month}');" + ] + end + + defp create_events_partitions(_partitioned=true, database, schema, column_data_type, _partman=false) do + years_months = 0..6 |> Enum.map( fn x -> + today = Date.utc_today() + + total_months = + today.year * 12 + + (today.month - 1) + x + + year = div(total_months, 12) + month = rem(total_months, 12) + 1 + + {to_string(year), String.pad_leading(to_string(month), 2, "0")} + end) + create_events_partition(years_months, [], database, schema, column_data_type) ++ + [ + """ + CREATE TABLE #{schema}.events_default PARTITION OF #{schema}.events DEFAULT; + """, + """ + CREATE INDEX IF NOT EXISTS events_default_created_at_idx ON #{schema}.events_default USING btree (created_at); + """, + """ + CREATE INDEX IF NOT EXISTS events_default_event_type_created_at_idx ON #{schema}.events_default USING btree (event_type, created_at) + """ + ] + end + + defp create_events_partitions(false, _, _, _, _) do + [] + end + + def create_events_partition( [{_, _}], acc, _, _, _) do + acc + end + + def create_events_partition( + [{start_year, start_month} | list_years_months], + acc, + database, + schema, + column_data_type) do + partition_name = start_year <> start_month <> "01" + {end_year, end_month} = hd(list_years_months) + partition = [ + """ + CREATE TABLE #{schema}.events_p#{partition_name} PARTITION OF #{schema}.events + FOR VALUES FROM ('#{start_year}-#{start_month}-01 00:00:00+01') TO ('#{end_year}-#{end_month}-01 00:00:00+01'); + """, + """ + CREATE INDEX IF NOT EXISTS events_#{partition_name}_created_at_idx ON #{schema}.events_p#{partition_name} USING btree (created_at); + """ + ] ++ + if String.downcase(column_data_type) == 'jsonb' do + [ + """ + CREATE INDEX IF NOT EXISTS events_p#{partition_name}_data_idx ON #{schema}.events_p#{partition_name} USING gin (data jsonb_path_ops); + """ + ] + else + ["SELECT 1;"] + end ++ + [ + """ + CREATE INDEX IF NOT EXISTS events_p#{partition_name}_event_type_created_at_idx + ON #{schema}.events_p#{partition_name} USING btree (event_type, created_at); + """ + ] + create_events_partition(list_years_months, acc ++ partition, database, schema, column_data_type) + end + + # Create role + defp create_role(schema, name) do + [ + """ + CREATE OR REPLACE FUNCTION #{schema}.create_role_if_not_exists(role_name TEXT) + RETURNS VOID AS $$ + BEGIN + -- Check if role already exists + IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = role_name) THEN + EXECUTE format('CREATE ROLE %I WITH LOGIN', role_name); + END IF; + END;$$ + LANGUAGE plpgsql; + """, + "SELECT #{schema}.create_role_if_not_exists('#{name}');" + ] + end + # Create `events` table - defp create_events_table(column_data_type) do + defp create_events_table(schema, column_data_type) do """ - CREATE TABLE events + CREATE TABLE #{schema}.events ( event_id uuid PRIMARY KEY NOT NULL, event_type text NOT NULL, @@ -123,32 +270,32 @@ defmodule EventStore.Sql.Init do end # Create `events` indexes - defp create_events_index_1 do - """ - CREATE INDEX event_store_events_created_at_idx ON events(created_at); - """ - end - - defp create_events_index_2 do - """ - CREATE INDEX event_store_events_event_type_idx ON events(event_type, created_at); - """ - end - - defp create_events_index_3(column_data_type) do - # Adding an index if data is a jsonb - if String.downcase(column_data_type) == "jsonb" do + defp create_events_indexes(schema, column_data_type) do + indexes_queries = + [ """ - CREATE INDEX ON events USING GIN ("data" jsonb_path_ops); + CREATE INDEX IF NOT EXISTS event_store_events_created_at_idx ON #{schema}.events(created_at); + """, """ + CREATE INDEX IF NOT EXISTS event_store_events_event_type_idx ON #{schema}.events(event_type, created_at); + """ + ] + # Adding an index if data is a jsonb + if String.downcase(column_data_type) == "jsonb" do + indexes_queries ++ + [ + """ + CREATE INDEX IF NOT EXISTS event_store_events_data_idx ON #{schema}.events USING GIN ("data" jsonb_path_ops); + """ + ] else - "SELECT 1;" + indexes_queries end end - defp create_event_store_exception_function do + defp create_event_store_exception_function(schema) do """ - CREATE OR REPLACE FUNCTION event_store_exception() + CREATE OR REPLACE FUNCTION #{schema}.event_store_exception() RETURNS trigger AS $$ DECLARE message text; @@ -162,9 +309,9 @@ defmodule EventStore.Sql.Init do end # Prevent DELETE operations unless hard deletes have been enabled. - defp create_event_store_delete_function do + defp create_event_store_delete_function(schema) do """ - CREATE OR REPLACE FUNCTION event_store_delete() + CREATE OR REPLACE FUNCTION #{schema}.event_store_delete() RETURNS trigger AS $$ DECLARE message text; @@ -186,23 +333,23 @@ defmodule EventStore.Sql.Init do end # prevent updates to `events` table - defp prevent_event_update() do + defp prevent_event_update(schema) do """ CREATE TRIGGER no_update_events - BEFORE UPDATE ON events + BEFORE UPDATE ON #{schema}.events FOR EACH STATEMENT - EXECUTE PROCEDURE event_store_exception('Cannot update events'); + EXECUTE PROCEDURE #{schema}.event_store_exception('Cannot update events'); """ end # prevent updates to `events_root` table - defp prevent_event_root_update(partitioned) do + defp prevent_event_root_update(partitioned, schema) do if partitioned do """ CREATE TRIGGER no_update_events_root - BEFORE UPDATE ON events_root + BEFORE UPDATE ON #{schema}.events_root FOR EACH STATEMENT - EXECUTE PROCEDURE event_store_exception('Cannot update events_root'); + EXECUTE PROCEDURE #{schema}.event_store_exception('Cannot update events_root'); """ else "SELECT 1;" @@ -210,36 +357,36 @@ defmodule EventStore.Sql.Init do end # prevent deletion from `events` table - defp prevent_event_delete() do + defp prevent_event_delete(schema) do """ CREATE TRIGGER no_delete_events - BEFORE DELETE ON events + BEFORE DELETE ON #{schema}.events FOR EACH STATEMENT - EXECUTE PROCEDURE event_store_delete('Cannot delete events'); + EXECUTE PROCEDURE #{schema}.event_store_delete('Cannot delete events'); """ end # prevent deletion from `events_root` table - defp prevent_event_root_delete(partitioned) do + defp prevent_event_root_delete(partitioned, schema) do if partitioned do """ CREATE TRIGGER no_delete_events_root - BEFORE DELETE ON events_root + BEFORE DELETE ON #{schema}.events_root FOR EACH STATEMENT - EXECUTE PROCEDURE event_store_delete('Cannot delete events_root'); + EXECUTE PROCEDURE #{schema}.event_store_delete('Cannot delete events_root'); """ else "SELECT 1;" end end - defp create_stream_events_table(partitioned) do + defp create_stream_events_table(partitioned, schema) do events_table = if partitioned do - "events_root" + "#{schema}.events_root" else - "events" + "#{schema}.events" end """ CREATE TABLE stream_events @@ -254,44 +401,44 @@ defmodule EventStore.Sql.Init do """ end - defp create_stream_events_index do + defp create_stream_events_index(schema) do """ - CREATE UNIQUE INDEX ix_stream_events ON stream_events (stream_id, stream_version); + CREATE UNIQUE INDEX ix_stream_events ON #{schema}.stream_events (stream_id, stream_version); """ end # prevent updates to `stream_events` table - defp prevent_stream_events_update do + defp prevent_stream_events_update(schema) do """ CREATE TRIGGER no_update_stream_events - BEFORE UPDATE ON stream_events + BEFORE UPDATE ON #{schema}.stream_events FOR EACH STATEMENT - EXECUTE PROCEDURE event_store_exception('Cannot update stream events'); + EXECUTE PROCEDURE #{schema}.event_store_exception('Cannot update stream events'); """ end # prevent deletion from `stream_events` table - def prevent_stream_events_delete do + def prevent_stream_events_delete(schema) do """ CREATE TRIGGER no_delete_stream_events - BEFORE DELETE ON stream_events + BEFORE DELETE ON #{schema}.stream_events FOR EACH STATEMENT - EXECUTE PROCEDURE event_store_delete('Cannot delete stream events'); + EXECUTE PROCEDURE #{schema}.event_store_delete('Cannot delete stream events'); """ end - def prevent_streams_delete do + def prevent_streams_delete(schema) do """ CREATE TRIGGER no_delete_streams - BEFORE DELETE ON streams + BEFORE DELETE ON #{schema}.streams FOR EACH STATEMENT - EXECUTE PROCEDURE event_store_delete('Cannot delete streams'); + EXECUTE PROCEDURE #{schema}.event_store_delete('Cannot delete streams'); """ end - defp create_notify_events_function do + defp create_notify_events_function(schema) do """ - CREATE OR REPLACE FUNCTION notify_events() + CREATE OR REPLACE FUNCTION #{schema}.notify_events() RETURNS trigger AS $$ DECLARE old_stream_version bigint; @@ -323,17 +470,17 @@ defmodule EventStore.Sql.Init do """ end - defp create_event_notification_trigger do + defp create_event_notification_trigger(schema) do """ CREATE TRIGGER event_notification - AFTER INSERT OR UPDATE ON streams - FOR EACH ROW EXECUTE PROCEDURE notify_events(); + AFTER INSERT OR UPDATE ON #{schema}.streams + FOR EACH ROW EXECUTE PROCEDURE #{schema}.notify_events(); """ end - defp create_subscriptions_table do + defp create_subscriptions_table(schema) do """ - CREATE TABLE subscriptions + CREATE TABLE #{schema}.subscriptions ( subscription_id bigserial PRIMARY KEY NOT NULL, stream_uuid text NOT NULL, @@ -344,15 +491,15 @@ defmodule EventStore.Sql.Init do """ end - defp create_subscription_index do + defp create_subscription_index(schema) do """ - CREATE UNIQUE INDEX ix_subscriptions_stream_uuid_subscription_name ON subscriptions (stream_uuid, subscription_name); + CREATE UNIQUE INDEX ix_subscriptions_stream_uuid_subscription_name ON #{schema}.subscriptions (stream_uuid, subscription_name); """ end - defp create_snapshots_table(column_data_type) do + defp create_snapshots_table(schema, column_data_type) do """ - CREATE TABLE snapshots + CREATE TABLE #{schema}.snapshots ( source_uuid text PRIMARY KEY NOT NULL, source_version bigint NOT NULL, @@ -365,9 +512,9 @@ defmodule EventStore.Sql.Init do end # record execution of upgrade scripts - defp create_schema_migrations_table do + defp create_schema_migrations_table(schema) do """ - CREATE TABLE schema_migrations + CREATE TABLE #{schema}.schema_migrations ( major_version int NOT NULL, minor_version int NOT NULL, @@ -379,9 +526,9 @@ defmodule EventStore.Sql.Init do end # record current event store schema version - defp record_event_store_schema_version do + defp record_event_store_schema_version(schema) do """ - INSERT INTO schema_migrations (major_version, minor_version, patch_version) + INSERT INTO #{schema}.schema_migrations (major_version, minor_version, patch_version) VALUES (1, 3, 2); """ end diff --git a/lib/event_store/sql/reset.ex b/lib/event_store/sql/reset.ex index 393b0b7e..37577fe7 100644 --- a/lib/event_store/sql/reset.ex +++ b/lib/event_store/sql/reset.ex @@ -5,26 +5,87 @@ defmodule EventStore.Sql.Reset do def statements(config) do schema = Keyword.fetch!(config, :schema) + partitioned = Keyword.fetch!(config, :partitioned_events) || false + partman = Keyword.fetch!(config, :use_pg_partman) || false [ ~s(SET LOCAL search_path TO "#{schema}";), - ~s(SET LOCAL eventstore.reset TO 'on';), - truncate_tables(), - seed_all_stream() + ~s(SET LOCAL eventstore.reset TO 'on';) + ] + ++ + if partitioned and partman do + undo_partman_partitions(schema) + else + [] + end + ++ truncate_tables(partitioned, schema) ++ + [ + seed_all_stream(schema) ] end - defp truncate_tables do - """ - TRUNCATE TABLE snapshots, subscriptions, stream_events, streams, events - RESTART IDENTITY; - """ + defp undo_partman_partitions(_schema) do + [] + end + + defp _undo_partman_partitions(schema) do + [ + """ + CREATE OR REPLACE FUNCTION #{schema}.undo_events_partitions_if_exist( + schema_name TEXT + ) + RETURNS VOID AS $$ + DECLARE + r RECORD; + dropped BOOLEAN; + BEGIN + dropped := FALSE; + -- Loop on existing partitions to undo them + FOR r IN + SELECT + child.relname AS partition_name + FROM pg_inherits + JOIN pg_class parent ON pg_inherits.inhparent = parent.oid + JOIN pg_class child ON pg_inherits.inhrelid = child.oid + JOIN pg_namespace nsp ON parent.relnamespace = nsp.oid + WHERE parent.relname = 'events' + AND nsp.nspname = schema_name + LOOP + EXECUTE format(' + DROP TABLE %s.%s CASCADE; + ', schema_name, r.partition_name); + dropped := TRUE; + END LOOP; + IF dropped THEN + EXECUTE format(' + DELETE FROM partman.part_config WHERE parent_table = ''%s.events''; + ', schema_name); + END IF; + END;$$ + LANGUAGE plpgsql; + """, + "SELECT #{schema}.undo_events_partitions_if_exist('#{schema}');" + ] + end + + defp truncate_tables(partitioned, schema) do + events_root = if partitioned do + ", #{schema}.events_root" + else + "" + end + [ + """ + TRUNCATE TABLE #{schema}.snapshots, #{schema}.subscriptions, #{schema}.stream_events, #{schema}.streams, #{schema}.events#{events_root} + RESTART IDENTITY; + """ + ] end # Create `$all` stream - defp seed_all_stream do + defp seed_all_stream(schema) do """ - INSERT INTO streams (stream_id, stream_uuid, stream_version) VALUES (0, '$all', 0); + INSERT INTO #{schema}.streams (stream_id, stream_uuid, stream_version) VALUES (0, '$all', 0); """ end end diff --git a/lib/event_store/sql/statements.ex b/lib/event_store/sql/statements.ex index 61288444..5bb68d5e 100644 --- a/lib/event_store/sql/statements.ex +++ b/lib/event_store/sql/statements.ex @@ -8,16 +8,16 @@ defmodule EventStore.Sql.Statements do defdelegate initializers(config), to: Init, as: :statements defdelegate reset(config), to: Reset, as: :statements - partitioned_events = Application.get_env(:eventstore, EventStore)[:partitioned_events] + partitioned = Application.get_env(:eventstore, EventStore)[:partitioned_events] for {fun, args} <- [ {:count_streams, [:schema]}, {:create_stream, [:schema]}, - {:insert_events, [:schema, :stream_id, :number_of_events, :created_at, :partitioned_events]}, - {:insert_events_any_version, [:schema, :stream_id, :number_of_events, :created_at, :partitioned_events]}, + {:insert_events, [:schema, :stream_id, :number_of_events, :created_at, :partitioned]}, + {:insert_events_any_version, [:schema, :stream_id, :number_of_events, :created_at, :partitioned]}, {:insert_link_events, [:schema, :number_of_events]}, {:soft_delete_stream, [:schema]}, - {:hard_delete_stream, [:schema]}, + {:hard_delete_stream, [:schema, :partitioned]}, {:insert_subscription, [:schema]}, {:delete_subscription, [:schema]}, {:try_advisory_lock, [:schema]}, @@ -37,6 +37,7 @@ defmodule EventStore.Sql.Statements do @external_resource file + #EEx.function_from_file(:def, fun, file, args ++ [:partitioned], engine: EventStore.EExIOListEngine) EEx.function_from_file(:def, fun, file, args, engine: EventStore.EExIOListEngine) end end diff --git a/lib/event_store/sql/statements/hard_delete_stream.sql.eex b/lib/event_store/sql/statements/hard_delete_stream.sql.eex index 22610b80..4db6acc9 100644 --- a/lib/event_store/sql/statements/hard_delete_stream.sql.eex +++ b/lib/event_store/sql/statements/hard_delete_stream.sql.eex @@ -7,6 +7,12 @@ linked_events AS ( DELETE FROM "<%= schema %>".stream_events WHERE event_id IN (SELECT event_id FROM deleted_stream_events) ), +<%= if partitioned do %> +events_root AS ( + DELETE FROM "<%= schema %>".events_root + WHERE event_id IN (SELECT event_id FROM deleted_stream_events) +), +<% end %> events AS ( DELETE FROM "<%= schema %>".events WHERE event_id IN (SELECT event_id FROM deleted_stream_events) diff --git a/lib/event_store/sql/statements/insert_events.sql.eex b/lib/event_store/sql/statements/insert_events.sql.eex index a88c70f7..d36ca13b 100644 --- a/lib/event_store/sql/statements/insert_events.sql.eex +++ b/lib/event_store/sql/statements/insert_events.sql.eex @@ -34,10 +34,10 @@ WITH ($<%= i*9+3 %>::uuid, $<%= i*9+10 %>::int, $<%= i*9+11 %>::bigint) <% end %> ), + <%= if partitioned do %> <% # if events table is a partioned table then we need to begin by inserting in events_root table %> - <%= if @partitioned_events do %> events_root AS ( <% # insert the new events into the events_root table diff --git a/lib/event_store/sql/statements/insert_events_any_version.sql.eex b/lib/event_store/sql/statements/insert_events_any_version.sql.eex index 17c0febe..03cc9524 100644 --- a/lib/event_store/sql/statements/insert_events_any_version.sql.eex +++ b/lib/event_store/sql/statements/insert_events_any_version.sql.eex @@ -1,5 +1,5 @@ WITH - stream AS ( +stream AS ( <%= cond do %> <% stream_id -> %> UPDATE "<%= schema %>".streams @@ -23,8 +23,8 @@ WITH ($<%= i*9+3 %>::uuid, $<%= i*9+10 %>::int, $<%= i*9+11 %>::bigint) <% end %> ), - <%= if @partitioned_events do %> - events_root AS ( + <%= if partitioned do %> + events_root AS ( <% # insert the new events into the events_root table # using the 7 bind variables from 3 to 9 inclusive diff --git a/lib/event_store/storage.ex b/lib/event_store/storage.ex index e8290b17..0f2af5ac 100644 --- a/lib/event_store/storage.ex +++ b/lib/event_store/storage.ex @@ -11,8 +11,6 @@ defmodule EventStore.Storage do Subscription } - partitioned = Application.get_env(:eventstore, EventStore)[:partitioned_events] || false - @doc """ Create a new event stream with the given unique identifier. """ @@ -21,7 +19,7 @@ defmodule EventStore.Storage do @doc """ Append the given list of recorded events to storage. """ - defdelegate append_to_stream(conn, stream_id, events, opts, partitioned), to: Appender, as: :append + defdelegate append_to_stream(conn, stream_id, events, opts), to: Appender, as: :append @doc """ Link the existing event ids already present in a stream to the given stream. diff --git a/lib/event_store/storage/appender.ex b/lib/event_store/storage/appender.ex index 895ba20c..8930476f 100644 --- a/lib/event_store/storage/appender.ex +++ b/lib/event_store/storage/appender.ex @@ -13,7 +13,7 @@ defmodule EventStore.Storage.Appender do Returns `:ok` on success, `{:error, reason}` on failure. """ - def append(conn, stream_id, events, opts, partitioned) do + def append(conn, stream_id, events, opts) do [%RecordedEvent{stream_uuid: stream_uuid} | _] = events try do @@ -24,7 +24,7 @@ defmodule EventStore.Storage.Appender do event_count = length(batch) with {:ok, new_stream_id} <- - insert_event_batch(conn, stream_id, stream_uuid, batch, event_count, opts, partitioned) do + insert_event_batch(conn, stream_id, stream_uuid, batch, event_count, opts) do Logger.debug("Appended #{event_count} event(s) to stream #{inspect(stream_uuid)}") new_stream_id else @@ -98,10 +98,12 @@ defmodule EventStore.Storage.Appender do defp encode_uuid(nil), do: nil defp encode_uuid(value), do: UUID.string_to_binary!(value) - defp insert_event_batch(conn, stream_id, stream_uuid, events, event_count, opts, partitioned) do + defp insert_event_batch(conn, stream_id, stream_uuid, events, event_count, opts) do {schema, opts} = Keyword.pop(opts, :schema) {expected_version, opts} = Keyword.pop(opts, :expected_version) {created_at, opts} = Keyword.pop(opts, :created_at_override) + partitioned = Keyword.get(opts, :partitioned_events, false) + {debug, opts} = Keyword.pop(opts, :debug) statement = case expected_version do @@ -112,6 +114,10 @@ defmodule EventStore.Storage.Appender do Statements.insert_events(schema, stream_id, event_count, created_at, partitioned) end + if debug do + IO.puts("Statement : #{statement}") + end + stream_id_or_uuid = stream_id || stream_uuid params = @@ -119,6 +125,10 @@ defmodule EventStore.Storage.Appender do |> Enum.concat(build_insert_parameters(events)) |> append_if(!stream_id, created_at) + if debug do + IO.inspect(params) + end + case Postgrex.query(conn, statement, params, opts) do {:ok, %Postgrex.Result{num_rows: 0}} -> {:error, :not_found} diff --git a/lib/event_store/storage/delete_stream.ex b/lib/event_store/storage/delete_stream.ex index ef70ac2c..d9a27e19 100644 --- a/lib/event_store/storage/delete_stream.ex +++ b/lib/event_store/storage/delete_stream.ex @@ -34,8 +34,9 @@ defmodule EventStore.Storage.DeleteStream do def hard_delete(conn, stream_id, opts) do {schema, opts} = Keyword.pop(opts, :schema) + partitioned = Keyword.get(opts, :partitioned_events, false) - query = Statements.hard_delete_stream(schema) + query = Statements.hard_delete_stream(schema, partitioned) case Postgrex.query(conn, query, [stream_id], opts) do {:ok, %Postgrex.Result{num_rows: 1, rows: [[^stream_id]]}} -> diff --git a/lib/event_store/streams/stream.ex b/lib/event_store/streams/stream.ex index d01561bf..499bf9be 100644 --- a/lib/event_store/streams/stream.ex +++ b/lib/event_store/streams/stream.ex @@ -4,18 +4,21 @@ defmodule EventStore.Streams.Stream do alias EventStore.{EventData, RecordedEvent, Storage, UUID} alias EventStore.Streams.StreamInfo - def append_to_stream(conn, stream_uuid, expected_version, events, opts, partitioned) + def append_to_stream(conn, stream_uuid, expected_version, events, opts) when length(events) < 1000 do + #IO.inspect(events) {serializer, new_opts} = Keyword.pop(opts, :serializer) + + #IO.inspect(stream_info(conn, stream_uuid, expected_version, new_opts)) with {:ok, stream} <- stream_info(conn, stream_uuid, expected_version, new_opts), - :ok <- do_append_to_storage(conn, stream, events, expected_version, serializer, new_opts, partitioned) do + :ok <- do_append_to_storage(conn, stream, events, expected_version, serializer, new_opts) do :ok end - |> maybe_retry_once(conn, stream_uuid, expected_version, events, opts, partitioned) + |> maybe_retry_once(conn, stream_uuid, expected_version, events, opts) end - def append_to_stream(conn, stream_uuid, expected_version, events, opts, partitioned) do + def append_to_stream(conn, stream_uuid, expected_version, events, opts) do {serializer, new_opts} = Keyword.pop(opts, :serializer) transaction( @@ -29,8 +32,7 @@ defmodule EventStore.Streams.Stream do events, expected_version, serializer, - new_opts, - partitioned + new_opts ) do :ok else @@ -39,7 +41,7 @@ defmodule EventStore.Streams.Stream do end, new_opts ) - |> maybe_retry_once(conn, stream_uuid, expected_version, events, opts, partitioned) + |> maybe_retry_once(conn, stream_uuid, expected_version, events, opts) end def link_to_stream(conn, stream_uuid, expected_version, events_or_event_ids, opts) do @@ -144,12 +146,12 @@ defmodule EventStore.Streams.Stream do events, expected_version, serializer, - opts, - partitioned + opts ) do prepared_events = prepare_events(events, stream, serializer, opts) + #IO.inspect(prepared_events) - write_to_stream(conn, prepared_events, stream, expected_version, opts, partitioned) + write_to_stream(conn, prepared_events, stream, expected_version, opts) end defp prepare_events(events, %StreamInfo{} = stream, serializer, opts) do @@ -212,12 +214,13 @@ defmodule EventStore.Streams.Stream do raise ArgumentError, message: "Invalid event id, expected a UUID but got: #{inspect(invalid)}" end - defp write_to_stream(conn, prepared_events, %StreamInfo{} = stream, expected_version, opts, partitioned) do + defp write_to_stream(conn, prepared_events, %StreamInfo{} = stream, expected_version, opts) do %StreamInfo{stream_id: stream_id} = stream opts = Keyword.put(opts, :expected_version, expected_version) + #IO.inspect(opts) - Storage.append_to_stream(conn, stream_id, prepared_events, opts, partitioned) + Storage.append_to_stream(conn, stream_id, prepared_events, opts) end defp read_storage_forward(conn, %StreamInfo{} = stream, start_version, count, opts) do @@ -339,19 +342,18 @@ defmodule EventStore.Streams.Stream do stream_uuid, expected_version, events, - opts, - partitioned + opts ) do unless Keyword.has_key?(opts, :retried_once) do opts = Keyword.put(opts, :retried_once, true) - append_to_stream(conn, stream_uuid, expected_version, events, opts, partitioned) + append_to_stream(conn, stream_uuid, expected_version, events, opts) else {:error, {:already_retried_once, :duplicate_stream_uuid}} end end - defp maybe_retry_once(error, _conn, _stream_uuid, _expected_version, _events, _opts, _partitioned), do: error + defp maybe_retry_once(error, _conn, _stream_uuid, _expected_version, _events, _opts), do: error defp transaction(conn, transaction_fun, opts) do case Postgrex.transaction(conn, transaction_fun, opts) do diff --git a/test/multi_event_store_test.exs b/test/multi_event_store_test.exs index dd04669c..b8dbf4bd 100644 --- a/test/multi_event_store_test.exs +++ b/test/multi_event_store_test.exs @@ -11,12 +11,16 @@ defmodule EventStore.MultiEventStoreTest do :ok end + def partitioned?(evs) do + Application.get_env(:eventstore, evs)[:partitioned_events] || false + end + describe "append to multiple event stores" do test "should append events to single store" do stream_uuid = UUID.uuid4() events = EventFactory.create_events(3) - :ok = TestEventStore.append_to_stream(stream_uuid, 0, events) + :ok = TestEventStore.append_to_stream(stream_uuid, 0, events, partitioned_events: partitioned?(TestEventStore)) assert_read_stream_events(TestEventStore, stream_uuid, events) assert_read_all_stream_events(TestEventStore, events) @@ -28,8 +32,8 @@ defmodule EventStore.MultiEventStoreTest do stream_uuid = UUID.uuid4() events = EventFactory.create_events(3) - :ok = TestEventStore.append_to_stream(stream_uuid, 0, events) - :ok = SecondEventStore.append_to_stream(stream_uuid, 0, events) + :ok = TestEventStore.append_to_stream(stream_uuid, 0, events, partitioned_events: partitioned?(TestEventStore)) + :ok = SecondEventStore.append_to_stream(stream_uuid, 0, events, partitioned_events: partitioned?(SecondEventStore)) assert_read_stream_events(TestEventStore, stream_uuid, events) assert_read_stream_events(SecondEventStore, stream_uuid, events) @@ -46,8 +50,8 @@ defmodule EventStore.MultiEventStoreTest do :ok = TestEventStore.subscribe(stream_uuid) - :ok = TestEventStore.append_to_stream(stream_uuid, 0, events) - :ok = SecondEventStore.append_to_stream(stream_uuid, 0, events) + :ok = TestEventStore.append_to_stream(stream_uuid, 0, events, partitioned_events: partitioned?(TestEventStore)) + :ok = SecondEventStore.append_to_stream(stream_uuid, 0, events, partitioned_events: partitioned?(SecondEventStore)) assert_receive_events(stream_uuid, events) refute_receive {:events, _events} @@ -60,8 +64,8 @@ defmodule EventStore.MultiEventStoreTest do :ok = TestEventStore.subscribe(stream_uuid) :ok = SecondEventStore.subscribe(stream_uuid) - :ok = TestEventStore.append_to_stream(stream_uuid, 0, events) - :ok = SecondEventStore.append_to_stream(stream_uuid, 0, events) + :ok = TestEventStore.append_to_stream(stream_uuid, 0, events, partitioned_events: partitioned?(TestEventStore)) + :ok = SecondEventStore.append_to_stream(stream_uuid, 0, events, partitioned_events: partitioned?(SecondEventStore)) assert_receive_events(stream_uuid, events) assert_receive_events(stream_uuid, events) @@ -83,8 +87,8 @@ defmodule EventStore.MultiEventStoreTest do assert_receive {:subscribed, ^subscription1} assert_receive {:subscribed, ^subscription2} - :ok = TestEventStore.append_to_stream(stream_uuid, 0, events) - :ok = SecondEventStore.append_to_stream(stream_uuid, 0, events) + :ok = TestEventStore.append_to_stream(stream_uuid, 0, events, partitioned_events: partitioned?(TestEventStore)) + :ok = SecondEventStore.append_to_stream(stream_uuid, 0, events, partitioned_events: partitioned?(SecondEventStore)) assert_receive_events(stream_uuid, events) assert_receive_events(stream_uuid, events) diff --git a/test/schema_test.exs b/test/schema_test.exs index 490b45b7..b3444f1b 100644 --- a/test/schema_test.exs +++ b/test/schema_test.exs @@ -4,6 +4,10 @@ defmodule EventStore.SchemaTest do alias EventStore.{Config, EventFactory, UUID} alias EventStore.Storage.Initializer + def partitioned?(evs) do + Application.get_env(:eventstore, evs)[:partitioned_events] || false + end + setup_all do config = SchemaEventStore.config() postgrex_config = Config.default_postgrex_opts(config) @@ -91,8 +95,8 @@ defmodule EventStore.SchemaTest do events = EventFactory.create_events(1) - :ok = SchemaEventStore.append_to_stream(stream_uuid, 0, events) - :ok = TestEventStore.append_to_stream(stream_uuid, 0, events) + :ok = SchemaEventStore.append_to_stream(stream_uuid, 0, events, partitioned_events: partitioned?(SchemaEventStore)) + :ok = TestEventStore.append_to_stream(stream_uuid, 0, events, partitioned_events: partitioned?(TestEventStore)) assert_receive {:events, received_events} assert_events(events, received_events) @@ -114,7 +118,7 @@ defmodule EventStore.SchemaTest do defp do_append_to_stream(stream_uuid, count, expected_version \\ 0) do events = EventFactory.create_events(count, expected_version + 1) - :ok = SchemaEventStore.append_to_stream(stream_uuid, expected_version, events) + :ok = SchemaEventStore.append_to_stream(stream_uuid, expected_version, events, partitioned_events: partitioned?(SchemaEventStore)) {:ok, events} end diff --git a/test/shared_connection_pool_test.exs b/test/shared_connection_pool_test.exs index 30cce230..b7f75f7a 100644 --- a/test/shared_connection_pool_test.exs +++ b/test/shared_connection_pool_test.exs @@ -5,6 +5,10 @@ defmodule EventStore.SharedConnectionPoolTest do alias EventStore.MonitoredServer.State, as: MonitoredServerState alias EventStore.Tasks.{Create, Drop, Init} + def partitioned? do + Application.get_env(:eventstore, TestEventStore)[:partitioned_events] || false + end + describe "connection pool sharing" do setup do for schema <- ["schema1", "schema2"] do @@ -183,7 +187,8 @@ defmodule EventStore.SharedConnectionPoolTest do :ok = TestEventStore.append_to_stream(stream_uuid, expected_version, events, - name: event_store_name + name: event_store_name, + partitioned_events: partitioned?() ) {:ok, events} diff --git a/test/storage/append_events_test.exs b/test/storage/append_events_test.exs index aae7649a..d201cac9 100644 --- a/test/storage/append_events_test.exs +++ b/test/storage/append_events_test.exs @@ -4,18 +4,22 @@ defmodule EventStore.Storage.AppendEventsTest do alias EventStore.{EventFactory, RecordedEvent, UUID} alias EventStore.Storage.{Appender, CreateStream} + def partitioned? do + Application.get_env(:eventstore, EventStore)[:partitioned_events] || false + end + test "append single event to new stream", %{conn: conn, schema: schema} = context do {:ok, stream_uuid, stream_id} = create_stream(context) recorded_events = EventFactory.create_recorded_events(1, stream_uuid) - assert :ok = Appender.append(conn, stream_id, recorded_events, schema: schema) + assert :ok = Appender.append(conn, stream_id, recorded_events, schema: schema, partitioned_events: partitioned?()) end test "append multiple events to new stream", %{conn: conn, schema: schema} = context do {:ok, stream_uuid, stream_id} = create_stream(context) recorded_events = EventFactory.create_recorded_events(3, stream_uuid) - assert :ok = Appender.append(conn, stream_id, recorded_events, schema: schema) + assert :ok = Appender.append(conn, stream_id, recorded_events, schema: schema, partitioned_events: partitioned?()) end test "append single event to existing stream, in separate writes", @@ -25,8 +29,8 @@ defmodule EventStore.Storage.AppendEventsTest do recorded_events1 = EventFactory.create_recorded_events(1, stream_uuid) recorded_events2 = EventFactory.create_recorded_events(1, stream_uuid, 2, 2) - assert :ok = Appender.append(conn, stream_id, recorded_events1, schema: schema) - assert :ok = Appender.append(conn, stream_id, recorded_events2, schema: schema) + assert :ok = Appender.append(conn, stream_id, recorded_events1, schema: schema, partitioned_events: partitioned?()) + assert :ok = Appender.append(conn, stream_id, recorded_events2, schema: schema, partitioned_events: partitioned?()) end test "append multiple events to existing stream, in separate writes", @@ -35,7 +39,8 @@ defmodule EventStore.Storage.AppendEventsTest do assert :ok = Appender.append(conn, stream_id, EventFactory.create_recorded_events(3, stream_uuid), - schema: schema + schema: schema, + partitioned_events: partitioned?() ) assert :ok = @@ -43,7 +48,8 @@ defmodule EventStore.Storage.AppendEventsTest do conn, stream_id, EventFactory.create_recorded_events(3, stream_uuid, 4, 4), - schema: schema + schema: schema, + partitioned_events: partitioned?() ) end @@ -56,7 +62,8 @@ defmodule EventStore.Storage.AppendEventsTest do conn, stream1_id, EventFactory.create_recorded_events(2, stream1_uuid), - schema: schema + schema: schema, + partitioned_events: partitioned?() ) assert :ok = @@ -64,7 +71,8 @@ defmodule EventStore.Storage.AppendEventsTest do conn, stream2_id, EventFactory.create_recorded_events(2, stream2_uuid, 3), - schema: schema + schema: schema, + partitioned_events: partitioned?() ) end @@ -77,7 +85,8 @@ defmodule EventStore.Storage.AppendEventsTest do conn, stream1_id, EventFactory.create_recorded_events(2, stream1_uuid), - schema: schema + schema: schema, + partitioned_events: partitioned?() ) assert :ok = @@ -85,7 +94,8 @@ defmodule EventStore.Storage.AppendEventsTest do conn, stream2_id, EventFactory.create_recorded_events(2, stream2_uuid, 3), - schema: schema + schema: schema, + partitioned_events: partitioned?() ) assert :ok = @@ -93,7 +103,8 @@ defmodule EventStore.Storage.AppendEventsTest do conn, stream1_id, EventFactory.create_recorded_events(2, stream1_uuid, 5, 3), - schema: schema + schema: schema, + partitioned_events: partitioned?() ) assert :ok = @@ -101,7 +112,8 @@ defmodule EventStore.Storage.AppendEventsTest do conn, stream2_id, EventFactory.create_recorded_events(2, stream2_uuid, 7, 3), - schema: schema + schema: schema, + partitioned_events: partitioned?() ) end @@ -110,12 +122,12 @@ defmodule EventStore.Storage.AppendEventsTest do {:ok, stream_uuid, stream_id} = create_stream(context) events = EventFactory.create_recorded_events(1, stream_uuid) - :ok = Appender.append(conn, stream_id, events, schema: schema) + :ok = Appender.append(conn, stream_id, events, schema: schema, partitioned_events: partitioned?()) events = EventFactory.create_recorded_events(1, stream_uuid) assert {:error, :wrong_expected_version} = - Appender.append(conn, stream_id, events, schema: schema) + Appender.append(conn, stream_id, events, schema: schema, partitioned_events: partitioned?()) end test "append to stream that does not exist", %{conn: conn, schema: schema} do @@ -123,7 +135,7 @@ defmodule EventStore.Storage.AppendEventsTest do stream_id = 1 events = EventFactory.create_recorded_events(1, stream_uuid) - assert {:error, :not_found} = Appender.append(conn, stream_id, events, schema: schema) + assert {:error, :not_found} = Appender.append(conn, stream_id, events, schema: schema, partitioned_events: partitioned?()) end test "append to existing stream, but wrong expected version", @@ -131,12 +143,12 @@ defmodule EventStore.Storage.AppendEventsTest do {:ok, stream_uuid, stream_id} = create_stream(context) events = EventFactory.create_recorded_events(2, stream_uuid) - :ok = Appender.append(conn, stream_id, events, schema: schema) + :ok = Appender.append(conn, stream_id, events, schema: schema, partitioned_events: partitioned?()) events = EventFactory.create_recorded_events(2, stream_uuid) assert {:error, :wrong_expected_version} = - Appender.append(conn, stream_id, events, schema: schema) + Appender.append(conn, stream_id, events, schema: schema, partitioned_events: partitioned?()) end test "append events to same stream concurrently", %{conn: conn, schema: schema} = context do @@ -148,7 +160,7 @@ defmodule EventStore.Storage.AppendEventsTest do Task.async(fn -> events = EventFactory.create_recorded_events(10, stream_uuid) - Appender.append(conn, stream_id, events, schema: schema) + Appender.append(conn, stream_id, events, schema: schema, partitioned_events: partitioned?()) end) end) |> Enum.map(&Task.await/1) @@ -168,9 +180,9 @@ defmodule EventStore.Storage.AppendEventsTest do {:ok, stream_uuid, stream_id} = create_stream(context) events = EventFactory.create_recorded_events(3, stream_uuid) - :ok = Appender.append(conn, stream_id, events, schema: schema) + :ok = Appender.append(conn, stream_id, events, schema: schema, partitioned_events: partitioned?()) - {:error, :duplicate_event} = Appender.append(conn, stream_id, events, schema: schema) + {:error, :duplicate_event} = Appender.append(conn, stream_id, events, schema: schema, partitioned_events: partitioned?()) end test "append existing events to the same stream should fail", @@ -178,12 +190,12 @@ defmodule EventStore.Storage.AppendEventsTest do {:ok, stream_uuid, stream_id} = create_stream(context) events = EventFactory.create_recorded_events(3, stream_uuid) - :ok = Appender.append(conn, stream_id, events, schema: schema) + :ok = Appender.append(conn, stream_id, events, schema: schema, partitioned_events: partitioned?()) for event <- events do events = [%RecordedEvent{event | stream_version: 4}] - assert {:error, :duplicate_event} = Appender.append(conn, stream_id, events, schema: schema) + assert {:error, :duplicate_event} = Appender.append(conn, stream_id, events, schema: schema, partitioned_events: partitioned?()) end end @@ -193,7 +205,7 @@ defmodule EventStore.Storage.AppendEventsTest do {:ok, stream2_uuid, stream2_id} = create_stream(context) events = EventFactory.create_recorded_events(3, stream1_uuid) - :ok = Appender.append(conn, stream1_id, events, schema: schema) + :ok = Appender.append(conn, stream1_id, events, schema: schema, partitioned_events: partitioned?()) for event <- events do events = [ @@ -201,7 +213,7 @@ defmodule EventStore.Storage.AppendEventsTest do ] assert {:error, :duplicate_event} = - Appender.append(conn, stream2_id, events, schema: schema) + Appender.append(conn, stream2_id, events, schema: schema, partitioned_events: partitioned?()) end end @@ -218,7 +230,7 @@ defmodule EventStore.Storage.AppendEventsTest do # Using Postgrex query timeout value of zero will cause a `DBConnection.ConnectionError` error # to be returned. assert {:error, %DBConnection.ConnectionError{}} = - Appender.append(conn, 1, recorded_events, schema: schema, timeout: 0) + Appender.append(conn, 1, recorded_events, schema: schema, timeout: 0, partitioned_events: partitioned?()) end defp create_stream(context) do diff --git a/test/storage/link_events_test.exs b/test/storage/link_events_test.exs index 6a1878fa..01e44b3f 100644 --- a/test/storage/link_events_test.exs +++ b/test/storage/link_events_test.exs @@ -110,7 +110,9 @@ defmodule EventStore.Storage.LinkEventsTest do defp append(context, stream_id, recorded_events) do %{conn: conn, schema: schema} = context - Appender.append(conn, stream_id, recorded_events, schema: schema) + partitioned = Application.get_env(:eventstore, EventStore)[:partitioned_events] || false + + Appender.append(conn, stream_id, recorded_events, schema: schema, partitioned_events: partitioned) end defp link(context, stream_id, recorded_events) do diff --git a/test/storage/read_events_test.exs b/test/storage/read_events_test.exs index 4d1d8b02..8ef7b5a6 100644 --- a/test/storage/read_events_test.exs +++ b/test/storage/read_events_test.exs @@ -147,7 +147,9 @@ defmodule EventStore.Storage.ReadEventsTest do defp append(context, stream_id, recorded_events) do %{conn: conn, schema: schema} = context - Appender.append(conn, stream_id, recorded_events, schema: schema) + partitioned = Application.get_env(:eventstore, EventStore)[:partitioned_events] || false + + Appender.append(conn, stream_id, recorded_events, schema: schema, partitioned_events: partitioned) end defp create_stream(context) do diff --git a/test/storage/stream_persistence_test.exs b/test/storage/stream_persistence_test.exs index 751405f5..faab9186 100644 --- a/test/storage/stream_persistence_test.exs +++ b/test/storage/stream_persistence_test.exs @@ -143,13 +143,15 @@ defmodule EventStore.Storage.StreamPersistenceTest do initial_event_number \\ 1 ) do %{conn: conn, schema: schema} = context + + partitioned = Application.get_env(:eventstore, EventStore)[:partitioned_events] || false {:ok, stream_id} = CreateStream.execute(conn, stream_uuid, schema: schema) recorded_events = EventFactory.create_recorded_events(number_of_events, stream_uuid, initial_event_number) - :ok = Appender.append(conn, stream_id, recorded_events, schema: schema) + :ok = Appender.append(conn, stream_id, recorded_events, schema: schema, partitioned_events: partitioned) {:ok, stream_id} end diff --git a/test/streams/all_stream_test.exs b/test/streams/all_stream_test.exs index 7f594d4f..899ac5ac 100644 --- a/test/streams/all_stream_test.exs +++ b/test/streams/all_stream_test.exs @@ -266,11 +266,14 @@ defmodule EventStore.Streams.AllStreamTest do refute_receive {:events, _received_events} events = EventFactory.create_events(1, 4) + + partitioned = Application.get_env(:eventstore, TestEventStore)[:partitioned_events] || false :ok = Stream.append_to_stream(conn, stream1_uuid, 3, events, schema: schema, - serializer: serializer + serializer: serializer, + partitioned_events: partitioned ) assert_receive {:events, received_events} @@ -320,7 +323,11 @@ defmodule EventStore.Streams.AllStreamTest do stream_uuid = UUID.uuid4() events = EventFactory.create_events(3) - :ok = Stream.append_to_stream(conn, stream_uuid, 0, events, opts) + #IO.inspect(opts) + + partitioned = Application.get_env(:eventstore, TestEventStore)[:partitioned_events] || false + + :ok = Stream.append_to_stream(conn, stream_uuid, 0, events, Keyword.put(opts, :partitioned_events, partitioned)) {stream_uuid, events} end diff --git a/test/streams/hard_delete_stream_test.exs b/test/streams/hard_delete_stream_test.exs index a27f80a4..1e2399fd 100644 --- a/test/streams/hard_delete_stream_test.exs +++ b/test/streams/hard_delete_stream_test.exs @@ -4,6 +4,10 @@ defmodule EventStore.Streams.HardDeleteStreamTest do alias EventStore.{EventFactory, ProcessHelper, RecordedEvent, UUID} alias TestEventStore, as: EventStore + def partitioned? do + Application.get_env(:eventstore, EventStore)[:partitioned_events] || false + end + describe "hard delete stream when enabled" do setup [:enable_hard_deletes, :append_events_to_stream] @@ -59,7 +63,7 @@ defmodule EventStore.Streams.HardDeleteStreamTest do events = EventFactory.create_events(1) - assert :ok = EventStore.append_to_stream(stream_uuid, 0, events) + assert :ok = EventStore.append_to_stream(stream_uuid, 0, events, partitioned_events: partitioned?()) assert {:ok, [event]} = EventStore.read_stream_forward(stream_uuid) assert match?(%RecordedEvent{stream_uuid: ^stream_uuid, stream_version: 1}, event) @@ -74,8 +78,8 @@ defmodule EventStore.Streams.HardDeleteStreamTest do stream2_uuid = UUID.uuid4() stream3_uuid = UUID.uuid4() - :ok = EventStore.append_to_stream(stream2_uuid, 0, EventFactory.create_events(2)) - :ok = EventStore.append_to_stream(stream3_uuid, 0, EventFactory.create_events(1)) + :ok = EventStore.append_to_stream(stream2_uuid, 0, EventFactory.create_events(2), partitioned_events: partitioned?()) + :ok = EventStore.append_to_stream(stream3_uuid, 0, EventFactory.create_events(1), partitioned_events: partitioned?()) :ok = EventStore.delete_stream(stream2_uuid, :any_version, :hard) @@ -120,7 +124,7 @@ defmodule EventStore.Streams.HardDeleteStreamTest do stream2_uuid = UUID.uuid4() events = EventFactory.create_events(1) - :ok = EventStore.append_to_stream(stream2_uuid, 0, events) + :ok = EventStore.append_to_stream(stream2_uuid, 0, events, partitioned_events: partitioned?()) {:ok, subscription} = EventStore.subscribe_to_all_streams("test", self(), start_from: :origin) @@ -164,7 +168,7 @@ defmodule EventStore.Streams.HardDeleteStreamTest do stream2_uuid = UUID.uuid4() events = EventFactory.create_events(1) - :ok = EventStore.append_to_stream(stream2_uuid, 0, events) + :ok = EventStore.append_to_stream(stream2_uuid, 0, events, partitioned_events: partitioned?()) assert_receive {:events, [event]} assert match?(%RecordedEvent{stream_uuid: ^stream2_uuid, event_number: 4}, event) @@ -184,7 +188,7 @@ defmodule EventStore.Streams.HardDeleteStreamTest do stream2_uuid = UUID.uuid4() events = EventFactory.create_events(1) - :ok = EventStore.append_to_stream(stream2_uuid, 0, events) + :ok = EventStore.append_to_stream(stream2_uuid, 0, events, partitioned_events: partitioned?()) assert {:ok, [event]} = EventStore.read_all_streams_forward() @@ -233,7 +237,7 @@ defmodule EventStore.Streams.HardDeleteStreamTest do stream_uuid = UUID.uuid4() events = EventFactory.create_events(3) - :ok = EventStore.append_to_stream(stream_uuid, 0, events) + :ok = EventStore.append_to_stream(stream_uuid, 0, events, partitioned_events: partitioned?()) [stream_uuid: stream_uuid, events: events] end diff --git a/test/streams/single_stream_test.exs b/test/streams/single_stream_test.exs index 8789abe9..8a7a52e2 100644 --- a/test/streams/single_stream_test.exs +++ b/test/streams/single_stream_test.exs @@ -7,6 +7,10 @@ defmodule EventStore.Streams.SingleStreamTest do @subscription_name "test_subscription" + def partitioned? do + Application.get_env(:eventstore, EventStore)[:partitioned_events] || false + end + describe "append events to stream" do setup [:append_events_to_stream] @@ -62,7 +66,8 @@ defmodule EventStore.Streams.SingleStreamTest do assert {:error, :wrong_expected_version} = Stream.append_to_stream(conn, stream_uuid, 0, events, schema: schema, - serializer: serializer + serializer: serializer, + partitioned_events: partitioned?() ) end end @@ -194,7 +199,8 @@ defmodule EventStore.Streams.SingleStreamTest do :ok = Stream.append_to_stream(conn, stream_uuid, :any_version, [event], schema: schema, - serializer: serializer + serializer: serializer, + partitioned_events: partitioned?() ) assert_receive {:events, [received_event | _]} @@ -491,7 +497,8 @@ defmodule EventStore.Streams.SingleStreamTest do :ok = Stream.append_to_stream(conn, stream_uuid, 3, events, schema: schema, - serializer: serializer + serializer: serializer, + partitioned_events: partitioned?() ) assert_receive {:events, received_events} @@ -514,7 +521,8 @@ defmodule EventStore.Streams.SingleStreamTest do :ok = Stream.append_to_stream(conn, stream_uuid, 0, events, schema: schema, - serializer: serializer + serializer: serializer, + partitioned_events: partitioned?() ) # stream above needed for preventing accidental event_number/stream_version match @@ -524,7 +532,8 @@ defmodule EventStore.Streams.SingleStreamTest do :ok = Stream.append_to_stream(conn, stream_uuid, 0, events, schema: schema, - serializer: serializer + serializer: serializer, + partitioned_events: partitioned?() ) assert {:ok, 3} = Stream.stream_version(conn, stream_uuid, schema: schema) @@ -539,7 +548,8 @@ defmodule EventStore.Streams.SingleStreamTest do :ok = Stream.append_to_stream(conn, stream_uuid, 0, events, schema: schema, - serializer: serializer + serializer: serializer, + partitioned_events: partitioned?() ) [ @@ -557,7 +567,8 @@ defmodule EventStore.Streams.SingleStreamTest do :ok = Stream.append_to_stream(conn, stream_uuid, 0, events, schema: schema, - serializer: serializer + serializer: serializer, + partitioned_events: partitioned?() ) [ diff --git a/test/streams/soft_delete_stream_test.exs b/test/streams/soft_delete_stream_test.exs index 0c8cc3df..10587b51 100644 --- a/test/streams/soft_delete_stream_test.exs +++ b/test/streams/soft_delete_stream_test.exs @@ -4,6 +4,10 @@ defmodule EventStore.Streams.SoftDeleteStreamTest do alias EventStore.{EventFactory, RecordedEvent, UUID} alias TestEventStore, as: EventStore + def partitioned? do + Application.get_env(:eventstore, EventStore)[:partitioned_events] || false + end + describe "soft delete stream" do setup [:append_events_to_stream] @@ -49,7 +53,7 @@ defmodule EventStore.Streams.SoftDeleteStreamTest do events = EventFactory.create_events(1) - assert {:error, :stream_deleted} = EventStore.append_to_stream(stream_uuid, 3, events) + assert {:error, :stream_deleted} = EventStore.append_to_stream(stream_uuid, 3, events, partitioned_events: partitioned?()) end test "should prevent deleting global `$all` events stream" do @@ -122,7 +126,7 @@ defmodule EventStore.Streams.SoftDeleteStreamTest do stream_uuid = UUID.uuid4() events = EventFactory.create_events(3) - :ok = EventStore.append_to_stream(stream_uuid, 0, events) + :ok = EventStore.append_to_stream(stream_uuid, 0, events, partitioned_events: partitioned?()) [stream_uuid: stream_uuid, events: events] end diff --git a/test/subscriptions/all_streams_subscription_test.exs b/test/subscriptions/all_streams_subscription_test.exs index 7ec19d7c..5a91a59f 100644 --- a/test/subscriptions/all_streams_subscription_test.exs +++ b/test/subscriptions/all_streams_subscription_test.exs @@ -13,13 +13,15 @@ defmodule EventStore.Subscriptions.AllStreamsSubscriptionTest do defp append_events_to_stream(context) do %{conn: conn, schema: schema} = context + + partitioned = Application.get_env(:eventstore, EventStore)[:partitioned_events] || false stream_uuid = UUID.uuid4() recorded_events = EventFactory.create_recorded_events(3, stream_uuid) - {:ok, stream_id} = CreateStream.execute(conn, stream_uuid, schema: schema) + {:ok, stream_id} = CreateStream.execute(conn, stream_uuid, schema: schema, partitioned_events: partitioned) - :ok = Appender.append(conn, stream_id, recorded_events, schema: schema) + :ok = Appender.append(conn, stream_id, recorded_events, schema: schema, partitioned_events: partitioned) [recorded_events: recorded_events] end diff --git a/test/subscriptions/concurrent_subscription_test.exs b/test/subscriptions/concurrent_subscription_test.exs index da48f93b..e13eea49 100644 --- a/test/subscriptions/concurrent_subscription_test.exs +++ b/test/subscriptions/concurrent_subscription_test.exs @@ -875,4 +875,5 @@ defmodule EventStore.Subscriptions.ConcurrentSubscriptionTest do :ok = Subscription.ack(subscription, received_events) end + end diff --git a/test/subscriptions/linked_event_stream_subscription_test.exs b/test/subscriptions/linked_event_stream_subscription_test.exs index 1066b90b..d477a7ba 100644 --- a/test/subscriptions/linked_event_stream_subscription_test.exs +++ b/test/subscriptions/linked_event_stream_subscription_test.exs @@ -4,6 +4,10 @@ defmodule EventStore.Subscriptions.LinkedEventSubscriptionFsmTest do alias EventStore.{EventFactory, ProcessHelper, UUID} alias TestEventStore, as: EventStore + def partitioned? do + Application.get_env(:eventstore, EventStore)[:partitioned_events] || false + end + describe "subscription to linked event stream" do test "should receive linked events" do linked_stream_uuid = UUID.uuid4() @@ -55,7 +59,7 @@ defmodule EventStore.Subscriptions.LinkedEventSubscriptionFsmTest do source_stream_uuid = UUID.uuid4() events = EventFactory.create_events(3) - with :ok <- EventStore.append_to_stream(source_stream_uuid, 0, events), + with :ok <- EventStore.append_to_stream(source_stream_uuid, 0, events, partitioned_events: partitioned?()), {:ok, read_events} <- EventStore.read_stream_forward(source_stream_uuid, 0, 3), :ok <- EventStore.link_to_stream(link_to_stream_uuid, expected_version, read_events) do {:ok, source_stream_uuid, events} diff --git a/test/subscriptions/monitor_subscription_test.exs b/test/subscriptions/monitor_subscription_test.exs index e49488cd..57df5af2 100644 --- a/test/subscriptions/monitor_subscription_test.exs +++ b/test/subscriptions/monitor_subscription_test.exs @@ -6,6 +6,10 @@ defmodule EventStore.Subscriptions.MonitorSubscriptionTest do @event_store TestEventStore + def partitioned? do + Application.get_env(:eventstore, EventStore)[:partitioned_events] || false + end + describe "monitor subscription" do test "should shutdown all stream subscription on subscriber shutdown" do subscription_name = UUID.uuid4() @@ -29,7 +33,7 @@ defmodule EventStore.Subscriptions.MonitorSubscriptionTest do assert Process.alive?(subscriber2) # Appending events to stream should notify subscription 2 - :ok = EventStore.append_to_stream(stream_uuid, 0, events) + :ok = EventStore.append_to_stream(stream_uuid, 0, events, partitioned_events: partitioned?()) # Subscription 2 should still receive events assert_receive {:events, received_events} @@ -66,7 +70,7 @@ defmodule EventStore.Subscriptions.MonitorSubscriptionTest do assert Process.alive?(subscriber2) # Should still notify subscription 2 - :ok = EventStore.append_to_stream(stream_uuid, 0, events) + :ok = EventStore.append_to_stream(stream_uuid, 0, events, partitioned_events: partitioned?()) # Subscription 2 should still receive events assert_receive {:events, received_events} @@ -85,7 +89,7 @@ defmodule EventStore.Subscriptions.MonitorSubscriptionTest do :ok = Subscriptions.unsubscribe_from_stream(@event_store, stream_uuid, subscription_name) - :ok = EventStore.append_to_stream(stream_uuid, 0, events) + :ok = EventStore.append_to_stream(stream_uuid, 0, events, partitioned_events: partitioned?()) refute_receive {:events, _received_events} refute Process.alive?(subscription) diff --git a/test/subscriptions/single_stream_subscription_test.exs b/test/subscriptions/single_stream_subscription_test.exs index 5bd87504..5f4f7d1e 100644 --- a/test/subscriptions/single_stream_subscription_test.exs +++ b/test/subscriptions/single_stream_subscription_test.exs @@ -13,11 +13,15 @@ defmodule EventStore.Subscriptions.SingleSubscriptionFsmTest do defp append_events_to_stream(context) do %{conn: conn, schema: schema, stream_uuid: stream_uuid} = context + partitioned = Application.get_env(:eventstore, TestEventStore)[:partitioned_events] || false + recorded_events = EventFactory.create_recorded_events(3, stream_uuid) {:ok, stream_id} = CreateStream.execute(conn, stream_uuid, schema: schema) - :ok = Appender.append(conn, stream_id, recorded_events, schema: schema) + result = Appender.append(conn, stream_id, recorded_events, schema: schema, partitioned_events: partitioned) + + :ok = result [ recorded_events: recorded_events diff --git a/test/subscriptions/subscribe_to_stream_test.exs b/test/subscriptions/subscribe_to_stream_test.exs index 01d8c7b8..7a649243 100644 --- a/test/subscriptions/subscribe_to_stream_test.exs +++ b/test/subscriptions/subscribe_to_stream_test.exs @@ -24,6 +24,10 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do {:ok, %{subscription_name: subscription_name}} end + def partitioned? do + Application.get_env(:eventstore, EventStore)[:partitioned_events] || false + end + describe "single stream subscription" do setup [:append_events_to_another_stream] @@ -57,7 +61,7 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do {:ok, _subscription} = subscribe_to_stream(stream_uuid, subscription_name, self()) - :ok = EventStore.append_to_stream(stream_uuid, 0, events) + :ok = EventStore.append_to_stream(stream_uuid, 0, events, partitioned_events: partitioned?()) assert_receive {:events, received_events} assert pluck(received_events, :event_number) == [1, 2, 3] @@ -72,17 +76,17 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do end test "subscribe to single stream from given stream version should only receive later events", - %{subscription_name: subscription_name} do + %{subscription_name: subscription_name} do stream_uuid = UUID.uuid4() initial_events = EventFactory.create_events(1) new_events = EventFactory.create_events(1, 2) - :ok = EventStore.append_to_stream(stream_uuid, 0, initial_events) + :ok = EventStore.append_to_stream(stream_uuid, 0, initial_events, partitioned_events: partitioned?()) {:ok, _subscription} = subscribe_to_stream(stream_uuid, subscription_name, self(), start_from: 1) - :ok = EventStore.append_to_stream(stream_uuid, 1, new_events) + :ok = EventStore.append_to_stream(stream_uuid, 1, new_events, partitioned_events: partitioned?()) assert_receive {:events, received_events} assert pluck(received_events, :event_number) == [2] @@ -102,10 +106,10 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do stream_uuid = UUID.uuid4() assert {:ok, _subscription} = - EventStore.subscribe_to_stream(stream_uuid, subscription_name, self()) + EventStore.subscribe_to_stream(stream_uuid, subscription_name, self()) assert {:error, :subscription_already_exists} = - EventStore.subscribe_to_stream(stream_uuid, subscription_name, self()) + EventStore.subscribe_to_stream(stream_uuid, subscription_name, self()) end test "subscribe to single stream should ignore events from another stream", %{ @@ -120,8 +124,8 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do {:ok, _subscription} = subscribe_to_stream(interested_stream_uuid, subscription_name, self()) - :ok = EventStore.append_to_stream(interested_stream_uuid, 0, interested_events) - :ok = EventStore.append_to_stream(other_stream_uuid, 0, other_events) + :ok = EventStore.append_to_stream(interested_stream_uuid, 0, interested_events, partitioned_events: partitioned?()) + :ok = EventStore.append_to_stream(other_stream_uuid, 0, other_events, partitioned_events: partitioned?()) # received events should not include events from the other stream assert_receive {:events, received_events} @@ -130,7 +134,7 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do end test "subscribe to single stream with mapper function should receive all its mapped events", - %{subscription_name: subscription_name} do + %{subscription_name: subscription_name} do stream_uuid = UUID.uuid4() events = EventFactory.create_events(3) @@ -142,14 +146,14 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do mapper: fn event -> event.event_number end ) - :ok = EventStore.append_to_stream(stream_uuid, 0, events) + :ok = EventStore.append_to_stream(stream_uuid, 0, events, partitioned_events: partitioned?()) assert_receive {:events, received_mapped_events} assert received_mapped_events == [1, 2, 3] end test "subscribe to single stream with selector function should receive only filtered events", - %{subscription_name: subscription_name} do + %{subscription_name: subscription_name} do stream_uuid = UUID.uuid4() events = EventFactory.create_events(4) @@ -161,14 +165,15 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do selector: fn event -> rem(event.event_number, 2) == 0 end ) - :ok = EventStore.append_to_stream(stream_uuid, 0, events) + :ok = EventStore.append_to_stream(stream_uuid, 0, events, partitioned_events: partitioned?()) assert_receive_events(subscription, [2, 4]) end test "subscribe to single stream with selector function should continue to receive only filtered events", - %{subscription_name: subscription_name} do + %{subscription_name: subscription_name} do stream_uuid = UUID.uuid4() + events = EventFactory.create_events(3) {:ok, subscription} = @@ -179,11 +184,11 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do selector: fn event -> rem(event.event_number, 2) == 0 end ) - :ok = EventStore.append_to_stream(stream_uuid, 0, events) + :ok = EventStore.append_to_stream(stream_uuid, 0, events, partitioned_events: partitioned?()) assert_receive_events(subscription, [2]) - :ok = EventStore.append_to_stream(stream_uuid, 3, events) + :ok = EventStore.append_to_stream(stream_uuid, 3, events, partitioned_events: partitioned?()) assert_receive_events(subscription, [4, 6]) @@ -191,7 +196,7 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do end test "subscribe to single stream with selector function during catch-up should continue to receive only filtered events", - %{subscription_name: subscription_name} do + %{subscription_name: subscription_name} do stream_uuid = UUID.uuid4() :ok = EventStore.append_to_stream(stream_uuid, 0, EventFactory.create_events(3)) @@ -212,7 +217,7 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do end test "subscribe to single stream with selector function and mapper function should receive only filtered events and mapped events", - %{subscription_name: subscription_name} do + %{subscription_name: subscription_name} do stream_uuid = UUID.uuid4() events = EventFactory.create_events(4) @@ -228,7 +233,7 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do mapper: mapper ) - :ok = EventStore.append_to_stream(stream_uuid, 0, events) + :ok = EventStore.append_to_stream(stream_uuid, 0, events, partitioned_events: partitioned?()) assert_receive {:events, [2, 4]} end @@ -266,8 +271,8 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do initial_events = EventFactory.create_events(1) new_events = EventFactory.create_events(1, 2) - :ok = EventStore.append_to_stream(stream_uuid, 0, initial_events) - :ok = EventStore.append_to_stream(stream_uuid, 1, new_events) + :ok = EventStore.append_to_stream(stream_uuid, 0, initial_events, partitioned_events: partitioned?()) + :ok = EventStore.append_to_stream(stream_uuid, 1, new_events, partitioned_events: partitioned?()) {:ok, subscription} = EventStore.subscribe_to_stream(stream_uuid, subscription_name, self()) @@ -288,10 +293,10 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do test "should catch-up from unseen events", %{subscription_name: subscription_name} do stream_uuid = UUID.uuid4() - :ok = EventStore.append_to_stream(stream_uuid, 0, EventFactory.create_events(1)) - :ok = EventStore.append_to_stream(stream_uuid, 1, EventFactory.create_events(2)) - :ok = EventStore.append_to_stream(stream_uuid, 3, EventFactory.create_events(3)) - :ok = EventStore.append_to_stream(stream_uuid, 6, EventFactory.create_events(4)) + :ok = EventStore.append_to_stream(stream_uuid, 0, EventFactory.create_events(1), partitioned_events: partitioned?()) + :ok = EventStore.append_to_stream(stream_uuid, 1, EventFactory.create_events(2), partitioned_events: partitioned?()) + :ok = EventStore.append_to_stream(stream_uuid, 3, EventFactory.create_events(3), partitioned_events: partitioned?()) + :ok = EventStore.append_to_stream(stream_uuid, 6, EventFactory.create_events(4), partitioned_events: partitioned?()) {:ok, subscription} = subscribe_to_stream(stream_uuid, subscription_name, self()) @@ -323,7 +328,7 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do Wait.until(fn -> assert_hibernated(subscription) end) # Appending events to the stream should resume the subscription's event loop - :ok = EventStore.append_to_stream(stream_uuid, 0, events) + :ok = EventStore.append_to_stream(stream_uuid, 0, events, partitioned_events: partitioned?()) assert_receive {:events, received_events} @@ -354,8 +359,8 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do {:ok, subscription} = subscribe_to_all_streams(subscription_name, self(), buffer_size: 1) - :ok = EventStore.append_to_stream(stream1_uuid, 0, stream1_events) - :ok = EventStore.append_to_stream(stream2_uuid, 0, stream2_events) + :ok = EventStore.append_to_stream(stream1_uuid, 0, stream1_events, partitioned_events: partitioned?()) + :ok = EventStore.append_to_stream(stream2_uuid, 0, stream2_events, partitioned_events: partitioned?()) assert_receive {:events, stream1_received_events} assert pluck(stream1_received_events, :event_number) == [1] @@ -363,7 +368,7 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do assert pluck(stream1_received_events, :stream_version) == [1] assert pluck(stream1_received_events, :correlation_id) == - pluck(stream1_events, :correlation_id) + pluck(stream1_events, :correlation_id) assert pluck(stream1_received_events, :causation_id) == pluck(stream1_events, :causation_id) assert pluck(stream1_received_events, :event_type) == pluck(stream1_events, :event_type) @@ -379,7 +384,7 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do assert pluck(stream2_received_events, :stream_version) == [1] assert pluck(stream2_received_events, :correlation_id) == - pluck(stream2_events, :correlation_id) + pluck(stream2_events, :correlation_id) assert pluck(stream2_received_events, :causation_id) == pluck(stream2_events, :causation_id) assert pluck(stream2_received_events, :event_type) == pluck(stream2_events, :event_type) @@ -389,7 +394,7 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do end test "subscribe to all streams from given stream id should only receive later events from all streams", - %{subscription_name: subscription_name} do + %{subscription_name: subscription_name} do stream1_uuid = UUID.uuid4() stream2_uuid = UUID.uuid4() @@ -398,14 +403,14 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do stream1_new_events = EventFactory.create_events(1, 2) stream2_new_events = EventFactory.create_events(1, 2) - :ok = EventStore.append_to_stream(stream1_uuid, 0, stream1_initial_events) - :ok = EventStore.append_to_stream(stream2_uuid, 0, stream2_initial_events) + :ok = EventStore.append_to_stream(stream1_uuid, 0, stream1_initial_events, partitioned_events: partitioned?()) + :ok = EventStore.append_to_stream(stream2_uuid, 0, stream2_initial_events, partitioned_events: partitioned?()) {:ok, subscription} = subscribe_to_all_streams(subscription_name, self(), buffer_size: 1, start_from: 2) - :ok = EventStore.append_to_stream(stream1_uuid, 1, stream1_new_events) - :ok = EventStore.append_to_stream(stream2_uuid, 1, stream2_new_events) + :ok = EventStore.append_to_stream(stream1_uuid, 1, stream1_new_events, partitioned_events: partitioned?()) + :ok = EventStore.append_to_stream(stream2_uuid, 1, stream2_new_events, partitioned_events: partitioned?()) assert_receive {:events, stream1_received_events} @@ -425,7 +430,7 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do {:ok, subscription} = subscribe_to_all_streams(subscription_name, self(), buffer_size: 3) - :ok = EventStore.append_to_stream(stream_uuid, 0, events) + :ok = EventStore.append_to_stream(stream_uuid, 0, events, partitioned_events: partitioned?()) assert_receive_events(subscription, [1, 2, 3]) @@ -443,7 +448,7 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do {:ok, subscription} = subscribe_to_all_streams(subscription_name, self(), buffer_size: 3) - :ok = EventStore.append_to_stream(stream_uuid, 0, initial_events) + :ok = EventStore.append_to_stream(stream_uuid, 0, initial_events, partitioned_events: partitioned?()) assert_receive {:events, initial_received_events} assert length(initial_received_events) == 3 @@ -453,7 +458,7 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do :ok = Subscription.ack(subscription, hd(initial_received_events)) refute_receive {:events, _events} - :ok = EventStore.append_to_stream(stream_uuid, 3, remaining_events) + :ok = EventStore.append_to_stream(stream_uuid, 3, remaining_events, partitioned_events: partitioned?()) # Acknowledge receipt of all initial events Subscription.ack(subscription, initial_received_events) @@ -477,8 +482,8 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do refute_receive {:events, _events} - :ok = EventStore.append_to_stream(stream1_uuid, 0, stream1_events) - :ok = EventStore.append_to_stream(stream2_uuid, 0, stream2_events) + :ok = EventStore.append_to_stream(stream1_uuid, 0, stream1_events, partitioned_events: partitioned?()) + :ok = EventStore.append_to_stream(stream2_uuid, 0, stream2_events, partitioned_events: partitioned?()) assert_receive {:events, stream1_received_events} @@ -510,7 +515,7 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do start_from: 3 ) - :ok = EventStore.append_to_stream(stream_uuid, 0, events) + :ok = EventStore.append_to_stream(stream_uuid, 0, events, partitioned_events: partitioned?()) # Should receive the same three events from both subscriptions assert_receive {:events, [%RecordedEvent{event_number: 1} | _events] = received_events1} @@ -523,11 +528,11 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do end defp assert_received_events( - received_events, - stream_uuid, - expected_events, - expected_event_numbers - ) do + received_events, + stream_uuid, + expected_events, + expected_event_numbers + ) do assert pluck(received_events, :event_number) == expected_event_numbers assert pluck(received_events, :stream_uuid) == [stream_uuid, stream_uuid, stream_uuid] assert pluck(received_events, :stream_version) == [1, 2, 3] @@ -558,8 +563,8 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do assert_receive {:subscribed, ^subscriber3} assert_receive {:subscribed, ^subscriber4} - :ok = EventStore.append_to_stream(stream1_uuid, 0, stream1_events) - :ok = EventStore.append_to_stream(stream2_uuid, 0, stream2_events) + :ok = EventStore.append_to_stream(stream1_uuid, 0, stream1_events, partitioned_events: partitioned?()) + :ok = EventStore.append_to_stream(stream2_uuid, 0, stream2_events, partitioned_events: partitioned?()) Wait.until(fn -> all_received_events = @@ -586,7 +591,7 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do initial_events = EventFactory.create_events(1) new_events = EventFactory.create_events(1, 2) - :ok = EventStore.append_to_stream(stream_uuid, 0, initial_events) + :ok = EventStore.append_to_stream(stream_uuid, 0, initial_events, partitioned_events: partitioned?()) {:ok, subscription} = EventStore.subscribe_to_stream(stream_uuid, subscription_name, self()) @@ -599,7 +604,7 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do Process.exit(subscription, :kill) refute Process.info(subscription) - :ok = EventStore.append_to_stream(stream_uuid, 1, new_events) + :ok = EventStore.append_to_stream(stream_uuid, 1, new_events, partitioned_events: partitioned?()) {:ok, subscription} = EventStore.subscribe_to_stream(stream_uuid, subscription_name, self()) assert_receive {:subscribed, ^subscription} @@ -616,7 +621,7 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do initial_events = EventFactory.create_events(1) new_events = EventFactory.create_events(1, 2) - :ok = EventStore.append_to_stream(stream_uuid, 0, initial_events) + :ok = EventStore.append_to_stream(stream_uuid, 0, initial_events, partitioned_events: partitioned?()) {:ok, subscription} = EventStore.subscribe_to_stream(stream_uuid, subscription_name, self(), transient: true) @@ -630,7 +635,7 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do Process.exit(subscription, :kill) refute Process.info(subscription) - :ok = EventStore.append_to_stream(stream_uuid, 1, new_events) + :ok = EventStore.append_to_stream(stream_uuid, 1, new_events, partitioned_events: partitioned?()) {:ok, subscription} = EventStore.subscribe_to_stream(stream_uuid, subscription_name, self(), transient: true) @@ -648,8 +653,8 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do new_events = EventFactory.create_events(1, 2) after_restart_events = EventFactory.create_events(1, 3) - :ok = EventStore.append_to_stream(stream_uuid, 0, initial_events) - :ok = EventStore.append_to_stream(stream_uuid, 1, new_events) + :ok = EventStore.append_to_stream(stream_uuid, 0, initial_events, partitioned_events: partitioned?()) + :ok = EventStore.append_to_stream(stream_uuid, 1, new_events, partitioned_events: partitioned?()) {:ok, subscription} = EventStore.subscribe_to_stream(stream_uuid, subscription_name, self(), transient: true) @@ -666,7 +671,7 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do Process.exit(subscription, :kill) refute Process.info(subscription) - :ok = EventStore.append_to_stream(stream_uuid, 2, after_restart_events) + :ok = EventStore.append_to_stream(stream_uuid, 2, after_restart_events, partitioned_events: partitioned?()) {:ok, subscription} = EventStore.subscribe_to_stream(stream_uuid, subscription_name, self(), @@ -687,7 +692,7 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do initial_events = EventFactory.create_events(1) new_events = EventFactory.create_events(1, 2) - :ok = EventStore.append_to_stream(stream_uuid, 0, initial_events) + :ok = EventStore.append_to_stream(stream_uuid, 0, initial_events, partitioned_events: partitioned?()) {:ok, subscription} = EventStore.subscribe_to_stream(stream_uuid, subscription_name, self(), transient: true) @@ -701,7 +706,7 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do Process.exit(subscription, :kill) refute Process.info(subscription) - :ok = EventStore.append_to_stream(stream_uuid, 1, new_events) + :ok = EventStore.append_to_stream(stream_uuid, 1, new_events, partitioned_events: partitioned?()) {:ok, subscription} = EventStore.subscribe_to_stream(stream_uuid, subscription_name, self(), transient: false) @@ -719,7 +724,7 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do initial_events = EventFactory.create_events(1) new_events = EventFactory.create_events(1, 2) - :ok = EventStore.append_to_stream(stream_uuid, 0, initial_events) + :ok = EventStore.append_to_stream(stream_uuid, 0, initial_events, partitioned_events: partitioned?()) {:ok, subscription} = EventStore.subscribe_to_stream(stream_uuid, subscription_name, self(), transient: false) @@ -733,7 +738,7 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do Process.exit(subscription, :kill) refute Process.info(subscription) - :ok = EventStore.append_to_stream(stream_uuid, 1, new_events) + :ok = EventStore.append_to_stream(stream_uuid, 1, new_events, partitioned_events: partitioned?()) {:ok, subscription} = EventStore.subscribe_to_stream(stream_uuid, subscription_name, self(), transient: true) @@ -762,8 +767,8 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do initial_events = EventFactory.create_events(1) new_events = EventFactory.create_events(1, 2) - :ok = EventStore.append_to_stream(stream_uuid, 0, initial_events) - :ok = EventStore.append_to_stream(stream_uuid, 1, new_events) + :ok = EventStore.append_to_stream(stream_uuid, 0, initial_events, partitioned_events: partitioned?()) + :ok = EventStore.append_to_stream(stream_uuid, 1, new_events, partitioned_events: partitioned?()) {:ok, subscription} = EventStore.subscribe_to_stream(stream_uuid, subscription_name, self()) @@ -814,7 +819,7 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do stream_uuid = UUID.uuid4() events = EventFactory.create_events(3) - :ok = EventStore.append_to_stream(stream_uuid, 0, events) + :ok = EventStore.append_to_stream(stream_uuid, 0, events, partitioned_events: partitioned?()) end # Subscribe to a single stream and wait for the subscription to be subscribed diff --git a/test/subscriptions/subscription_acknowledgement_test.exs b/test/subscriptions/subscription_acknowledgement_test.exs index 29b46e22..7ec29141 100644 --- a/test/subscriptions/subscription_acknowledgement_test.exs +++ b/test/subscriptions/subscription_acknowledgement_test.exs @@ -5,6 +5,10 @@ defmodule EventStore.Subscriptions.SubscriptionAcknowledgementTest do alias EventStore.Subscriptions.Subscription alias TestEventStore, as: EventStore + def partitioned? do + Application.get_env(:eventstore, EventStore)[:partitioned_events] || false + end + describe "subscription acknowledgement" do test "should checkpoint after each event by default", %{conn: conn} do {:ok, subscription} = subscribe_to_all_streams(buffer_size: 1) @@ -191,7 +195,7 @@ defmodule EventStore.Subscriptions.SubscriptionAcknowledgementTest do defp append_to_stream(stream_uuid, event_count) do events = EventFactory.create_events(event_count) - EventStore.append_to_stream(stream_uuid, 0, events) + EventStore.append_to_stream(stream_uuid, 0, events, partitioned_events: partitioned?()) end # Subscribe to all streams and wait for the subscription to be subscribed. diff --git a/test/subscriptions/subscription_back_pressure_test.exs b/test/subscriptions/subscription_back_pressure_test.exs index c9b22b77..d3215cb1 100644 --- a/test/subscriptions/subscription_back_pressure_test.exs +++ b/test/subscriptions/subscription_back_pressure_test.exs @@ -5,6 +5,10 @@ defmodule EventStore.Subscriptions.SubscriptionBackPressureTest do alias EventStore.Subscriptions.Subscription alias TestEventStore, as: EventStore + def partitioned? do + Application.get_env(:eventstore, EventStore)[:partitioned_events] || false + end + describe "subscription back pressure" do test "should receive pending events once caught up" do {:ok, subscription} = subscribe_to_all_streams(buffer_size: 5, max_size: 5) @@ -91,7 +95,7 @@ defmodule EventStore.Subscriptions.SubscriptionBackPressureTest do defp append_to_stream(stream_uuid, event_count) do events = EventFactory.create_events(event_count) - :ok = EventStore.append_to_stream(stream_uuid, 0, events) + :ok = EventStore.append_to_stream(stream_uuid, 0, events, partitioned_events: partitioned?()) end # Subscribe to all streams and wait for the subscription to be subscribed. diff --git a/test/subscriptions/subscription_catch_up_test.exs b/test/subscriptions/subscription_catch_up_test.exs index 5918b7ef..33194a43 100644 --- a/test/subscriptions/subscription_catch_up_test.exs +++ b/test/subscriptions/subscription_catch_up_test.exs @@ -5,6 +5,10 @@ defmodule EventStore.Subscriptions.SubscriptionCatchUpTest do alias EventStore.Subscriptions.Subscription alias TestEventStore, as: EventStore + def partitioned? do + Application.get_env(:eventstore, EventStore)[:partitioned_events] || false + end + describe "catch-up subscription" do test "should receive all existing events" do restart_event_store_with_config(enable_hard_deletes: false) @@ -131,7 +135,7 @@ defmodule EventStore.Subscriptions.SubscriptionCatchUpTest do defp append_to_stream(stream_uuid, event_count, expected_version \\ 0) do events = EventFactory.create_events(event_count) - :ok = EventStore.append_to_stream(stream_uuid, expected_version, events) + :ok = EventStore.append_to_stream(stream_uuid, expected_version, events, partitioned_events: partitioned?()) end # Subscribe to all streams and wait for the subscription to be subscribed. diff --git a/test/subscriptions/subscription_recovery_test.exs b/test/subscriptions/subscription_recovery_test.exs index 9f4f6eb7..051dc21b 100644 --- a/test/subscriptions/subscription_recovery_test.exs +++ b/test/subscriptions/subscription_recovery_test.exs @@ -4,6 +4,10 @@ defmodule EventStore.Subscriptions.SubscriptionRecoveryTest do alias EventStore.{EventFactory, RecordedEvent, UUID, Wait} alias EventStore.Subscriptions.Subscription alias TestEventStore, as: EventStore + + def partitioned? do + Application.get_env(:eventstore, EventStore)[:partitioned_events] || false + end describe "subscription recovery" do test "should receive events after socket is closed" do @@ -61,7 +65,7 @@ defmodule EventStore.Subscriptions.SubscriptionRecoveryTest do defp append_to_stream(stream_uuid, event_count, expected_version \\ 0) do events = EventFactory.create_events(event_count) - :ok = EventStore.append_to_stream(stream_uuid, expected_version, events) + :ok = EventStore.append_to_stream(stream_uuid, expected_version, events, partitioned_events: partitioned?()) end # Subscribe to all streams and wait for the subscription to be subscribed. diff --git a/test/support/subscription_helpers.ex b/test/support/subscription_helpers.ex index bbead495..21cf1023 100644 --- a/test/support/subscription_helpers.ex +++ b/test/support/subscription_helpers.ex @@ -5,10 +5,14 @@ defmodule EventStore.SubscriptionHelpers do alias EventStore.Subscriptions.Subscription alias TestEventStore, as: EventStore + def partitioned? do + Application.get_env(:eventstore, EventStore)[:partitioned_events] || false + end + def append_to_stream(stream_uuid, event_count, expected_version \\ 0) do events = EventFactory.create_events(event_count, expected_version + 1) - EventStore.append_to_stream(stream_uuid, expected_version, events) + EventStore.append_to_stream(stream_uuid, expected_version, events, partitioned_events: partitioned?()) end @doc """ From f1c221b813de6cb9765815096e6e8b5e84ee88ae Mon Sep 17 00:00:00 2001 From: Thierry BOMANDOUKI Date: Tue, 27 Jan 2026 19:22:43 +0100 Subject: [PATCH 07/17] Fix bench --- bench/seed_eventstore.exs | 5 ++++- bench/storage/append_events_bench.exs | 3 ++- bench/storage/subscribe_to_stream_bench.exs | 3 ++- config/bench.exs | 6 ++++-- lib/event_store.ex | 5 +++++ 5 files changed, 17 insertions(+), 5 deletions(-) diff --git a/bench/seed_eventstore.exs b/bench/seed_eventstore.exs index 09934713..017383db 100644 --- a/bench/seed_eventstore.exs +++ b/bench/seed_eventstore.exs @@ -11,6 +11,9 @@ # defmodule EventBuilder do alias EventStore.{EventFactory, UUID} + alias TestEventStore, as: EventStore + + partitioned = Application.get_env(:eventstore, EventStore)[:partitioned_events] || false def seed(total_event_count, events_per_stream, initial_event_number \\ 0) @@ -24,7 +27,7 @@ defmodule EventBuilder do event_count = min(total_event_count, events_per_stream) events = EventFactory.create_events(event_count, initial_event_number) - :ok = TestEventStore.append_to_stream(stream_uuid, 0, events) + :ok = TestEventStore.append_to_stream(stream_uuid, 0, events, partitioned_events: partitioned) remaining_event_count = total_event_count - event_count next_event_number = initial_event_number + event_count diff --git a/bench/storage/append_events_bench.exs b/bench/storage/append_events_bench.exs index 5dee1619..68afcb90 100644 --- a/bench/storage/append_events_bench.exs +++ b/bench/storage/append_events_bench.exs @@ -40,13 +40,14 @@ defmodule AppendEventsBench do defp append_events(context, concurrency) do events = Keyword.fetch!(context, :events) + partitioned = Application.get_env(:eventstore, EventStore)[:partitioned_events] || false tasks = Enum.map(1..concurrency, fn _ -> stream_uuid = UUID.uuid4() Task.async(fn -> - EventStore.append_to_stream(stream_uuid, 0, events) + EventStore.append_to_stream(stream_uuid, 0, events, partitioned_events: partitioned) end) end) diff --git a/bench/storage/subscribe_to_stream_bench.exs b/bench/storage/subscribe_to_stream_bench.exs index e319b2df..18627f97 100644 --- a/bench/storage/subscribe_to_stream_bench.exs +++ b/bench/storage/subscribe_to_stream_bench.exs @@ -45,6 +45,7 @@ defmodule SubscribeToStreamBench do defp subscribe_to_stream(context, concurrency, opts \\ []) do events = Keyword.fetch!(context, :events) stream_uuid = UUID.uuid4() + partitioned = Application.get_env(:eventstore, EventStore)[:partitioned_events] || false tasks = Enum.map(1..concurrency, fn index -> @@ -67,7 +68,7 @@ defmodule SubscribeToStreamBench do append_task = Task.async(fn -> - :ok = EventStore.append_to_stream(stream_uuid, 0, events) + :ok = EventStore.append_to_stream(stream_uuid, 0, events, partitioned_events: partitioned) end) Enum.each([append_task | tasks], &Task.await(&1, @await_timeout_ms)) diff --git a/config/bench.exs b/config/bench.exs index f5ce46da..de733345 100644 --- a/config/bench.exs +++ b/config/bench.exs @@ -13,11 +13,13 @@ default_config = [ database: "eventstore_bench", hostname: "localhost", pool_size: 10, - serializer: EventStore.TermSerializer + serializer: EventStore.TermSerializer, + partitioned_events: true, # Default false, set to true if you want a partioned events table + use_pg_partman: true ] config :eventstore, TestEventStore, default_config config :eventstore, SchemaEventStore, default_config -config :eventstore, SecondEventStore, Keyword.put(default_config, :database, "eventstore_test_2") +config :eventstore, SecondEventStore, Keyword.put(default_config, :database, "eventstore_bench_2") config :eventstore, event_stores: [TestEventStore] diff --git a/lib/event_store.ex b/lib/event_store.ex index efaf1c12..c7641e2c 100644 --- a/lib/event_store.ex +++ b/lib/event_store.ex @@ -296,6 +296,11 @@ defmodule EventStore do @accepted_overrides_append_to_stream [:created_at_override] + def append_to_stream(stream_uuid, expected_version, events) do + partitioned = Application.get_env(:eventstore, EventStore)[:partitioned_events] || false + append_to_stream(stream_uuid, expected_version, events, partitioned_events: partitioned) + end + def append_to_stream(stream_uuid, expected_version, events, opts \\ []) def append_to_stream(@all_stream, _expected_version, _events, _opts), From 5a900df4e41b2b6b3eed8861fa9af95e5f15c583 Mon Sep 17 00:00:00 2001 From: Thierry Bomandouki Date: Thu, 29 Jan 2026 01:59:55 +0100 Subject: [PATCH 08/17] Update Getting Started.md Update getting started documentation --- guides/Getting Started.md | 33 +++++++++++++++++++++++++++++++++ 1 file changed, 33 insertions(+) diff --git a/guides/Getting Started.md b/guides/Getting Started.md index 4bb5e38d..9a8f49b2 100644 --- a/guides/Getting Started.md +++ b/guides/Getting Started.md @@ -250,3 +250,36 @@ config :my_app, MyApp.EventStore, ``` This will allow the EventStore to use your regular pool settings to connect to the database defined in `url` for most database operations. It will separately establish connections using the `session_mode_url` where necessary which you should point to PgBouncer in session mode or connected directly to the Postgres instance. + +## Working with partitioned events table + +For performance reasons, due to a very large number of events stored in the `events` table, it may be advisable to partition this table by date (using `created_at` as the partitioning key). + +### Enabling partitioning + +1. To enable partitioning support, add the `partitioned_events` parameter to the configuration file (e.g. `config/dev.exs`): + +```elixir +config :my_app, MyApp.EventStore, + serializer: EventStore.JsonSerializer, + username: "postgres", + password: "postgres", + database: "eventstore", + hostname: "localhost", + partitioned_events: true +``` + +To enable automatic partition management, it is also possible to use the PostgreSQL extension [pg_partman](https://github.com/pgpartman/pg_partman). After installing pg_partman, add the `use_pg_partman` parameter to the configuration file (e.g. `config/dev.exs`): + +```elixir +config :my_app, MyApp.EventStore, + serializer: EventStore.JsonSerializer, + username: "postgres", + password: "postgres", + database: "eventstore", + hostname: "localhost", + partitioned_events: true, + use_pg_partman: true +``` + + From 856e67e94fe4e1748a1c8cd6b86e8939f507ae1b Mon Sep 17 00:00:00 2001 From: Thierry BOMANDOUKI Date: Thu, 29 Jan 2026 11:45:13 +0100 Subject: [PATCH 09/17] Fix bad comment --- lib/event_store/sql/statements/insert_events.sql.eex | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib/event_store/sql/statements/insert_events.sql.eex b/lib/event_store/sql/statements/insert_events.sql.eex index d36ca13b..d56a6d2e 100644 --- a/lib/event_store/sql/statements/insert_events.sql.eex +++ b/lib/event_store/sql/statements/insert_events.sql.eex @@ -41,7 +41,7 @@ WITH events_root AS ( <% # insert the new events into the events_root table - # using the 7 bind variables from 3 to 9 inclusive + # using the 1 bind variables from 3 to 9 inclusive # n.b.: the bind for the event_id is re-generated here %> INSERT INTO "<%= schema %>".events_root From 1bab6dfca7f585070ab96368f80a61d21ed368ea Mon Sep 17 00:00:00 2001 From: Thierry BOMANDOUKI Date: Thu, 29 Jan 2026 11:48:48 +0100 Subject: [PATCH 10/17] Update comments --- config/config.exs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/config/config.exs b/config/config.exs index 80c91a0f..7d66c29e 100644 --- a/config/config.exs +++ b/config/config.exs @@ -1,9 +1,9 @@ import Config -# Configuration globale de EventStore +# Global configuration of EventStore config :eventstore, EventStore, - partitioned_events: false, # Default false, set to true if you want a partioned events table - use_pg_partman: false # Default false, set to true if you want to use postgresql extension pg_partman + partitioned_events: false, # Set to true if you want a partioned events table + use_pg_partman: false # Set to true if you want to use postgresql extension pg_partman config :eventstore, event_stores: [DevEventStore] From 2af97e2dd3db49bef066b17b2408eff08a2507f5 Mon Sep 17 00:00:00 2001 From: Thierry BOMANDOUKI Date: Thu, 29 Jan 2026 12:47:59 +0100 Subject: [PATCH 11/17] Use Keyword.get instead of Keyword.fetch for optional params --- lib/event_store/sql/init.ex | 4 ++-- lib/event_store/sql/reset.ex | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/lib/event_store/sql/init.ex b/lib/event_store/sql/init.ex index 82e59f64..d60ffda0 100644 --- a/lib/event_store/sql/init.ex +++ b/lib/event_store/sql/init.ex @@ -15,8 +15,8 @@ defmodule EventStore.Sql.Init do column_data_type = Keyword.fetch!(config, :column_data_type) schema = Keyword.fetch!(config, :schema) || 'event_store' database = Keyword.fetch!(config, :database) - partitioned = Keyword.fetch!(config, :partitioned_events) || false - partman = Keyword.fetch!(config, :use_pg_partman) || false + partitioned = Keyword.get(config, :partitioned_events, false) + partman = Keyword.get(config, :use_pg_partman, false) [ ~s(SET LOCAL search_path TO "#{schema}";), diff --git a/lib/event_store/sql/reset.ex b/lib/event_store/sql/reset.ex index 37577fe7..eb7204f7 100644 --- a/lib/event_store/sql/reset.ex +++ b/lib/event_store/sql/reset.ex @@ -5,8 +5,8 @@ defmodule EventStore.Sql.Reset do def statements(config) do schema = Keyword.fetch!(config, :schema) - partitioned = Keyword.fetch!(config, :partitioned_events) || false - partman = Keyword.fetch!(config, :use_pg_partman) || false + partitioned = Keyword.get(config, :partitioned_events, false) + partman = Keyword.get(config, :use_pg_partman, false) [ ~s(SET LOCAL search_path TO "#{schema}";), From 644aa2583d82fc41c14c6d5e8a52604aeedefe22 Mon Sep 17 00:00:00 2001 From: Thierry BOMANDOUKI Date: Thu, 29 Jan 2026 12:50:27 +0100 Subject: [PATCH 12/17] Remove dead code --- lib/event_store/streams/stream.ex | 1 - 1 file changed, 1 deletion(-) diff --git a/lib/event_store/streams/stream.ex b/lib/event_store/streams/stream.ex index 499bf9be..8361fc66 100644 --- a/lib/event_store/streams/stream.ex +++ b/lib/event_store/streams/stream.ex @@ -218,7 +218,6 @@ defmodule EventStore.Streams.Stream do %StreamInfo{stream_id: stream_id} = stream opts = Keyword.put(opts, :expected_version, expected_version) - #IO.inspect(opts) Storage.append_to_stream(conn, stream_id, prepared_events, opts) end From cf539b81ae902d7167427242c374ef3096fcb4d1 Mon Sep 17 00:00:00 2001 From: Thierry BOMANDOUKI Date: Thu, 29 Jan 2026 12:52:06 +0100 Subject: [PATCH 13/17] Revert personal change --- config/test.exs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/config/test.exs b/config/test.exs index e2446dd9..f0272785 100644 --- a/config/test.exs +++ b/config/test.exs @@ -23,7 +23,7 @@ default_config = [ ] config :eventstore, TestEventStore, default_config -config :eventstore, SecondEventStore, Keyword.put(default_config, :database, "thierryb_eventstore_test_2") +config :eventstore, SecondEventStore, Keyword.put(default_config, :database, "eventstore_test_2") config :eventstore, SchemaEventStore, default_config config :eventstore, event_stores: [TestEventStore, SecondEventStore, SchemaEventStore] From 7cc3fce68c3f62020c6f1a4e3f5c81532a7d58e8 Mon Sep 17 00:00:00 2001 From: Thierry BOMANDOUKI Date: Thu, 29 Jan 2026 12:55:02 +0100 Subject: [PATCH 14/17] Set default values to false for partitioned_events and use_pg_partman --- config/dev.exs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/config/dev.exs b/config/dev.exs index 24d7fe5d..7b464f7d 100644 --- a/config/dev.exs +++ b/config/dev.exs @@ -8,5 +8,5 @@ config :mix_test_watch, clear: true config :eventstore, DevEventStore, schema: "event_store", column_data_type: "jsonb", - partitioned_events: true, - use_pg_partman: true + partitioned_events: false, + use_pg_partman: false From 651893c574eefc6701d46fd3a3e7a8980e1aa7c6 Mon Sep 17 00:00:00 2001 From: Thierry BOMANDOUKI Date: Thu, 29 Jan 2026 16:07:25 +0100 Subject: [PATCH 15/17] Remove dead code --- lib/event_store/streams/stream.ex | 4 ---- 1 file changed, 4 deletions(-) diff --git a/lib/event_store/streams/stream.ex b/lib/event_store/streams/stream.ex index 8361fc66..eb1c0910 100644 --- a/lib/event_store/streams/stream.ex +++ b/lib/event_store/streams/stream.ex @@ -6,11 +6,8 @@ defmodule EventStore.Streams.Stream do def append_to_stream(conn, stream_uuid, expected_version, events, opts) when length(events) < 1000 do - #IO.inspect(events) {serializer, new_opts} = Keyword.pop(opts, :serializer) - #IO.inspect(stream_info(conn, stream_uuid, expected_version, new_opts)) - with {:ok, stream} <- stream_info(conn, stream_uuid, expected_version, new_opts), :ok <- do_append_to_storage(conn, stream, events, expected_version, serializer, new_opts) do :ok @@ -149,7 +146,6 @@ defmodule EventStore.Streams.Stream do opts ) do prepared_events = prepare_events(events, stream, serializer, opts) - #IO.inspect(prepared_events) write_to_stream(conn, prepared_events, stream, expected_version, opts) end From 1758f21fb75bb2a63b4e512a76fcc4355329aa43 Mon Sep 17 00:00:00 2001 From: Thierry BOMANDOUKI Date: Fri, 30 Jan 2026 15:04:06 +0100 Subject: [PATCH 16/17] Remove dead code --- lib/event_store/sql/statements.ex | 1 - test/streams/all_stream_test.exs | 2 -- 2 files changed, 3 deletions(-) diff --git a/lib/event_store/sql/statements.ex b/lib/event_store/sql/statements.ex index 5bb68d5e..9733774c 100644 --- a/lib/event_store/sql/statements.ex +++ b/lib/event_store/sql/statements.ex @@ -37,7 +37,6 @@ defmodule EventStore.Sql.Statements do @external_resource file - #EEx.function_from_file(:def, fun, file, args ++ [:partitioned], engine: EventStore.EExIOListEngine) EEx.function_from_file(:def, fun, file, args, engine: EventStore.EExIOListEngine) end end diff --git a/test/streams/all_stream_test.exs b/test/streams/all_stream_test.exs index 899ac5ac..d75c0a4f 100644 --- a/test/streams/all_stream_test.exs +++ b/test/streams/all_stream_test.exs @@ -323,8 +323,6 @@ defmodule EventStore.Streams.AllStreamTest do stream_uuid = UUID.uuid4() events = EventFactory.create_events(3) - #IO.inspect(opts) - partitioned = Application.get_env(:eventstore, TestEventStore)[:partitioned_events] || false :ok = Stream.append_to_stream(conn, stream_uuid, 0, events, Keyword.put(opts, :partitioned_events, partitioned)) From 9687d3c1ab813146f6977f10492b63ded10ac102 Mon Sep 17 00:00:00 2001 From: Thierry BOMANDOUKI Date: Fri, 30 Jan 2026 15:07:56 +0100 Subject: [PATCH 17/17] Remove bad comment and using pg_partman as default behaviour --- config/bench.exs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/config/bench.exs b/config/bench.exs index de733345..4f3b99db 100644 --- a/config/bench.exs +++ b/config/bench.exs @@ -14,8 +14,8 @@ default_config = [ hostname: "localhost", pool_size: 10, serializer: EventStore.TermSerializer, - partitioned_events: true, # Default false, set to true if you want a partioned events table - use_pg_partman: true + partitioned_events: true, + use_pg_partman: false ] config :eventstore, TestEventStore, default_config