diff --git a/backends/factos_cf/src/factos/factos_cf.gleam b/backends/factos_cf/src/factos/factos_cf.gleam index 879a1e8..f480f4a 100644 --- a/backends/factos_cf/src/factos/factos_cf.gleam +++ b/backends/factos_cf/src/factos/factos_cf.gleam @@ -347,7 +347,6 @@ pub fn dispatch( } } - fn append_with_condition( database: d1.Database, stream_name: String, @@ -392,10 +391,8 @@ fn append_events( |> promise.map(fn(result) { result |> result.map(fn(revision) { - let append = Append( - current_revision: revision, - position: factos.NoPosition, - ) + let append = + Append(current_revision: revision, position: factos.NoPosition) Dispatch(append:, events: []) }) }) @@ -433,10 +430,11 @@ fn decode_batch_result( case list.length(appended) == list.length(events) { True -> { let #(position, revision) = last_append_row(appended) - let append = Append( - current_revision: revision, - position: factos.SequencePosition(position), - ) + let append = + Append( + current_revision: revision, + position: factos.SequencePosition(position), + ) Ok(Dispatch( append: append, events: recorded_append_rows(stream_name, events, appended), diff --git a/backends/factos_cf/test/factos_cf_test.gleam b/backends/factos_cf/test/factos_cf_test.gleam index 7b91047..7fe584a 100644 --- a/backends/factos_cf/test/factos_cf_test.gleam +++ b/backends/factos_cf/test/factos_cf_test.gleam @@ -194,7 +194,6 @@ pub fn dispatch_context_rejects_changed_context_test() -> Promise(Nil) { promise.resolve(Nil) } - fn new_test_database() -> Promise(TestDatabase) { let worker = miniflare.worker( @@ -220,14 +219,12 @@ fn dispose(test_database: TestDatabase) -> Promise(Nil) { miniflare.dispose(test_database.miniflare) } - fn clear_events(database: d1.Database) -> Promise(Result(Nil, String)) { d1.prepare(database, "delete from factos_events") |> d1.run |> promise.map(fn(run_result) { run_result |> result.map(fn(_) { Nil }) }) } - fn error_to_string(error: factos_cf.Error(DomainError)) -> String { case error { factos_cf.DomainError(_) -> "domain error" @@ -238,7 +235,6 @@ fn error_to_string(error: factos_cf.Error(DomainError)) -> String { } } - fn reservation_decider() -> factos.Decider( Command, List(String), @@ -271,7 +267,6 @@ fn codec() -> factos_cf.EventCodec(Event, List(String)) { factos_cf.codec(encode:, decode:) } - fn encode(event: Event) -> factos_cf.Proposed(Event) { case event { Reserved(name) -> diff --git a/backends/factos_kurrentdb_erlang/src/factos/factos_kurrentdb_erlang.gleam b/backends/factos_kurrentdb_erlang/src/factos/factos_kurrentdb_erlang.gleam index e43bdb9..4b38aba 100644 --- a/backends/factos_kurrentdb_erlang/src/factos/factos_kurrentdb_erlang.gleam +++ b/backends/factos_kurrentdb_erlang/src/factos/factos_kurrentdb_erlang.gleam @@ -58,7 +58,10 @@ pub type Dispatch(event) { /// /// `append` is the KurrentDB append response. `events` are the committed events /// observed after the append, suitable for pure Factos reactors. - Dispatch(append: append_to_stream.Append, events: List(factos.Recorded(event))) + Dispatch( + append: append_to_stream.Append, + events: List(factos.Recorded(event)), + ) } pub type Error(domain_error, decode_error) { diff --git a/backends/factos_kurrentdb_erlang/test/factos_kurrentdb_erlang_test.gleam b/backends/factos_kurrentdb_erlang/test/factos_kurrentdb_erlang_test.gleam index 90a2e16..eb17d10 100644 --- a/backends/factos_kurrentdb_erlang/test/factos_kurrentdb_erlang_test.gleam +++ b/backends/factos_kurrentdb_erlang/test/factos_kurrentdb_erlang_test.gleam @@ -60,9 +60,10 @@ pub fn dispatch_stream_handles_many_events_integration_test() { type_: factos.event_type(event_type), ) let reactor = factos.reactor(react: fn(recorded) { [recorded.event] }) - assert factos.react_all(reactor: reactor, events: dispatch.events) == [ - Incremented(100), - ] + assert factos.react_all(reactor: reactor, events: dispatch.events) + == [ + Incremented(100), + ] let assert Ok(loaded) = factos_kurrentdb_erlang.load_stream( @@ -100,9 +101,10 @@ pub fn read_context_handles_many_streams_integration_test() { type_: factos.event_type(event_type), ) let reactor = factos.reactor(react: fn(recorded) { [recorded.event] }) - assert factos.react_all(reactor: reactor, events: dispatch.events) == [ - Incremented(1), - ] + assert factos.react_all(reactor: reactor, events: dispatch.events) + == [ + Incremented(1), + ] let assert Ok(context) = factos_kurrentdb_erlang.read_context( diff --git a/backends/factos_pog/README.md b/backends/factos_pog/README.md index ba32aed..a358dd5 100644 --- a/backends/factos_pog/README.md +++ b/backends/factos_pog/README.md @@ -20,19 +20,25 @@ delivery on top of the committed records returned by dispatch. [dependencies] factos = ">= 1.0.0 and < 2.0.0" factos_pog = ">= 1.0.0 and < 2.0.0" +gleam_erlang = ">= 1.0.0 and < 2.0.0" pog = ">= 4.1.0 and < 5.0.0" ``` ## Set up the schema -Start a `pog` pool in your application supervision tree, get a named connection, -and run the migration: +The backend ships its schema as `priv/migrations.sql`. In an Erlang-target +application, locate that file through the package `priv` directory and run it +with your migration tool before dispatching commands: ```gleam -let connection = pog.named_connection(pool_name) -let assert Ok(Nil) = factos_pog.migrate(connection) +import gleam/erlang/application + +let assert Ok(priv_directory) = application.priv_directory("factos_pog") +let migration_path = priv_directory <> "/migrations.sql" ``` +`factos_pog.migrate` still exists for v1 compatibility, but it is deprecated. + The migration creates: - `factos_events`: append-only event rows; diff --git a/backends/factos_pog/gleam.toml b/backends/factos_pog/gleam.toml index 6eea8d7..7c9da92 100644 --- a/backends/factos_pog/gleam.toml +++ b/backends/factos_pog/gleam.toml @@ -26,4 +26,7 @@ gleam_time = ">= 1.0.0 and < 2.0.0" [dev_dependencies] gleeunit = ">= 1.0.0 and < 2.0.0" gleam_erlang = ">= 1.0.0 and < 2.0.0" -global_value = ">= 1.0.0 and < 2.0.0" +testcontainer = ">= 1.0.2 and < 2.0.0" +testcontainer_formulas = ">= 1.0.0 and < 2.0.0" +simplifile = ">= 2.5.0 and < 3.0.0" +gleam_otp = ">= 1.2.0 and < 2.0.0" diff --git a/backends/factos_pog/manifest.toml b/backends/factos_pog/manifest.toml index 908ac11..01aa505 100644 --- a/backends/factos_pog/manifest.toml +++ b/backends/factos_pog/manifest.toml @@ -8,25 +8,34 @@ packages = [ { name = "backoff", version = "1.1.6", build_tools = ["rebar3"], requirements = [], otp_app = "backoff", source = "hex", outer_checksum = "CF0CFFF8995FB20562F822E5CC47D8CCF664C5ECDC26A684CBE85C225F9D7C39" }, + { name = "cowl", version = "1.1.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "cowl", source = "hex", outer_checksum = "7849E7C789D7228243A4253138FC883720A0BB44AEF406102328CADC64C3CA2B" }, + { name = "envie", version = "1.2.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "envie", source = "hex", outer_checksum = "E7EBA39310F32A40BF3EDDD7CD9C7A2BC289909983D357411C22873415BC322A" }, { name = "exception", version = "2.1.1", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "exception", source = "hex", outer_checksum = "6BDEA95248093599391C3B5DF1835C5C6A86C353C2F99CE539B450E3432FE117" }, { name = "factos", version = "1.0.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "factos", source = "hex", outer_checksum = "94ADCBF2A7DAE0C39C68710C4A150A0286DBA448A293600090C38CB01DCF8127" }, + { name = "filepath", version = "1.1.2", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "filepath", source = "hex", outer_checksum = "B06A9AF0BF10E51401D64B98E4B627F1D2E48C154967DA7AF4D0914780A6D40A" }, { name = "gleam_erlang", version = "1.3.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleam_erlang", source = "hex", outer_checksum = "1124AD3AA21143E5AF0FC5CF3D9529F6DB8CA03E43A55711B60B6B7B3874375C" }, + { name = "gleam_json", version = "3.1.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleam_json", source = "hex", outer_checksum = "44FDAA8847BE8FC48CA7A1C089706BD54BADCC4C45B237A992EDDF9F2CDB2836" }, { name = "gleam_otp", version = "1.2.0", build_tools = ["gleam"], requirements = ["gleam_erlang", "gleam_stdlib"], otp_app = "gleam_otp", source = "hex", outer_checksum = "BA6A294E295E428EC1562DC1C11EA7530DCB981E8359134BEABC8493B7B2258E" }, { name = "gleam_stdlib", version = "1.0.3", build_tools = ["gleam"], requirements = [], otp_app = "gleam_stdlib", source = "hex", outer_checksum = "1F543AFBA5D33DA493E6087F4E4C4F20D899411343512686C98A8ABB2963CF22" }, { name = "gleam_time", version = "1.8.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleam_time", source = "hex", outer_checksum = "533D8723774D61AD4998324F5DD1DABDCDBFABAFB9E87CB5D03C6955448FC97D" }, { name = "gleeunit", version = "1.11.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleeunit", source = "hex", outer_checksum = "EC31ABA74256AEA531EDF8169931D775BBB384FED0A8A1BDC4DD9354E3E21826" }, - { name = "global_value", version = "1.0.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "global_value", source = "hex", outer_checksum = "23F74C91A7B819C43ABCCBF49DAD5BB8799D81F2A3736BA9A534BD47F309FF4F" }, { name = "opentelemetry_api", version = "1.5.0", build_tools = ["rebar3", "mix"], requirements = [], otp_app = "opentelemetry_api", source = "hex", outer_checksum = "F53EC8A1337AE4A487D43AC89DA4BD3A3C99DDF576655D071DEED8B56A2D5DDA" }, { name = "pg_types", version = "0.6.0", build_tools = ["rebar3"], requirements = [], otp_app = "pg_types", source = "hex", outer_checksum = "9949A4849DD13408FA249AB7B745E0D2DFDB9532AEE2B9722326E33CD082A778" }, { name = "pgo", version = "0.20.0", build_tools = ["rebar3"], requirements = ["backoff", "opentelemetry_api", "pg_types"], otp_app = "pgo", source = "hex", outer_checksum = "2F11E6649CEB38E569EF56B16BE1D04874AE5B11A02867080A2817CE423C683B" }, { name = "pog", version = "4.1.0", build_tools = ["gleam"], requirements = ["exception", "gleam_erlang", "gleam_otp", "gleam_stdlib", "gleam_time", "pgo"], otp_app = "pog", source = "hex", outer_checksum = "E4AFBA39A5FAA2E77291836C9683ADE882E65A06AB28CA7D61AE7A3AD61EBBD5" }, + { name = "simplifile", version = "2.5.0", build_tools = ["gleam"], requirements = ["filepath", "gleam_stdlib"], otp_app = "simplifile", source = "hex", outer_checksum = "6C72DCCDF25C38A5931740B30E823969F33106831FD1637719B5EDBCA30027A4" }, + { name = "testcontainer", version = "1.0.2", build_tools = ["gleam"], requirements = ["cowl", "envie", "gleam_erlang", "gleam_json", "gleam_stdlib"], otp_app = "testcontainer", source = "hex", outer_checksum = "784768485ED2380AA543A0CC3F06F7A368B0DD209102C57E800E0A87E1D2FC81" }, + { name = "testcontainer_formulas", version = "1.0.0", build_tools = ["gleam"], requirements = ["cowl", "gleam_stdlib", "testcontainer"], otp_app = "testcontainer_formulas", source = "hex", outer_checksum = "F9A86A2F8400A0C72FE98F56EF5B3FD1CE10F0A63D968A1C57EA0087A3E5802B" }, ] [requirements] factos = { version = ">= 1.0.0 and < 2.0.0" } gleam_erlang = { version = ">= 1.0.0 and < 2.0.0" } +gleam_otp = { version = ">= 1.2.0 and < 2.0.0" } gleam_stdlib = { version = ">= 1.0.0 and < 2.0.0" } gleam_time = { version = ">= 1.0.0 and < 2.0.0" } gleeunit = { version = ">= 1.0.0 and < 2.0.0" } -global_value = { version = ">= 1.0.0 and < 2.0.0" } pog = { version = ">= 4.1.0 and < 5.0.0" } +simplifile = { version = ">= 2.5.0 and < 3.0.0" } +testcontainer = { version = ">= 1.0.2 and < 2.0.0" } +testcontainer_formulas = { version = ">= 1.0.0 and < 2.0.0" } diff --git a/backends/factos_pog/priv/migrations.sql b/backends/factos_pog/priv/migrations.sql new file mode 100644 index 0000000..9c18b02 --- /dev/null +++ b/backends/factos_pog/priv/migrations.sql @@ -0,0 +1,37 @@ +create table if not exists factos_events ( + position bigint generated always as identity primary key, + id text not null, + stream text not null, + revision integer not null, + type text not null, + version integer not null, + tags text not null, + metadata text not null, + data bytea not null, + unique(stream, revision) +); + +create index if not exists factos_events_stream_revision + on factos_events(stream, revision); + +create index if not exists factos_events_position + on factos_events(position); + +create table if not exists factos_event_tags ( + position bigint not null references factos_events(position) on delete cascade, + tag text not null, + primary key(position, tag) +); + +create index if not exists factos_events_type_position + on factos_events(type, position); + +create index if not exists factos_event_tags_tag_position + on factos_event_tags(tag, position); + +insert into factos_event_tags(position, tag) +select factos_events.position, split_tags.tag +from factos_events +cross join lateral regexp_split_to_table(factos_events.tags, E'\n') as split_tags(tag) +where split_tags.tag <> '' +on conflict do nothing; diff --git a/backends/factos_pog/src/factos/factos_pog.gleam b/backends/factos_pog/src/factos/factos_pog.gleam index 2b581fd..da6d16f 100644 --- a/backends/factos_pog/src/factos/factos_pog.gleam +++ b/backends/factos_pog/src/factos/factos_pog.gleam @@ -125,11 +125,15 @@ pub fn codec( EventCodec(encode:, decode:) } +/// Deprecated: read `priv/migrations.sql` from the `factos_pog` application +/// with `gleam/erlang/application.priv_directory` and run it with your +/// application's migration tool instead. /// The schema is an append-only `factos_events` table with a global identity /// `position`, per-stream `revision`, event `type`, newline-encoded `tags`, and /// opaque `data` bytes. Queryable tags are also mirrored into /// `factos_event_tags`, which gives event-type/tag context reads indexed SQL /// plans instead of loading the whole event table into the application. +@deprecated("Use the SQL file at factos_pog/priv/migrations.sql instead.") pub fn migrate(connection: pog.Connection) -> Result(Nil, Error(_)) { // `pog` uses prepared statements, and PostgreSQL does not allow multiple SQL // commands in one prepared statement. Keep migrations split into individual @@ -414,10 +418,11 @@ fn append_stream_events( ) -> Result(Dispatch(event), Error(domain_error)) { case events { [] -> { - let append = Append( - current_revision: revision_to_int(expected), - position: factos.NoPosition, - ) + let append = + Append( + current_revision: revision_to_int(expected), + position: factos.NoPosition, + ) Ok(Dispatch(append:, events: [])) } [_, ..] -> { @@ -483,17 +488,18 @@ fn insert_events( [position, ..] -> factos.SequencePosition(position) [] -> position } - let recorded = factos.Recorded( - id: id, - stream: stream_name, - revision: revision, - position: position, - type_: type_, - version: version, - tags: tags, - metadata: metadata, - event: event, - ) + let recorded = + factos.Recorded( + id: id, + stream: stream_name, + revision: revision, + position: position, + type_: type_, + version: version, + tags: tags, + metadata: metadata, + event: event, + ) use _ <- result.try(insert_event_tags(connection, position, tags)) insert_events( connection, diff --git a/backends/factos_pog/test/factos_pog_test.gleam b/backends/factos_pog/test/factos_pog_test.gleam index 1e0d1a1..37a4541 100644 --- a/backends/factos_pog/test/factos_pog_test.gleam +++ b/backends/factos_pog/test/factos_pog_test.gleam @@ -1,14 +1,20 @@ import factos import factos/factos_pog import gleam/bit_array +import gleam/erlang/application import gleam/erlang/process import gleam/int import gleam/list import gleam/option.{Some} +import gleam/otp/actor import gleam/result +import gleam/string import gleeunit -import global_value import pog +import simplifile +import testcontainer +import testcontainer/error as testcontainer_error +import testcontainer_formulas/postgres pub fn main() -> Nil { gleeunit.main() @@ -43,164 +49,194 @@ type CounterState { CounterState(total: Int) } -type TestGlobalData { - TestGlobalData(connection: pog.Connection) +pub type Timeout(a) { + Timeout(time: Int, function: fn() -> a) } +pub fn dispatch_alias_persists_events_test_() -> Timeout(Nil) { + use <- Timeout(120) + let assert Ok(Nil) = + with_test_connection(fn(connection) { + reset_schema(connection) -pub fn dispatch_alias_persists_events_test() { - let TestGlobalData(connection) = global_data() - reset_schema(connection) - - let assert Ok(dispatch) = - factos_pog.dispatch( - connection, - stream: "user-renata", - decider: decider(), - codec: codec(), - command: RegisterUser("renata"), - ) + let assert Ok(dispatch) = + factos_pog.dispatch( + connection, + stream: "user-renata", + decider: decider(), + codec: codec(), + command: RegisterUser("renata"), + ) - let assert factos_pog.Append( - current_revision: 0, - position: factos.SequencePosition(_), - ) = dispatch.append - let assert [recorded] = dispatch.events - assert_user_recorded( - recorded, - stream: "user-renata", - revision: 0, - position: dispatch.append.position, - username: "renata", - ) - let reactor = factos.reactor(react: fn(recorded) { [recorded.event] }) - assert factos.react_all(reactor: reactor, events: dispatch.events) == [ - UserRegistered("renata"), - ] - - let assert Ok(loaded) = - factos_pog.load_stream( - connection, - stream: "user-renata", - decider: decider(), - codec: codec(), - ) + let assert factos_pog.Append( + current_revision: 0, + position: factos.SequencePosition(_), + ) = dispatch.append + let assert [recorded] = dispatch.events + assert_user_recorded( + recorded, + stream: "user-renata", + revision: 0, + position: dispatch.append.position, + username: "renata", + ) + let reactor = factos.reactor(react: fn(recorded) { [recorded.event] }) + assert factos.react_all(reactor: reactor, events: dispatch.events) + == [ + UserRegistered("renata"), + ] + + let assert Ok(loaded) = + factos_pog.load_stream( + connection, + stream: "user-renata", + decider: decider(), + codec: codec(), + ) - assert loaded.state == Taken - assert loaded.revision == factos.CurrentRevision(0) + assert loaded.state == Taken + assert loaded.revision == factos.CurrentRevision(0) + Nil + }) + Nil } -pub fn dispatch_with_query_filters_before_decoding_unknown_events_test() { - let TestGlobalData(connection) = global_data() - reset_schema(connection) - insert_unknown_event(connection) - - let query = username_query("renata") +pub fn dispatch_with_query_filters_before_decoding_unknown_events_test_() -> Timeout( + Nil, +) { + use <- Timeout(120) + let assert Ok(Nil) = + with_test_connection(fn(connection) { + reset_schema(connection) + insert_unknown_event(connection) - let assert Ok(dispatch) = - factos_pog.dispatch_with_query( - connection, - stream: "user-renata", - query: query, - decider: decider(), - codec: codec(), - command: RegisterUser("renata"), - ) + let query = username_query("renata") - let assert factos_pog.Append( - current_revision: 0, - position: factos.SequencePosition(_), - ) = dispatch.append - let assert [recorded] = dispatch.events - assert_user_recorded( - recorded, - stream: "user-renata", - revision: 0, - position: dispatch.append.position, - username: "renata", - ) - let reactor = factos.reactor(react: fn(recorded) { [recorded.event] }) - assert factos.react_all(reactor: reactor, events: dispatch.events) == [ - UserRegistered("renata"), - ] - - let assert Ok(context) = - factos_pog.read_context( - connection, - query: query, - decider: decider(), - codec: codec(), - ) + let assert Ok(dispatch) = + factos_pog.dispatch_with_query( + connection, + stream: "user-renata", + query: query, + decider: decider(), + codec: codec(), + command: RegisterUser("renata"), + ) - let assert [event] = context.events - assert event.event == UserRegistered("renata") - assert context.state == Taken - assert context.position != factos.NoPosition -} - -pub fn dispatch_with_query_handles_many_streams_test() { - let TestGlobalData(connection) = global_data() - reset_schema(connection) - - let query = - factos.query([ - factos.query_item(types: [factos.event_type("Incremented")], tags: [ - factos.tag("counter:load"), - ]), - ]) - - let assert Ok(dispatch) = dispatch_counter_context_many(connection, query, 25) - let assert factos_pog.Append( - current_revision: 0, - position: factos.SequencePosition(_), - ) = dispatch.append - let assert [recorded] = dispatch.events - assert_counter_recorded( - recorded, - stream: "counter-context-1", - revision: 0, - position: dispatch.append.position, - value: 25, - type_: factos.event_type("Incremented"), - ) - let reactor = factos.reactor(react: fn(recorded) { [recorded.event] }) - assert factos.react_all(reactor: reactor, events: dispatch.events) == [ - Incremented(25), - ] - - let assert Ok(context) = - factos_pog.read_context( - connection, - query: query, - decider: counter_decider(), - codec: counter_codec(), - ) + let assert factos_pog.Append( + current_revision: 0, + position: factos.SequencePosition(_), + ) = dispatch.append + let assert [recorded] = dispatch.events + assert_user_recorded( + recorded, + stream: "user-renata", + revision: 0, + position: dispatch.append.position, + username: "renata", + ) + let reactor = factos.reactor(react: fn(recorded) { [recorded.event] }) + assert factos.react_all(reactor: reactor, events: dispatch.events) + == [ + UserRegistered("renata"), + ] + + let assert Ok(context) = + factos_pog.read_context( + connection, + query: query, + decider: decider(), + codec: codec(), + ) - assert context.state == CounterState(25) - assert list.length(context.events) == 25 + let assert [event] = context.events + assert event.event == UserRegistered("renata") + assert context.state == Taken + assert context.position != factos.NoPosition + Nil + }) + Nil } +pub fn dispatch_with_query_handles_many_streams_test_() -> Timeout(Nil) { + use <- Timeout(120) + let assert Ok(Nil) = + with_test_connection(fn(connection) { + reset_schema(connection) + + let query = + factos.query([ + factos.query_item(types: [factos.event_type("Incremented")], tags: [ + factos.tag("counter:load"), + ]), + ]) + + let assert Ok(dispatch) = + dispatch_counter_context_many(connection, query, 25) + let assert factos_pog.Append( + current_revision: 0, + position: factos.SequencePosition(_), + ) = dispatch.append + let assert [recorded] = dispatch.events + assert_counter_recorded( + recorded, + stream: "counter-context-1", + revision: 0, + position: dispatch.append.position, + value: 25, + type_: factos.event_type("Incremented"), + ) + let reactor = factos.reactor(react: fn(recorded) { [recorded.event] }) + assert factos.react_all(reactor: reactor, events: dispatch.events) + == [ + Incremented(25), + ] + + let assert Ok(context) = + factos_pog.read_context( + connection, + query: query, + decider: counter_decider(), + codec: counter_codec(), + ) -fn global_data() -> TestGlobalData { - global_value.create_with_unique_name("factos_pog_test.global.data", fn() { - TestGlobalData(connection: start_test_connection()) - }) + assert context.state == CounterState(25) + assert list.length(context.events) == 25 + Nil + }) + Nil } -fn start_test_connection() -> pog.Connection { +fn with_test_connection( + body: fn(pog.Connection) -> Nil, +) -> Result(Nil, testcontainer_error.Error) { + use postgres_container <- testcontainer.with_formula( + postgres.new() |> postgres.formula(), + ) + let #(pool_pid, connection) = start_test_connection(postgres_container) + body(connection) + process.send_exit(pool_pid) + process.sleep(100) + Ok(Nil) +} + +fn start_test_connection( + postgres_container: postgres.PostgresContainer, +) -> #(process.Pid, pog.Connection) { + let postgres.PostgresContainer(host:, port:, database:, username:, ..) = + postgres_container let pool_name = process.new_name("factos_pog_test") let config = pog.default_config(pool_name) - |> pog.host("127.0.0.1") - |> pog.port(55432) - |> pog.database("factos_pog") - |> pog.user("postgres") + |> pog.host(host) + |> pog.port(port) + |> pog.database(database) + |> pog.user(username) |> pog.password(Some("postgres")) |> pog.ssl(pog.SslDisabled) - let assert Ok(_) = pog.start(config) + let assert Ok(actor.Started(pid:, ..)) = pog.start(config) process.sleep(100) - pog.named_connection(pool_name) + #(pid, pog.named_connection(pool_name)) } fn reset_schema(connection: pog.Connection) -> Nil { @@ -210,8 +246,24 @@ fn reset_schema(connection: pog.Connection) -> Nil { let assert Ok(_) = pog.query("drop table if exists factos_events") |> pog.execute(on: connection) - let assert Ok(Nil) = factos_pog.migrate(connection) - Nil + execute_migration_file(connection) +} + +fn execute_migration_file(connection: pog.Connection) -> Nil { + let assert Ok(priv_directory) = application.priv_directory("factos_pog") + let assert Ok(sql) = simplifile.read(priv_directory <> "/migrations.sql") + + sql + |> string.split(";") + |> list.each(fn(statement) { + case string.trim(statement) { + "" -> Nil + statement -> { + let assert Ok(_) = pog.query(statement) |> pog.execute(on: connection) + Nil + } + } + }) } fn insert_unknown_event(connection: pog.Connection) -> Nil { @@ -386,10 +438,7 @@ fn counter_evolve(state: CounterState, event: CounterEvent) -> CounterState { } fn counter_codec() -> factos_pog.EventCodec(CounterEvent) { - factos_pog.codec( - encode: encode_counter_event, - decode: decode_counter_event, - ) + factos_pog.codec(encode: encode_counter_event, decode: decode_counter_event) } fn encode_counter_event( diff --git a/backends/factos_sqlight/gleam.toml b/backends/factos_sqlight/gleam.toml index e2faf16..f5d57c5 100644 --- a/backends/factos_sqlight/gleam.toml +++ b/backends/factos_sqlight/gleam.toml @@ -14,3 +14,5 @@ sqlight = ">= 1.1.0 and < 2.0.0" [dev_dependencies] gleeunit = ">= 1.0.0 and < 2.0.0" +gleam_erlang = ">= 1.3.0 and < 2.0.0" +simplifile = ">= 2.5.0 and < 3.0.0" diff --git a/backends/factos_sqlight/manifest.toml b/backends/factos_sqlight/manifest.toml index a41f74c..254d8d9 100644 --- a/backends/factos_sqlight/manifest.toml +++ b/backends/factos_sqlight/manifest.toml @@ -9,13 +9,18 @@ packages = [ { name = "esqlite", version = "0.9.0", build_tools = ["rebar3"], requirements = [], otp_app = "esqlite", source = "hex", outer_checksum = "CCF72258A4EE152EC7AD92AA9A03552EB6CA1B06B65C93AD5B6E55C302E05855" }, { name = "factos", version = "1.0.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], source = "local", path = "../.." }, + { name = "filepath", version = "1.1.2", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "filepath", source = "hex", outer_checksum = "B06A9AF0BF10E51401D64B98E4B627F1D2E48C154967DA7AF4D0914780A6D40A" }, + { name = "gleam_erlang", version = "1.3.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleam_erlang", source = "hex", outer_checksum = "1124AD3AA21143E5AF0FC5CF3D9529F6DB8CA03E43A55711B60B6B7B3874375C" }, { name = "gleam_stdlib", version = "1.0.3", build_tools = ["gleam"], requirements = [], otp_app = "gleam_stdlib", source = "hex", outer_checksum = "1F543AFBA5D33DA493E6087F4E4C4F20D899411343512686C98A8ABB2963CF22" }, { name = "gleeunit", version = "1.11.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleeunit", source = "hex", outer_checksum = "EC31ABA74256AEA531EDF8169931D775BBB384FED0A8A1BDC4DD9354E3E21826" }, + { name = "simplifile", version = "2.5.0", build_tools = ["gleam"], requirements = ["filepath", "gleam_stdlib"], otp_app = "simplifile", source = "hex", outer_checksum = "6C72DCCDF25C38A5931740B30E823969F33106831FD1637719B5EDBCA30027A4" }, { name = "sqlight", version = "1.1.0", build_tools = ["gleam"], requirements = ["esqlite", "gleam_stdlib"], otp_app = "sqlight", source = "hex", outer_checksum = "ECA1A4B45C35EB9EFCEEB7FAAC7BF5D8B2C777A7C1FC8A9C12CB67D54CED42E7" }, ] [requirements] factos = { path = "../.." } +gleam_erlang = { version = ">= 1.3.0 and < 2.0.0" } gleam_stdlib = { version = ">= 1.0.0 and < 2.0.0" } gleeunit = { version = ">= 1.0.0 and < 2.0.0" } +simplifile = { version = ">= 2.5.0 and < 3.0.0" } sqlight = { version = ">= 1.1.0 and < 2.0.0" } diff --git a/backends/factos_sqlight/priv/migrations.sql b/backends/factos_sqlight/priv/migrations.sql new file mode 100644 index 0000000..77c96ec --- /dev/null +++ b/backends/factos_sqlight/priv/migrations.sql @@ -0,0 +1,18 @@ +create table if not exists factos_events ( + position integer primary key autoincrement, + id text not null, + stream text not null, + revision integer not null, + type text not null, + version integer not null, + tags text not null, + metadata text not null default '', + data blob not null, + unique(stream, revision) +); + +create index if not exists factos_events_stream_revision + on factos_events(stream, revision); + +create index if not exists factos_events_position + on factos_events(position); diff --git a/backends/factos_sqlight/src/factos/factos_sqlight.gleam b/backends/factos_sqlight/src/factos/factos_sqlight.gleam index e28d3e6..c5ccdfd 100644 --- a/backends/factos_sqlight/src/factos/factos_sqlight.gleam +++ b/backends/factos_sqlight/src/factos/factos_sqlight.gleam @@ -104,12 +104,16 @@ pub type Error(domain_error, decode_error) { AppendConditionFailed(factos.AppendCondition) } +/// Deprecated: read `priv/migrations.sql` from the `factos_sqlight` application +/// with `gleam/erlang/application.priv_directory` and run it with your +/// application's migration tool instead. /// Create or update the SQLite schema required by this backend. /// /// The schema is an append-only `factos_events` table with a global autoincrement /// `position`, per-stream `revision`, event `type`, newline-encoded `tags`, and /// opaque `data` bytes. It also creates indexes for stream/revision reads and /// position-based context checks. +@deprecated("Use the SQL file at factos_sqlight/priv/migrations.sql instead.") pub fn migrate(connection: sqlight.Connection) -> Result(Nil, Error(_, _)) { sqlight.exec( " @@ -348,10 +352,11 @@ fn append_stream_events( ) -> Result(Dispatch(event), Error(domain_error, decode_error)) { case events { [] -> { - let append = Append( - current_revision: revision_to_int(expected), - position: factos.NoPosition, - ) + let append = + Append( + current_revision: revision_to_int(expected), + position: factos.NoPosition, + ) Ok(Dispatch(append:, events: [])) } [_, ..] -> { @@ -419,17 +424,18 @@ fn insert_events( [position, ..] -> factos.SequencePosition(position) [] -> position } - let recorded = factos.Recorded( - id: id, - stream: stream_name, - revision: revision, - position: position, - type_: type_, - version: version, - tags: tags, - metadata: metadata, - event: event, - ) + let recorded = + factos.Recorded( + id: id, + stream: stream_name, + revision: revision, + position: position, + type_: type_, + version: version, + tags: tags, + metadata: metadata, + event: event, + ) insert_events( connection, stream_name, diff --git a/backends/factos_sqlight/test/factos_sqlight_test.gleam b/backends/factos_sqlight/test/factos_sqlight_test.gleam index 7344078..372365e 100644 --- a/backends/factos_sqlight/test/factos_sqlight_test.gleam +++ b/backends/factos_sqlight/test/factos_sqlight_test.gleam @@ -1,10 +1,12 @@ import factos import factos/factos_sqlight import gleam/bit_array +import gleam/erlang/application import gleam/int import gleam/list import gleam/result import gleeunit +import simplifile import sqlight pub fn main() -> Nil { @@ -47,7 +49,7 @@ type CounterState { pub fn dispatch_stream_persists_events_test() { use connection <- sqlight.with_connection(":memory:") - let assert Ok(Nil) = factos_sqlight.migrate(connection) + execute_migration_file(connection) let assert Ok(dispatch) = factos_sqlight.dispatch_stream( @@ -71,9 +73,10 @@ pub fn dispatch_stream_persists_events_test() { username: "renata", ) let reactor = factos.reactor(react: fn(recorded) { [recorded.event] }) - assert factos.react_all(reactor: reactor, events: dispatch.events) == [ - UserRegistered("renata"), - ] + assert factos.react_all(reactor: reactor, events: dispatch.events) + == [ + UserRegistered("renata"), + ] let assert Ok(loaded) = factos_sqlight.load_stream( @@ -89,7 +92,7 @@ pub fn dispatch_stream_persists_events_test() { pub fn dispatch_stream_handles_many_events_test() { use connection <- sqlight.with_connection(":memory:") - let assert Ok(Nil) = factos_sqlight.migrate(connection) + execute_migration_file(connection) let assert Ok(dispatch) = dispatch_counter_stream_many(connection, 250) let assert factos_sqlight.Append( @@ -105,9 +108,10 @@ pub fn dispatch_stream_handles_many_events_test() { value: 250, ) let reactor = factos.reactor(react: fn(recorded) { [recorded.event] }) - assert factos.react_all(reactor: reactor, events: dispatch.events) == [ - Incremented(250), - ] + assert factos.react_all(reactor: reactor, events: dispatch.events) + == [ + Incremented(250), + ] let assert Ok(loaded) = factos_sqlight.load_stream( @@ -124,7 +128,7 @@ pub fn dispatch_stream_handles_many_events_test() { pub fn dispatch_context_handles_many_streams_test() { use connection <- sqlight.with_connection(":memory:") - let assert Ok(Nil) = factos_sqlight.migrate(connection) + execute_migration_file(connection) let query = factos.query([ @@ -133,7 +137,8 @@ pub fn dispatch_context_handles_many_streams_test() { ]), ]) - let assert Ok(dispatch) = dispatch_counter_context_many(connection, query, 100) + let assert Ok(dispatch) = + dispatch_counter_context_many(connection, query, 100) let assert factos_sqlight.Append( current_revision: 0, position: factos.SequencePosition(_), @@ -147,9 +152,10 @@ pub fn dispatch_context_handles_many_streams_test() { value: 100, ) let reactor = factos.reactor(react: fn(recorded) { [recorded.event] }) - assert factos.react_all(reactor: reactor, events: dispatch.events) == [ - Incremented(100), - ] + assert factos.react_all(reactor: reactor, events: dispatch.events) + == [ + Incremented(100), + ] let assert Ok(context) = factos_sqlight.read_context( @@ -164,6 +170,13 @@ pub fn dispatch_context_handles_many_streams_test() { assert context.position != factos.NoPosition } +fn execute_migration_file(connection: sqlight.Connection) -> Nil { + let assert Ok(priv_directory) = application.priv_directory("factos_sqlight") + let assert Ok(sql) = simplifile.read(priv_directory <> "/migrations.sql") + let assert Ok(Nil) = sqlight.exec(sql, on: connection) + Nil +} + fn decider() -> factos.Decider(Command, State, Event, DomainError) { factos.decider(initial: Available, decide:, evolve:) } @@ -237,7 +250,10 @@ fn assert_user_recorded( fn dispatch_counter_stream_many( connection: sqlight.Connection, remaining: Int, -) -> Result(factos_sqlight.Dispatch(CounterEvent), factos_sqlight.Error(Nil, DecodeError)) { +) -> Result( + factos_sqlight.Dispatch(CounterEvent), + factos_sqlight.Error(Nil, DecodeError), +) { case remaining { 0 -> factos_sqlight.dispatch_stream( @@ -269,7 +285,10 @@ fn dispatch_counter_context_many( connection: sqlight.Connection, query: factos.Query, remaining: Int, -) -> Result(factos_sqlight.Dispatch(CounterEvent), factos_sqlight.Error(Nil, DecodeError)) { +) -> Result( + factos_sqlight.Dispatch(CounterEvent), + factos_sqlight.Error(Nil, DecodeError), +) { case remaining { 0 -> factos_sqlight.dispatch_context( diff --git a/docs/postgresql-backend.md b/docs/postgresql-backend.md index 84d9152..a40f703 100644 --- a/docs/postgresql-backend.md +++ b/docs/postgresql-backend.md @@ -35,13 +35,19 @@ and applications decide how reactors/effect delivery should run. ## Schema -Run migrations once before dispatching commands: +The backend ships its schema as `priv/migrations.sql`. Run that SQL once before +dispatching commands. In an Erlang-target application, locate the file through +the package `priv` directory: ```gleam -let connection = pog.named_connection(pool_name) -let assert Ok(Nil) = factos_pog.migrate(connection) +import gleam/erlang/application + +let assert Ok(priv_directory) = application.priv_directory("factos_pog") +let migration_path = priv_directory <> "/migrations.sql" ``` +`factos_pog.migrate` still exists for v1 compatibility, but it is deprecated. + The backend creates two tables. `factos_events` is the append-only log: diff --git a/test/factos_test.gleam b/test/factos_test.gleam index 35f625e..1f1d969 100644 --- a/test/factos_test.gleam +++ b/test/factos_test.gleam @@ -201,20 +201,27 @@ pub fn merge_views_projects_same_events_into_tuple_state_test() { pub fn react_maps_one_recorded_event_into_effect_values_test() -> Nil { let event = - recorded(UserRegistered("renata"), [factos.tag("username:renata")], revision: 5) + recorded( + UserRegistered("renata"), + [factos.tag("username:renata")], + revision: 5, + ) let reactor = factos.reactor(react: fn(recorded) { case recorded.event { UserRegistered(username) -> [ - recorded.id <> ":" <> int.to_string(recorded.revision) <> ":" <> username, + recorded.id + <> ":" + <> int.to_string(recorded.revision) + <> ":" + <> username, ] UsernameReserved(_) -> [] DisplayNameChanged(_) -> [] } }) - assert factos.react(reactor: reactor, event: event) - == ["event-5:5:renata"] + assert factos.react(reactor: reactor, event: event) == ["event-5:5:renata"] } pub fn react_all_flattens_reactions_in_recorded_event_order_test() -> Nil { @@ -247,7 +254,11 @@ pub fn react_all_flattens_reactions_in_recorded_event_order_test() -> Nil { pub fn merge_reactors_combines_outputs_for_the_same_recorded_event_test() -> Nil { let event = - recorded(UserRegistered("renata"), [factos.tag("username:renata")], revision: 7) + recorded( + UserRegistered("renata"), + [factos.tag("username:renata")], + revision: 7, + ) let audit = factos.reactor(react: fn(recorded) { case recorded.event {