diff --git a/Gemfile b/Gemfile index 15f6f4c..51d5372 100644 --- a/Gemfile +++ b/Gemfile @@ -13,4 +13,5 @@ gem "minitest-mock" gem "concurrent-ruby" gem "mutant-minitest" gem "pg-ephemeral" +gem "sequel" gem "syntax_tree" diff --git a/Gemfile.lock b/Gemfile.lock index 075bab6..a8f1d64 100644 --- a/Gemfile.lock +++ b/Gemfile.lock @@ -70,6 +70,8 @@ GEM reline (0.6.3) io-console (~> 0.5) securerandom (0.4.1) + sequel (5.103.0) + bigdecimal sorbet-runtime (0.6.13164) stringio (3.2.0) syntax_tree (6.3.0) @@ -96,6 +98,7 @@ DEPENDENCIES mutant-minitest pg-ephemeral rake + sequel syntax_tree CHECKSUMS @@ -134,6 +137,7 @@ CHECKSUMS regexp_parser (2.12.0) sha256=35a916a1d63190ab5c9009457136ae5f3c0c7512d60291d0d1378ba18ce08ebb reline (0.6.3) sha256=1198b04973565b36ec0f11542ab3f5cfeeec34823f4e54cebde90968092b1835 securerandom (0.4.1) sha256=cc5193d414a4341b6e225f0cb4446aceca8e50d5e1888743fac16987638ea0b1 + sequel (5.103.0) sha256=51bf23374cc585724fc51a2ae8b95283422c97ca4585ce28f3db27a26ce55a69 sorbet-runtime (0.6.13164) sha256=4561933ac3be0b2477c146540be8e8a3a8b23971ef46cfc89f5d45272830a651 stringio (3.2.0) sha256=c37cb2e58b4ffbd33fe5cd948c05934af997b36e0b6ca6fdf43afa234cf222e1 syntax_tree (6.3.0) sha256=56e25a9692c798ec94c5442fe94c5e94af76bef91edc8bb02052cbdecf35f13d diff --git a/lib/en57.rb b/lib/en57.rb index 5b1a9e1..e5dc747 100644 --- a/lib/en57.rb +++ b/lib/en57.rb @@ -6,6 +6,7 @@ require_relative "en57/json_serializer" require_relative "en57/query" require_relative "en57/scope" require_relative "en57/pg_adapter" +require_relative "en57/sequel_adapter" require_relative "en57/repository" require_relative "en57/event_store" diff --git a/lib/en57/sequel_adapter.rb b/lib/en57/sequel_adapter.rb new file mode 100644 index 0000000..c3bef49 --- /dev/null +++ b/lib/en57/sequel_adapter.rb @@ -0,0 +1,29 @@ +# frozen_string_literal: true + +require "pg" + +module En57 + class SequelAdapter + def initialize(database) + @database = database + end + + def with_connection(&block) + @database.synchronize(&block) + end + + def with_serializable_transaction + @database.transaction(isolation: :serializable) do + @database.synchronize { |connection| yield connection } + end + rescue StandardError => e + if defined?(Sequel::DatabaseError) && + e.is_a?(Sequel::DatabaseError) && + e.wrapped_exception.is_a?(PG::Error) + raise e.wrapped_exception + end + + raise + end + end +end diff --git a/test/test_helper.rb b/test/test_helper.rb index af7fc77..1e6e60c 100644 --- a/test/test_helper.rb +++ b/test/test_helper.rb @@ -8,16 +8,23 @@ require "en57" require "securerandom" require "concurrent-ruby" require "pg_ephemeral" +require "sequel" module En57 class IntegrationTest < Minitest::Test SERVER = PgEphemeral.start CONNECTION = PG.connect(SERVER.url) + SEQUEL_DB = Sequel.connect(SERVER.url) + ADAPTERS = { + pg: -> { PgAdapter.new(SERVER.url) }, + sequel: -> { SequelAdapter.new(SEQUEL_DB) }, + } def setup = CONNECTION.exec("TRUNCATE TABLE tags, events RESTART IDENTITY CASCADE") Minitest.after_run do + SEQUEL_DB.disconnect CONNECTION.close SERVER.shutdown end diff --git a/test/test_integration.rb b/test/test_integration.rb index cd12708..8554137 100644 --- a/test/test_integration.rb +++ b/test/test_integration.rb @@ -4,202 +4,216 @@ require "test_helper" module En57 class TestIntegration < IntegrationTest - def test_happy_path - with_event_store do |event_store| - events = [ - Event.new(id: ids[0], type: "CreditsToppedUp", data: { amount: 100 }), - Event.new(id: ids[1], type: "CreditsToppedUp", data: { amount: 50 }), - ] - - assert_equal(events, event_store.append(events).read.each.to_a) + ADAPTERS.each do |name, factory| + define_method "test_#{name}_happy_path" do + with_event_store(factory) do |event_store| + events = [ + Event.new( + id: ids[0], + type: "CreditsToppedUp", + data: { + amount: 100, + }, + ), + Event.new( + id: ids[1], + type: "CreditsToppedUp", + data: { + amount: 50, + }, + ), + ] + + assert_equal(events, event_store.append(events).read.each.to_a) + end end - end - def test_append_with_fail_if_and_no_matches_appends_events - with_event_store do |event_store| - event = Event.new(id: ids[0], type: "OrderPlaced") - event_store.append( - [event], - fail_if: event_store.read.of_type("PriceChanged"), - ) + define_method "test_#{name}_append_with_fail_if_and_no_matches_appends_events" do + with_event_store(factory) do |event_store| + event = Event.new(id: ids[0], type: "OrderPlaced") + event_store.append( + [event], + fail_if: event_store.read.of_type("PriceChanged"), + ) - assert_equal([event], event_store.read.each.to_a) + assert_equal([event], event_store.read.each.to_a) + end end - end - def test_append_with_fail_if_and_matches_raises_append_condition_violated - with_event_store do |event_store| - existing_event = Event.new(id: ids[0], type: "OrderPlaced") - event_store.append([existing_event]) + define_method "test_#{name}_append_with_fail_if_and_matches_raises_append_condition_violated" do + with_event_store(factory) do |event_store| + existing_event = Event.new(id: ids[0], type: "OrderPlaced") + event_store.append([existing_event]) - assert_raises(AppendConditionViolated) do + assert_raises(AppendConditionViolated) do + event_store.append( + [Event.new(id: ids[1], type: "ShipmentScheduled")], + fail_if: event_store.read.of_type("OrderPlaced"), + ) + end + + assert_equal([existing_event], event_store.read.each.to_a) + end + end + + define_method "test_#{name}_append_with_after_ignores_matches_at_or_before_cutoff" do + with_event_store(factory) do |event_store| + existing_event = Event.new(id: ids[0], type: "OrderPlaced") + event_store.append([existing_event]) event_store.append( [Event.new(id: ids[1], type: "ShipmentScheduled")], - fail_if: event_store.read.of_type("OrderPlaced"), + fail_if: event_store.read.of_type("OrderPlaced").after(1), ) - end - assert_equal([existing_event], event_store.read.each.to_a) + assert_equal( + [existing_event, Event.new(id: ids[1], type: "ShipmentScheduled")], + event_store.read.each.to_a, + ) + end end - end - def test_append_with_after_ignores_matches_at_or_before_cutoff - with_event_store do |event_store| - existing_event = Event.new(id: ids[0], type: "OrderPlaced") - event_store.append([existing_event]) - event_store.append( - [Event.new(id: ids[1], type: "ShipmentScheduled")], - fail_if: event_store.read.of_type("OrderPlaced").after(1), - ) - - assert_equal( - [existing_event, Event.new(id: ids[1], type: "ShipmentScheduled")], - event_store.read.each.to_a, - ) - end - end + define_method "test_#{name}_append_with_after_raises_if_match_is_after_cutoff" do + with_event_store(factory) do |event_store| + existing_event = Event.new(id: ids[0], type: "OrderPlaced") + event_store.append([existing_event]) - def test_append_with_after_raises_if_match_is_after_cutoff - with_event_store do |event_store| - existing_event = Event.new(id: ids[0], type: "OrderPlaced") - event_store.append([existing_event]) + assert_raises(AppendConditionViolated) do + event_store.append( + [Event.new(id: ids[1], type: "ShipmentScheduled")], + fail_if: event_store.read.of_type("OrderPlaced").after(0), + ) + end - assert_raises(AppendConditionViolated) do - event_store.append( - [Event.new(id: ids[1], type: "ShipmentScheduled")], - fail_if: event_store.read.of_type("OrderPlaced").after(0), - ) + assert_equal([existing_event], event_store.read.each.to_a) end - - assert_equal([existing_event], event_store.read.each.to_a) end - end - def test_tags_round_trip - with_event_store do |event_store| - event = - Event.new(id: ids[0], type: "OrderPlaced", tags: ["order_id:123"]) + define_method "test_#{name}_tags_round_trip" do + with_event_store(factory) do |event_store| + event = + Event.new(id: ids[0], type: "OrderPlaced", tags: ["order_id:123"]) - assert_equal([event], event_store.append([event]).read.each.to_a) + assert_equal([event], event_store.append([event]).read.each.to_a) + end end - end - def test_read_filters_after - with_event_store do |event_store| - events = [ - Event.new(id: ids[0], type: "OrderPlaced"), - Event.new(id: ids[1], type: "PriceChanged"), - ] - - assert_equal( - events.drop(1), - event_store.append(events).read.after(1).each.to_a, - ) + define_method "test_#{name}_read_filters_after" do + with_event_store(factory) do |event_store| + events = [ + Event.new(id: ids[0], type: "OrderPlaced"), + Event.new(id: ids[1], type: "PriceChanged"), + ] + + assert_equal( + events.drop(1), + event_store.append(events).read.after(1).each.to_a, + ) + end end - end - def test_read_filters_by_tags - with_event_store do |event_store| - events = [ - Event.new( - id: ids[0], - type: "OrderPlaced", - tags: %w[order_id:123 tenant_id:acme], - ), - Event.new( - id: ids[1], - type: "OrderPlaced", - tags: %w[order_id:456 tenant_id:acme], - ), - ] - - assert_equal( - events.take(1), - event_store - .append(events) - .read - .with_tag("order_id:123", "tenant_id:acme") - .each - .to_a, - ) + define_method "test_#{name}_read_filters_by_tags" do + with_event_store(factory) do |event_store| + events = [ + Event.new( + id: ids[0], + type: "OrderPlaced", + tags: %w[order_id:123 tenant_id:acme], + ), + Event.new( + id: ids[1], + type: "OrderPlaced", + tags: %w[order_id:456 tenant_id:acme], + ), + ] + + assert_equal( + events.take(1), + event_store + .append(events) + .read + .with_tag("order_id:123", "tenant_id:acme") + .each + .to_a, + ) + end end - end - def test_read_filters_by_type - with_event_store do |event_store| - events = [ - Event.new(id: ids[0], type: "OrderPlaced"), - Event.new(id: ids[1], type: "PriceChanged"), - ] - - assert_equal( - events.take(1), - event_store.append(events).read.of_type("OrderPlaced").each.to_a, - ) + define_method "test_#{name}_read_filters_by_type" do + with_event_store(factory) do |event_store| + events = [ + Event.new(id: ids[0], type: "OrderPlaced"), + Event.new(id: ids[1], type: "PriceChanged"), + ] + + assert_equal( + events.take(1), + event_store.append(events).read.of_type("OrderPlaced").each.to_a, + ) + end end - end - def test_read_filters_by_any_of_types - with_event_store do |event_store| - events = [ - Event.new(id: ids[0], type: "PriceChanged"), - Event.new(id: ids[1], type: "OrderPlaced"), - Event.new(id: ids[2], type: "OrderCancelled"), - ] - - assert_equal( - events.drop(1), - event_store - .append(events) - .read - .of_type("OrderPlaced", "OrderCancelled") - .each - .to_a, - ) + define_method "test_#{name}_read_filters_by_any_of_types" do + with_event_store(factory) do |event_store| + events = [ + Event.new(id: ids[0], type: "PriceChanged"), + Event.new(id: ids[1], type: "OrderPlaced"), + Event.new(id: ids[2], type: "OrderCancelled"), + ] + + assert_equal( + events.drop(1), + event_store + .append(events) + .read + .of_type("OrderPlaced", "OrderCancelled") + .each + .to_a, + ) + end end - end - def test_read_filters_by_type_and_tag_on_same_item - with_event_store do |event_store| - events = [ - Event.new(id: ids[0], type: "OrderPlaced", tags: ["order_id:123"]), - Event.new(id: ids[1], type: "OrderPlaced", tags: ["order_id:456"]), - Event.new(id: ids[2], type: "PriceChanged", tags: ["order_id:123"]), - ] - - assert_equal( - events.take(1), - event_store - .append(events) - .read - .of_type("OrderPlaced") - .with_tag("order_id:123") - .each - .to_a, - ) + define_method "test_#{name}_read_filters_by_type_and_tag_on_same_item" do + with_event_store(factory) do |event_store| + events = [ + Event.new(id: ids[0], type: "OrderPlaced", tags: ["order_id:123"]), + Event.new(id: ids[1], type: "OrderPlaced", tags: ["order_id:456"]), + Event.new(id: ids[2], type: "PriceChanged", tags: ["order_id:123"]), + ] + + assert_equal( + events.take(1), + event_store + .append(events) + .read + .of_type("OrderPlaced") + .with_tag("order_id:123") + .each + .to_a, + ) + end end - end - def test_read_or_combines_scopes_as_disjunction - with_event_store do |event_store| - events = [ - Event.new(id: ids[0], type: "OrderPlaced", tags: ["order_id:123"]), - Event.new(id: ids[1], type: "OrderPlaced", tags: ["order_id:456"]), - Event.new(id: ids[2], type: "PriceChanged", tags: ["order_id:999"]), - Event.new(id: ids[3], type: "InventoryAdjusted", tags: ["sku:A1"]), - Event.new( - id: ids[4], - type: "ShipmentScheduled", - tags: ["order_id:123"], - ), - ] - event_store.append(events) - - orders = - event_store.read.of_type("OrderPlaced").with_tag("order_id:123") - prices = event_store.read.of_type("PriceChanged") - - assert_equal(events.fetch_values(0, 2), (orders | prices).each.to_a) + define_method "test_#{name}_read_or_combines_scopes_as_disjunction" do + with_event_store(factory) do |event_store| + events = [ + Event.new(id: ids[0], type: "OrderPlaced", tags: ["order_id:123"]), + Event.new(id: ids[1], type: "OrderPlaced", tags: ["order_id:456"]), + Event.new(id: ids[2], type: "PriceChanged", tags: ["order_id:999"]), + Event.new(id: ids[3], type: "InventoryAdjusted", tags: ["sku:A1"]), + Event.new( + id: ids[4], + type: "ShipmentScheduled", + tags: ["order_id:123"], + ), + ] + event_store.append(events) + + orders = + event_store.read.of_type("OrderPlaced").with_tag("order_id:123") + prices = event_store.read.of_type("PriceChanged") + + assert_equal(events.fetch_values(0, 2), (orders | prices).each.to_a) + end end end @@ -207,11 +221,8 @@ module En57 def ids = @ids ||= Hash.new { |h, k| h[k] = SecureRandom.uuid_v7 } - def with_event_store = - yield( - EventStore.new( - Repository.new(PgAdapter.new(SERVER.url), JsonSerializer.new), - ) - ) + def with_event_store(factory) + yield(EventStore.new(Repository.new(factory.call, JsonSerializer.new))) + end end end diff --git a/test/test_sequel_adapter.rb b/test/test_sequel_adapter.rb new file mode 100644 index 0000000..44b2178 --- /dev/null +++ b/test/test_sequel_adapter.rb @@ -0,0 +1,83 @@ +# frozen_string_literal: true + +require "test_helper" + +module En57 + class TestSequelAdapter < Minitest::Test + cover SequelAdapter + + def test_with_connection_synchronizes_and_yields_connection + with_mock_adapter do |database, connection, adapter| + database.expect(:synchronize, :selected) do |&block| + block.call(connection) + true + end + connection.expect(:exec, :selected, ["SELECT 1"]) + + assert_equal :selected, + adapter.with_connection { |conn| conn.exec("SELECT 1") } + end + end + + def test_with_serializable_transaction_synchronizes_inside_transaction + with_mock_adapter do |database, connection, adapter| + database.expect(:transaction, :committed) do |options, &block| + assert_equal({ isolation: :serializable }, options) + block.call + true + end + database.expect(:synchronize, :written) do |&block| + block.call(connection) + true + end + connection.expect( + :exec_params, + :written, + ["SELECT append_events()", []], + ) + + assert_equal( + :committed, + adapter.with_serializable_transaction do |conn| + assert_equal :written, + conn.exec_params("SELECT append_events()", []) + end, + ) + end + end + + def test_with_serializable_transaction_unwraps_pg_errors + pg_error = PG::Error.new("boom") + sequel_error = Sequel::DatabaseError.new("wrapped") + sequel_error.wrapped_exception = pg_error + + raised = + assert_raises(PG::Error) do + with_mock_adapter do |database, _connection, adapter| + database.expect(:transaction, nil) do |options, &block| + assert_equal({ isolation: :serializable }, options) + block.call + true + end + database.expect(:synchronize, nil) { raise sequel_error } + + adapter.with_serializable_transaction { flunk "not yielded" } + end + end + + assert_same pg_error, raised + end + + private + + def with_mock_adapter + database = Minitest::Mock.new + connection = Minitest::Mock.new + + yield database, connection, SequelAdapter.new(database) + + database.verify + connection.verify + end + end +end diff --git a/test/test_stress.rb b/test/test_stress.rb index c5068e5..6650839 100644 --- a/test/test_stress.rb +++ b/test/test_stress.rb @@ -4,63 +4,64 @@ require "test_helper" module En57 class TestStress < IntegrationTest - def test_only_one_writer_can_consume_account_credits - with_event_store do |event_store| - event_store.append( - [ - Event.new( - type: "CreditsToppedUp", - data: { - amount: 100, - }, - tags: [account_tag], - ), - ], - ) + ADAPTERS.each do |adapter_name, factory| + define_method( + "test_only_one_writer_can_consume_account_credits_with_#{adapter_name}", + ) do + with_event_store(factory) do |event_store| + event_store.append( + [ + Event.new( + type: "CreditsToppedUp", + data: { + amount: 100, + }, + tags: [account_tag], + ), + ], + ) - barrier = Concurrent::CyclicBarrier.new(concurrency) - threads = - Array.new(concurrency) do - Thread.new do - with_event_store do |event_store_| - barrier.wait - account_scope = event_store_.read.with_tag(account_tag) + barrier = Concurrent::CyclicBarrier.new(concurrency) + threads = + Array.new(concurrency) do + Thread.new do + with_event_store(factory) do |event_store_| + barrier.wait + account_scope = event_store_.read.with_tag(account_tag) - if account_balance(account_scope) >= 100 - event_store_.append( - [ - Event.new( - type: "CreditsUsed", - data: { - amount: 100, - }, - tags: [account_tag], - ), - ], - fail_if: account_scope.of_type("CreditsUsed"), - ) + if account_balance(account_scope) >= 100 + event_store_.append( + [ + Event.new( + type: "CreditsUsed", + data: { + amount: 100, + }, + tags: [account_tag], + ), + ], + fail_if: account_scope.of_type("CreditsUsed"), + ) + end end + rescue AppendConditionViolated, PG::TRSerializationFailure end - rescue AppendConditionViolated, PG::TRSerializationFailure end - end - threads.each(&:join) + threads.each(&:join) - account_scope = event_store.read.with_tag(account_tag) + account_scope = event_store.read.with_tag(account_tag) - assert_equal 1, account_scope.of_type("CreditsUsed").each.count - assert_equal 0, account_balance(account_scope) + assert_equal 1, account_scope.of_type("CreditsUsed").each.count + assert_equal 0, account_balance(account_scope) + end end end private - def with_event_store = - yield( - EventStore.new( - Repository.new(PgAdapter.new(SERVER.url), JsonSerializer.new), - ) - ) + def with_event_store(factory) + yield EventStore.new(Repository.new(factory.call, JsonSerializer.new)) + end def concurrency = 8