diff --git a/README.md b/README.md index f273581..de379f8 100644 --- a/README.md +++ b/README.md @@ -1,24 +1,218 @@ -# cqrs +# Factos -[![Package Version](https://img.shields.io/hexpm/v/cqrs)](https://hex.pm/packages/cqrs) -[![Hex Docs](https://img.shields.io/badge/hex-docs-ffaff3)](https://hexdocs.pm/cqrs/) +Prototype context-first event-sourcing helpers for Gleam and KurrentDB. -```sh -gleam add cqrs@1 +This package deliberately does not model Event Sourcing as aggregates. A command +capability reads the facts relevant to one decision, folds a temporary decision +model, decides which new facts to record, and records them only when the relevant +context can be protected. + +## Opinion + +Event Sourcing is the persistence idea: accepted facts are the authoritative +state of the system. Aggregates, CQRS, projections, message brokers, and stream +versioning are implementation choices. + +This prototype follows the shape described by Command Context Consistency and +Dynamic Consistency Boundaries: + +1. A command defines the context it needs. +2. The context is expressed as a query over event types and tags. +3. The application folds only those facts into a decision model. +4. The application produces new facts. +5. The store must reject the append if matching facts appeared after the context + was observed. + +That last step requires store support. KurrentDB's normal append API supports +expected stream revision checks. That is useful, but it is not the same as a +DCB-style query-conditioned append. + +## Domain Model + +Keep commands, events, state, decisions, evolution, and errors in your app. + +```gleam +pub type Command { + RegisterUser(username: String) +} + +pub type Event { + UsernameReserved(username: String) + UserRegistered(username: String) +} + +pub type UsernameState { + UsernameAvailable + UsernameTaken +} + +pub type DomainError { + UsernameAlreadyTaken +} + +pub fn evolve(state: UsernameState, event: Event) -> UsernameState { + case state, event { + UsernameAvailable, UsernameReserved(_) -> UsernameTaken + UsernameAvailable, UserRegistered(_) -> UsernameTaken + UsernameTaken, UsernameReserved(_) -> state + UsernameTaken, UserRegistered(_) -> state + } +} + +pub fn decide(state: UsernameState, command: Command) { + case state, command { + UsernameAvailable, RegisterUser(username) -> Ok([UserRegistered(username)]) + UsernameTaken, RegisterUser(_) -> Error(UsernameAlreadyTaken) + } +} +``` + +## Command Context + +The command context is not a `User` aggregate. It is the facts relevant to the +decision: has this username been reserved or registered? + +```gleam +import factos + +pub fn username_context(username: String) -> factos.Query { + factos.query([ + factos.query_item( + types: [ + factos.event_type("UsernameReserved"), + factos.event_type("UserRegistered"), + ], + tags: [factos.tag("username:" <> username)], + ), + ]) +} ``` + +Query items are OR-combined. Within one item, event types are OR-combined and +tags are AND-combined. + +## Tags + +Tags are an explicit query contract. If a future command needs to select events +by username, account, invoice number, or product, that value must be exposed as a +tag when the event is written. + ```gleam -import cqrs +factos.tag("username:renata") +factos.tag("account:abc123") +``` -pub fn main() -> Nil { - // TODO: An example of the project in use +This is intentionally opinionated. Tags duplicate selected payload information, +but they make consistency and query needs visible at the event-store boundary. + +## Codecs + +The app owns encoding and decoding. The codec returns both the domain event and +the event-store metadata needed by the context API. + +```gleam +pub fn codec() -> factos.EventCodec(Event, DecodeError) { + factos.EventCodec(encode: encode, decode: decode_event) } ``` -Further documentation can be found at . +`encode` returns a `factos.Proposed` with the domain event, event type, tags, and +the KurrentDB append message. `decode` returns a `factos.Decoded` with the domain +event, event type, and tags read from the stored event. + +The library does not force one JSON shape for tags. In a real app, store tags in +custom metadata or payload fields consistently, and decode them at the boundary. + +## Reading A Context + +```gleam +factos.read_context( + connection, + query: username_context("renata"), + initial: UsernameAvailable, + evolve: evolve, + codec: codec(), + timeout: 5000, +) +``` + +`read_context` reads from `$all`, applies a server-side event-type filter when +possible, decodes events, filters them by the full query, folds state, and +returns a `factos.Context`. + +The returned context includes: + +1. The folded decision state. +2. The matching recorded events. +3. The highest observed sequence position. +4. A DCB-style append condition: `FailIfEventsMatch(query, after: position)`. + +## Appending + +The ideal append condition is: + +```gleam +factos.FailIfEventsMatch(query, after: position) +``` + +That means: append these new facts only if no facts matching the command context +appeared after the position used for the decision. + +This prototype models that condition, but the current KurrentDB-backed append +function returns `UnsupportedAppendCondition` for it because regular KurrentDB +append checks stream revisions, not arbitrary event-type/tag queries. + +```gleam +factos.append_with_condition( + connection, + stream: "facts", + events: events, + codec: codec(), + condition: context.append_condition, + timeout: 5000, +) +``` + +Use this shape for stores that support DCB-style atomic append conditions. + +## Stream Consistency + +KurrentDB can safely protect a single stream with expected revision checks. This +is still useful when a stream is the right consistency boundary. + +```gleam +factos.dispatch_stream( + connection, + stream: "user-renata", + initial: UsernameAvailable, + decide: decide, + evolve: evolve, + codec: codec(), + command: RegisterUser("renata"), + timeout: 5000, +) +``` + +This is aggregate-stream style consistency. It is not the definition of Event +Sourcing, and it may over-conflict when unrelated events share the same stream. + +## KurrentDB Tradeoffs + +KurrentDB support available through the current dependency: + +1. Read a single stream. +2. Append to a stream with `NoStream`, `Revision(n)`, `StreamExists`, or `Any`. +3. Read `$all` with event-type or stream-name filters. + +Newer KurrentDB versions also support secondary and user-defined indexes that +can be consumed through `$all` stream-prefix filters such as `$idx-et-...` or +`$idx-user-...`. Those improve reads, but the docs describe secondary indexes as +eventually consistent. They should not be treated as a command-decision +consistency guarantee unless the write path can atomically enforce the same +condition. ## Development ```sh -gleam run # Run the project -gleam test # Run the tests +gleam test ``` diff --git a/gleam.toml b/gleam.toml index 8713c24..6e91a47 100644 --- a/gleam.toml +++ b/gleam.toml @@ -1,4 +1,4 @@ -name = "cqrs" +name = "factos" version = "1.0.0" # Fill out these fields if you intend to generate HTML documentation or publish @@ -14,11 +14,10 @@ version = "1.0.0" [dependencies] gleam_stdlib = ">= 1.0.0 and < 2.0.0" -gleam_erlang = ">= 1.3.0 and < 2.0.0" -gleam_otp = ">= 1.2.0 and < 2.0.0" -group_registry = ">= 1.0.0 and < 2.0.0" -pog = ">= 4.1.0 and < 5.0.0" gleam_json = ">= 3.1.0 and < 4.0.0" +kurrentdb = { path = "../kurrentdb" } +kurrentdb_erlang = { path = "../kurrentdb/backends/kurrentdb_erlang" } +youid = ">= 1.6.0 and < 2.0.0" [dev_dependencies] gleeunit = ">= 1.0.0 and < 2.0.0" diff --git a/manifest.toml b/manifest.toml index a24e97f..e92a8cc 100644 --- a/manifest.toml +++ b/manifest.toml @@ -7,26 +7,33 @@ # You should check this file into your source control repository. packages = [ - { name = "backoff", version = "1.1.6", build_tools = ["rebar3"], requirements = [], otp_app = "backoff", source = "hex", outer_checksum = "CF0CFFF8995FB20562F822E5CC47D8CCF664C5ECDC26A684CBE85C225F9D7C39" }, - { name = "exception", version = "2.1.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "exception", source = "hex", outer_checksum = "329D269D5C2A314F7364BD2711372B6F2C58FA6F39981572E5CA68624D291F8C" }, + { name = "certifi", version = "2.17.0", build_tools = ["rebar3"], requirements = [], otp_app = "certifi", source = "hex", outer_checksum = "8122798A17F0293C80DAADA25D0F81C7F4D708C73FEF782C7C9B1950E26E4D21" }, + { name = "gleam_crypto", version = "1.6.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleam_crypto", source = "hex", outer_checksum = "2DE9E4EF53CF6FEE049D4F765731F7178F7A11AEFAE00EEE63BF7536B354AD3F" }, { name = "gleam_erlang", version = "1.3.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleam_erlang", source = "hex", outer_checksum = "1124AD3AA21143E5AF0FC5CF3D9529F6DB8CA03E43A55711B60B6B7B3874375C" }, + { name = "gleam_hackney", version = "1.4.0", build_tools = ["gleam"], requirements = ["gleam_http", "gleam_stdlib", "hackney"], source = "local", path = "../gleam_hackney" }, + { name = "gleam_http", version = "4.3.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleam_http", source = "hex", outer_checksum = "82EA6A717C842456188C190AFB372665EA56CE13D8559BF3B1DD9E40F619EE0C" }, { 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 = "group_registry", version = "1.0.0", build_tools = ["gleam"], requirements = ["gleam_erlang", "gleam_otp", "gleam_stdlib"], otp_app = "group_registry", source = "hex", outer_checksum = "BC798A53D6F2406DB94E27CB45C57052CB56B32ACF7CC16EA20F6BAEC7E36B90" }, - { 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 = "h2", version = "0.10.2", build_tools = ["rebar3"], requirements = [], otp_app = "h2", source = "hex", outer_checksum = "497A899F338B42E6A0B292524E635B0CE6F9379FA39395C8E38D06351CD9B9CF" }, + { name = "hackney", version = "4.4.5", build_tools = ["rebar3"], requirements = ["certifi", "h2", "idna", "mimerl", "parse_trans", "quic", "ssl_verify_fun", "webtransport"], otp_app = "hackney", source = "hex", outer_checksum = "6D72BEF4E135E94C522C271E11FBB6933EFB0006EF235A3933807D0BE73B71EC" }, + { name = "idna", version = "7.1.0", build_tools = ["rebar3"], requirements = [], otp_app = "idna", source = "hex", outer_checksum = "6AE959A025BF36DF61A8CAB8508D9654891B5426A84C44D82DEAFFD6DDF8C71F" }, + { name = "kurrentdb", version = "1.0.0", build_tools = ["gleam"], requirements = ["gleam_http", "gleam_json", "gleam_stdlib", "youid"], source = "local", path = "../kurrentdb" }, + { name = "kurrentdb_erlang", version = "1.0.0", build_tools = ["gleam"], requirements = ["gleam_erlang", "gleam_hackney", "gleam_http", "gleam_otp", "gleam_stdlib", "kurrentdb", "youid"], source = "local", path = "../kurrentdb/backends/kurrentdb_erlang" }, + { name = "mimerl", version = "1.5.0", build_tools = ["rebar3"], requirements = [], otp_app = "mimerl", source = "hex", outer_checksum = "DB648CE065BAE14EA84CA8B5DD123F42F49417CEF693541110BF6F9E9BE9ECC4" }, + { name = "parse_trans", version = "3.4.2", build_tools = ["rebar3"], requirements = [], otp_app = "parse_trans", source = "hex", outer_checksum = "4C25347DE3B7C35732D32E69AB43D1CEEE0BEAE3F3B3ADE1B59CBD3DD224D9CA" }, + { name = "quic", version = "1.6.5", build_tools = ["rebar3"], requirements = [], otp_app = "quic", source = "hex", outer_checksum = "DE1A88972C33201A50D1A17C8C4A14528BD1D8F25EF705897D680AE312D0AA78" }, + { name = "ssl_verify_fun", version = "1.1.7", build_tools = ["mix", "rebar3", "make"], requirements = [], otp_app = "ssl_verify_fun", source = "hex", outer_checksum = "FE4C190E8F37401D30167C8C405EDA19469F34577987C76DDE613E838BBC67F8" }, + { name = "webtransport", version = "0.4.1", build_tools = ["rebar3"], requirements = ["h2", "quic"], otp_app = "webtransport", source = "hex", outer_checksum = "006E4E52A8F03B69201D4637C85B424A4DDCCC1909F32C5742FED8D495A75174" }, + { name = "youid", version = "1.6.0", build_tools = ["gleam"], requirements = ["gleam_crypto", "gleam_stdlib", "gleam_time"], otp_app = "youid", source = "hex", outer_checksum = "7A3ABA44B1B38BC2BDCB5474C5317AA372BE58DFBC649815EE08B03526DDA18D" }, ] [requirements] -gleam_erlang = { version = ">= 1.3.0 and < 2.0.0" } gleam_json = { version = ">= 3.1.0 and < 4.0.0" } -gleam_otp = { version = ">= 1.2.0 and < 2.0.0" } gleam_stdlib = { version = ">= 1.0.0 and < 2.0.0" } gleeunit = { version = ">= 1.0.0 and < 2.0.0" } -group_registry = { version = ">= 1.0.0 and < 2.0.0" } -pog = { version = ">= 4.1.0 and < 5.0.0" } +kurrentdb = { path = "../kurrentdb" } +kurrentdb_erlang = { path = "../kurrentdb/backends/kurrentdb_erlang" } +youid = { version = ">= 1.6.0 and < 2.0.0" } diff --git a/src/cqrs.gleam b/src/cqrs.gleam deleted file mode 100644 index 13da9e3..0000000 --- a/src/cqrs.gleam +++ /dev/null @@ -1,503 +0,0 @@ -import gleam/dict.{type Dict} -import gleam/erlang/process -import gleam/int -import gleam/list -import gleam/otp/actor -import gleam/otp/supervision -import gleam/result -import group_registry - -pub type Metadata = - List(#(String, String)) - -pub type PendingEvent(event) { - PendingEvent(event_type: String, payload: event) -} - -pub type RecordedEvent(event) { - RecordedEvent( - global_position: Int, - stream_id: String, - stream_position: Int, - event_type: String, - payload: event, - metadata: Metadata, - ) -} - -pub type Subscription { - AllEvents - StreamEvents(stream_id: String) -} - -pub type SubscriberMessage(event) { - EventsCommitted(events: List(RecordedEvent(event))) -} - -pub type HandlerFailure(handler_error) { - HandlerFailure(handler: String, reason: handler_error) -} - -pub type Error(domain_error, store_error) { - CommandRejected(reason: domain_error) - EventStoreFailed(reason: store_error) - NoEventsToCommit - EventLogTimedOut(operation: String, timeout_ms: Int) -} - -pub type CommandHandler(command, event, domain_error) = - fn(command) -> Result(List(PendingEvent(event)), domain_error) - -pub type EventHandler(event, handler_error) = - fn(List(RecordedEvent(event))) -> Result(Nil, handler_error) - -pub type EventStore(store, event, store_error) { - EventStore( - store: store, - append: fn(store, String, List(PendingEvent(event)), Metadata) -> - Result(#(store, List(RecordedEvent(event))), store_error), - load_all: fn(store) -> Result(List(RecordedEvent(event)), store_error), - load_stream: fn(store, String) -> - Result(List(RecordedEvent(event)), store_error), - ) -} - -pub opaque type MemoryStore(event) { - MemoryStore( - events: List(RecordedEvent(event)), - stream_positions: Dict(String, Int), - next_global_position: Int, - ) -} - -pub opaque type EventLog(store, event, store_error) { - EventLog(event_store: EventStore(store, event, store_error)) -} - -pub opaque type System(event, store_error) { - System( - event_log: process.Subject(Message(event, store_error)), - registry: group_registry.GroupRegistry(SubscriberMessage(event)), - timeout: Int, - ) -} - -type SystemMessage { - StopSystem -} - -pub type Message(event, store_error) { - Append( - stream_id: String, - events: List(PendingEvent(event)), - metadata: Metadata, - reply_to: process.Subject(Result(List(RecordedEvent(event)), store_error)), - ) - LoadAll( - reply_to: process.Subject(Result(List(RecordedEvent(event)), store_error)), - ) - LoadStream( - stream_id: String, - reply_to: process.Subject(Result(List(RecordedEvent(event)), store_error)), - ) - Stop -} - -pub type HandlerMessage(event, handler_error) { - HandleEvents(events: List(RecordedEvent(event))) - GetHandlerFailures( - reply_to: process.Subject(List(HandlerFailure(handler_error))), - ) - StopHandler -} - -pub const default_timeout: Int = 100 - -pub fn memory_store() -> EventStore(MemoryStore(event), event, Nil) { - EventStore( - store: MemoryStore( - events: [], - stream_positions: dict.new(), - next_global_position: 1, - ), - append: append_to_memory_store, - load_all: fn(store) { Ok(store.events) }, - load_stream: fn(store, stream_id) { - Ok(list.filter(store.events, fn(event) { event.stream_id == stream_id })) - }, - ) -} - -pub fn start( - registry_name: process.Name(group_registry.Message(SubscriberMessage(event))), - timeout timeout_value: Int, - event_store event_store: EventStore(store, event, store_error), -) -> actor.StartResult(System(event, store_error)) { - actor.new_with_initialiser(default_timeout, fn(_subject) { - use registry <- result.try( - group_registry.start(registry_name) - |> result.replace_error("CQRS registry could not be started"), - ) - use event_log <- result.try( - start_event_log(event_store) - |> result.replace_error("CQRS event log could not be started"), - ) - - actor.initialised(Nil) - |> actor.returning(System( - event_log: event_log.data, - registry: registry.data, - timeout: timeout_value, - )) - |> Ok - }) - |> actor.on_message(handle_system_message) - |> actor.start -} - -pub fn supervised( - registry_name registry_name: process.Name( - group_registry.Message(SubscriberMessage(event)), - ), - timeout timeout_value: Int, - event_store event_store: EventStore(store, event, store_error), -) -> supervision.ChildSpecification(System(event, store_error)) { - supervision.worker(fn() { - start(registry_name, timeout: timeout_value, event_store: event_store) - }) -} - -fn start_event_log( - event_store: EventStore(store, event, store_error), -) -> actor.StartResult(process.Subject(Message(event, store_error))) { - actor.new(EventLog(event_store: event_store)) - |> actor.on_message(handle_message) - |> actor.start -} - -fn handle_system_message( - _state: Nil, - message: SystemMessage, -) -> actor.Next(Nil, SystemMessage) { - case message { - StopSystem -> actor.stop() - } -} - -pub fn subscribe( - system: System(event, store_error), - subscription: Subscription, -) -> process.Subject(SubscriberMessage(event)) { - group_registry.join(system.registry, group_name(subscription), process.self()) -} - -pub fn start_handler( - system: System(event, store_error), - name: String, - subscription: Subscription, - handler: EventHandler(event, handler_error), -) -> actor.StartResult(process.Subject(HandlerMessage(event, handler_error))) { - actor.new_with_initialiser(default_timeout, fn(subject) { - let events = - group_registry.join( - system.registry, - group_name(subscription), - process.self(), - ) - - actor.initialised([]) - |> actor.returning(subject) - |> actor.selecting( - process.new_selector() - |> process.select(subject) - |> process.select_map(events, fn(message) { - case message { - EventsCommitted(events) -> HandleEvents(events) - } - }), - ) - |> Ok - }) - |> actor.on_message(handle_handler_message(name, handler)) - |> actor.start -} - -pub fn handler_failures( - handler: process.Subject(HandlerMessage(event, handler_error)), - timeout timeout_value: Int, -) -> Result(List(HandlerFailure(handler_error)), Nil) { - let reply_to = process.new_subject() - process.send(handler, GetHandlerFailures(reply_to)) - process.receive(reply_to, timeout_value) -} - -pub fn stop_handler( - handler: process.Subject(HandlerMessage(event, handler_error)), -) -> Nil { - process.send(handler, StopHandler) -} - -@internal -pub fn append( - system: System(event, store_error), - stream_id: String, - events: List(PendingEvent(event)), - metadata: Metadata, -) -> Result(List(RecordedEvent(event)), Error(domain_error, store_error)) { - case events { - [] -> Error(NoEventsToCommit) - [_, ..] -> { - let reply_to = process.new_subject() - process.send( - system.event_log, - Append(stream_id, events, metadata, reply_to), - ) - - case process.receive(reply_to, system.timeout) { - Ok(Ok(recorded_events)) -> { - publish(system, recorded_events) - Ok(recorded_events) - } - Ok(Error(reason)) -> Error(EventStoreFailed(reason)) - Error(Nil) -> Error(EventLogTimedOut("append", system.timeout)) - } - } - } -} - -pub fn execute( - system: System(event, store_error), - stream_id: String, - command: command, - metadata: Metadata, - handle: CommandHandler(command, event, domain_error), -) -> Result(List(RecordedEvent(event)), Error(domain_error, store_error)) { - case handle(command) { - Ok(events) -> append(system, stream_id, events, metadata) - Error(reason) -> Error(CommandRejected(reason)) - } -} - -pub fn load_all( - system: System(event, store_error), -) -> Result(List(RecordedEvent(event)), Error(domain_error, store_error)) { - let reply_to = process.new_subject() - process.send(system.event_log, LoadAll(reply_to)) - - case process.receive(reply_to, system.timeout) { - Ok(Ok(events)) -> Ok(events) - Ok(Error(reason)) -> Error(EventStoreFailed(reason)) - Error(Nil) -> Error(EventLogTimedOut("load_all", system.timeout)) - } -} - -pub fn load_stream( - system: System(event, store_error), - stream_id: String, -) -> Result(List(RecordedEvent(event)), Error(domain_error, store_error)) { - let reply_to = process.new_subject() - process.send(system.event_log, LoadStream(stream_id, reply_to)) - - case process.receive(reply_to, system.timeout) { - Ok(Ok(events)) -> Ok(events) - Ok(Error(reason)) -> Error(EventStoreFailed(reason)) - Error(Nil) -> Error(EventLogTimedOut("load_stream", system.timeout)) - } -} - -pub fn stop(system: System(event, store_error)) -> Nil { - process.send(system.event_log, Stop) -} - -fn handle_message( - event_log: EventLog(store, event, store_error), - message: Message(event, store_error), -) -> actor.Next( - EventLog(store, event, store_error), - Message(event, store_error), -) { - case message { - Append(_stream_id, [], _metadata, reply_to) -> { - process.send(reply_to, Ok([])) - actor.continue(event_log) - } - Append(stream_id, [_, ..] as events, metadata, reply_to) -> { - case - event_log.event_store.append( - event_log.event_store.store, - stream_id, - events, - metadata, - ) - { - Ok(#(store, recorded_events)) -> { - let event_store = EventStore(..event_log.event_store, store: store) - process.send(reply_to, Ok(recorded_events)) - actor.continue(EventLog(event_store: event_store)) - } - Error(reason) -> { - process.send(reply_to, Error(reason)) - actor.continue(event_log) - } - } - } - LoadAll(reply_to) -> { - let events = event_log.event_store.load_all(event_log.event_store.store) - process.send(reply_to, events) - actor.continue(event_log) - } - LoadStream(stream_id, reply_to) -> { - let events = - event_log.event_store.load_stream( - event_log.event_store.store, - stream_id, - ) - process.send(reply_to, events) - actor.continue(event_log) - } - Stop -> actor.stop() - } -} - -fn handle_handler_message( - name: String, - handler: EventHandler(event, handler_error), -) -> fn( - List(HandlerFailure(handler_error)), - HandlerMessage(event, handler_error), -) -> - actor.Next( - List(HandlerFailure(handler_error)), - HandlerMessage(event, handler_error), - ) { - fn(failures, message) { - case message { - HandleEvents(events) -> { - case handler(events) { - Ok(Nil) -> actor.continue(failures) - Error(reason) -> - actor.continue([ - HandlerFailure(handler: name, reason: reason), - ..failures - ]) - } - } - GetHandlerFailures(reply_to) -> { - process.send(reply_to, list.reverse(failures)) - actor.continue(failures) - } - StopHandler -> actor.stop() - } - } -} - -fn append_to_memory_store( - memory_store: MemoryStore(event), - stream_id: String, - events: List(PendingEvent(event)), - metadata: Metadata, -) -> Result(#(MemoryStore(event), List(RecordedEvent(event))), Nil) { - let stream_position = - memory_store.stream_positions - |> dict.get(stream_id) - |> result.unwrap(0) - - let #(recorded_events, next_global_position, next_stream_position) = - record_events( - events, - stream_id, - metadata, - memory_store.next_global_position, - stream_position, - ) - - let memory_store = - MemoryStore( - events: list.append(memory_store.events, recorded_events), - stream_positions: dict.insert( - memory_store.stream_positions, - stream_id, - next_stream_position, - ), - next_global_position: next_global_position, - ) - - Ok(#(memory_store, recorded_events)) -} - -fn record_events( - events: List(PendingEvent(event)), - stream_id: String, - metadata: Metadata, - next_global_position: Int, - stream_position: Int, -) -> #(List(RecordedEvent(event)), Int, Int) { - let #(recorded_events, next_global_position, next_stream_position) = - list.fold( - events, - #([], next_global_position, stream_position), - fn(accumulator, pending_event) { - let #(recorded_events, global_position, stream_position) = accumulator - let PendingEvent(event_type, payload) = pending_event - let stream_position = stream_position + 1 - let recorded_event = - RecordedEvent( - global_position: global_position, - stream_id: stream_id, - stream_position: stream_position, - event_type: event_type, - payload: payload, - metadata: metadata, - ) - - #( - [recorded_event, ..recorded_events], - global_position + 1, - stream_position, - ) - }, - ) - - #(list.reverse(recorded_events), next_global_position, next_stream_position) -} - -fn publish( - system: System(event, store_error), - events: List(RecordedEvent(event)), -) -> Nil { - case events { - [] -> Nil - [first_event, ..] -> { - notify(system, AllEvents, events) - notify(system, StreamEvents(first_event.stream_id), events) - } - } -} - -fn notify( - system: System(event, store_error), - subscription: Subscription, - events: List(RecordedEvent(event)), -) -> Nil { - group_registry.members(system.registry, group_name(subscription)) - |> list.each(fn(subscriber) { - process.send(subscriber, EventsCommitted(events)) - }) -} - -fn group_name(subscription: Subscription) -> String { - case subscription { - AllEvents -> "cqrs$all" - StreamEvents(stream_id) -> "cqrs$stream:" <> stream_id - } -} - -pub fn recorded_event_to_string(event: RecordedEvent(a)) -> String { - int.to_string(event.global_position) - <> "@" - <> event.stream_id - <> ":" - <> int.to_string(event.stream_position) - <> "/" - <> event.event_type -} diff --git a/src/factos.gleam b/src/factos.gleam new file mode 100644 index 0000000..cf15f4c --- /dev/null +++ b/src/factos.gleam @@ -0,0 +1,673 @@ +//// Context-first event-sourcing helpers for Gleam and KurrentDB. +//// +//// Event Sourcing does not require aggregates. A command capability should read +//// the facts relevant to its decision, fold a temporary decision model, decide +//// which new facts to record, and then record those facts only when the relevant +//// context is still stable. This module models that API directly. +//// +//// KurrentDB's regular append API can guarantee stream revision consistency. It +//// cannot, through the operations used here, atomically guarantee a DCB-style +//// "fail if events matching this query were appended after position X" check. +//// The API keeps those concepts separate instead of pretending they are the same. + +import gleam/list +import gleam/result +import kurrentdb/operation/append_to_stream +import kurrentdb/operation/read_all +import kurrentdb/operation/read_stream +import kurrentdb_erlang +import youid/uuid.{type Uuid} + +pub type EventType { + EventType(String) +} + +pub type Tag { + Tag(String) +} + +pub type Query { + AllEvents + Query(items: List(QueryItem)) +} + +pub type QueryItem { + QueryItem(types: List(EventType), tags: List(Tag)) +} + +pub type SequencePosition { + NoPosition + SequencePosition(commit_position: Int, prepare_position: Int) +} + +pub type AppendCondition { + NoAppendCondition + FailIfEventsMatch(query: Query, after: SequencePosition) +} + +pub type Revision { + NoEvents + CurrentRevision(Int) +} + +pub type Decoded(event) { + Decoded(event: event, type_: EventType, tags: List(Tag)) +} + +pub type Proposed(event) { + Proposed( + event: event, + type_: EventType, + tags: List(Tag), + message: append_to_stream.Event, + ) +} + +pub type EventCodec(event, decode_error) { + EventCodec( + encode: fn(event) -> Proposed(event), + decode: fn(read_stream.RecordedEvent) -> + Result(Decoded(event), decode_error), + ) +} + +pub type Recorded(event) { + Recorded( + id: Uuid, + stream: String, + revision: Int, + position: SequencePosition, + metadata: List(#(String, String)), + custom_metadata: BitArray, + data: BitArray, + type_: EventType, + tags: List(Tag), + event: event, + ) +} + +pub type Context(event, state) { + Context( + query: Query, + state: state, + events: List(Recorded(event)), + position: SequencePosition, + append_condition: AppendCondition, + ) +} + +pub type LoadedStream(event, state) { + LoadedStream( + stream: String, + state: state, + events: List(Recorded(event)), + revision: Revision, + ) +} + +pub type Error(domain_error, decode_error) { + DomainError(domain_error) + DecodeError(decode_error) + ReadError(kurrentdb_erlang.Error(read_stream.ResponseError)) + AppendError(kurrentdb_erlang.Error(append_to_stream.ResponseError)) + ReadTimedOut + UnsupportedAppendCondition(AppendCondition) +} + +pub fn event_type(name: String) -> EventType { + EventType(name) +} + +pub fn tag(value: String) -> Tag { + Tag(value) +} + +pub fn query(items: List(QueryItem)) -> Query { + case items { + [] -> AllEvents + [_, ..] -> Query(items) + } +} + +pub fn query_item( + types types: List(EventType), + tags tags: List(Tag), +) -> QueryItem { + QueryItem(types:, tags:) +} + +pub fn read_context( + connection: kurrentdb_erlang.Connection, + query query: Query, + initial initial: state, + evolve evolve: fn(state, event) -> state, + codec codec: EventCodec(event, decode_error), + timeout timeout: Int, +) -> Result(Context(event, state), Error(domain_error, decode_error)) { + let stream = + kurrentdb_erlang.read_all( + connection, + config: read_all.configure() + |> read_all.filter(query_to_read_all_filter(query)), + ) + + receive_context( + stream, + query, + initial, + evolve, + codec, + [], + NoPosition, + timeout, + ) +} + +pub fn decide_context( + context: Context(event, state), + command: command, + decide: fn(state, command) -> Result(List(event), domain_error), +) -> Result( + #(Context(event, state), List(event)), + Error(domain_error, decode_error), +) { + use events <- result.try( + decide(context.state, command) + |> result.map_error(DomainError), + ) + + Ok(#(context, events)) +} + +pub fn append_with_condition( + connection: kurrentdb_erlang.Connection, + stream stream_name: String, + events events: List(event), + codec codec: EventCodec(event, decode_error), + condition condition: AppendCondition, + timeout timeout: Int, +) -> Result(append_to_stream.Append, Error(domain_error, decode_error)) { + case condition { + NoAppendCondition -> + append_to_stream_with_config( + connection, + stream_name, + events, + codec, + append_to_stream.configure() |> append_to_stream.any, + timeout, + ) + FailIfEventsMatch(_, _) -> Error(UnsupportedAppendCondition(condition)) + } +} + +pub fn load_stream( + connection: kurrentdb_erlang.Connection, + stream stream_name: String, + initial initial: state, + evolve evolve: fn(state, event) -> state, + codec codec: EventCodec(event, decode_error), + timeout timeout: Int, +) -> Result(LoadedStream(event, state), Error(domain_error, decode_error)) { + let stream = + kurrentdb_erlang.read_stream( + connection, + stream: stream_name, + config: read_stream.configure(), + ) + + receive_stream( + stream, + stream_name, + initial, + evolve, + codec, + [], + NoEvents, + timeout, + ) +} + +pub fn dispatch_stream( + connection: kurrentdb_erlang.Connection, + stream stream_name: String, + initial initial: state, + decide decide: fn(state, command) -> Result(List(event), domain_error), + evolve evolve: fn(state, event) -> state, + codec codec: EventCodec(event, decode_error), + command command: command, + timeout timeout: Int, +) -> Result(append_to_stream.Append, Error(domain_error, decode_error)) { + use loaded <- result.try(load_stream( + connection, + stream: stream_name, + initial: initial, + evolve: evolve, + codec: codec, + timeout: timeout, + )) + + use events <- result.try( + decide(loaded.state, command) + |> result.map_error(DomainError), + ) + + append_stream_events( + connection, + stream: stream_name, + events: events, + codec: codec, + expected: loaded.revision, + timeout: timeout, + ) +} + +pub fn append_stream_events( + connection: kurrentdb_erlang.Connection, + stream stream_name: String, + events events: List(event), + codec codec: EventCodec(event, decode_error), + expected expected: Revision, + timeout timeout: Int, +) -> Result(append_to_stream.Append, Error(domain_error, decode_error)) { + append_to_stream_with_config( + connection, + stream_name, + events, + codec, + append_config(expected), + timeout, + ) +} + +pub fn fold( + initial: state, + events: List(Recorded(event)), + evolve: fn(state, event) -> state, +) -> state { + use state, recorded <- list.fold(events, initial) + evolve(state, recorded.event) +} + +pub fn matches_query(recorded: Recorded(event), query: Query) -> Bool { + case query { + AllEvents -> True + Query(items) -> list.any(items, fn(item) { matches_item(recorded, item) }) + } +} + +pub fn highest_position( + left: SequencePosition, + right: SequencePosition, +) -> SequencePosition { + case left, right { + NoPosition, position -> position + position, NoPosition -> position + SequencePosition(left_commit, left_prepare), + SequencePosition(right_commit, right_prepare) + -> { + case + left_commit > right_commit + || { left_commit == right_commit && left_prepare >= right_prepare } + { + True -> left + False -> right + } + } + } +} + +fn append_to_stream_with_config( + connection: kurrentdb_erlang.Connection, + stream_name: String, + events: List(event), + codec: EventCodec(event, decode_error), + config: append_to_stream.Configuration, + timeout: Int, +) -> Result(append_to_stream.Append, Error(domain_error, decode_error)) { + case events { + [] -> + Ok(append_to_stream.Append( + current_revision: -1, + position: append_to_stream.NoPositionReturned, + )) + [_, ..] -> { + let EventCodec(encode, _) = codec + let task = + kurrentdb_erlang.append_to_stream( + connection, + stream: stream_name, + events: list.map(events, fn(event) { + let Proposed(message: message, ..) = encode(event) + message + }), + config: config, + ) + + kurrentdb_erlang.await(task, within: timeout) + |> result.map_error(AppendError) + } + } +} + +fn receive_context( + stream: kurrentdb_erlang.Stream, + query: Query, + state: state, + evolve: fn(state, event) -> state, + codec: EventCodec(event, decode_error), + events: List(Recorded(event)), + position: SequencePosition, + timeout: Int, +) -> Result(Context(event, state), Error(domain_error, decode_error)) { + case kurrentdb_erlang.receive(stream, within: timeout) { + Error(Nil) -> { + kurrentdb_erlang.close(stream) + Error(ReadTimedOut) + } + Ok(kurrentdb_erlang.StreamFinished) -> { + kurrentdb_erlang.close(stream) + Ok(Context( + query:, + state:, + events: list.reverse(events), + position:, + append_condition: FailIfEventsMatch(query, position), + )) + } + Ok(kurrentdb_erlang.StreamFailed(error)) -> { + kurrentdb_erlang.close(stream) + Error(ReadError(error)) + } + Ok(kurrentdb_erlang.ReadEvent(event)) -> + receive_context_event( + stream, + query, + state, + evolve, + codec, + events, + position, + timeout, + event, + ) + Ok(kurrentdb_erlang.ReadMessage(read_stream.ReadEvent(event))) -> + receive_context_event( + stream, + query, + state, + evolve, + codec, + events, + position, + timeout, + event, + ) + Ok(kurrentdb_erlang.ReadMessage(read_stream.LastAllStreamPosition(read_stream.Position( + commit_position: commit_position, + prepare_position: prepare_position, + )))) -> + receive_context( + stream, + query, + state, + evolve, + codec, + events, + highest_position( + position, + SequencePosition(commit_position, prepare_position), + ), + timeout, + ) + Ok(kurrentdb_erlang.ReadMessage(_)) -> + receive_context( + stream, + query, + state, + evolve, + codec, + events, + position, + timeout, + ) + } +} + +fn receive_context_event( + stream: kurrentdb_erlang.Stream, + query: Query, + state: state, + evolve: fn(state, event) -> state, + codec: EventCodec(event, decode_error), + events: List(Recorded(event)), + position: SequencePosition, + timeout: Int, + read_event: read_stream.ReadEvent, +) -> Result(Context(event, state), Error(domain_error, decode_error)) { + use recorded <- result.try(decode_recorded(read_event, codec)) + let next_position = highest_position(position, recorded.position) + + case matches_query(recorded, query) { + True -> + receive_context( + stream, + query, + evolve(state, recorded.event), + evolve, + codec, + [recorded, ..events], + next_position, + timeout, + ) + False -> + receive_context( + stream, + query, + state, + evolve, + codec, + events, + next_position, + timeout, + ) + } +} + +fn receive_stream( + stream: kurrentdb_erlang.Stream, + stream_name: String, + state: state, + evolve: fn(state, event) -> state, + codec: EventCodec(event, decode_error), + events: List(Recorded(event)), + revision: Revision, + timeout: Int, +) -> Result(LoadedStream(event, state), Error(domain_error, decode_error)) { + case kurrentdb_erlang.receive(stream, within: timeout) { + Error(Nil) -> { + kurrentdb_erlang.close(stream) + Error(ReadTimedOut) + } + Ok(kurrentdb_erlang.StreamFinished) -> { + kurrentdb_erlang.close(stream) + Ok(LoadedStream( + stream: stream_name, + state:, + events: list.reverse(events), + revision:, + )) + } + Ok(kurrentdb_erlang.StreamFailed(error)) -> { + kurrentdb_erlang.close(stream) + case error { + kurrentdb_erlang.OperationError(read_stream.ReadStreamNotFound(_)) -> + Ok(LoadedStream( + stream: stream_name, + state:, + events: [], + revision: NoEvents, + )) + _ -> Error(ReadError(error)) + } + } + Ok(kurrentdb_erlang.ReadEvent(event)) -> + receive_stream_event( + stream, + stream_name, + state, + evolve, + codec, + events, + timeout, + event, + ) + Ok(kurrentdb_erlang.ReadMessage(read_stream.ReadEvent(event))) -> + receive_stream_event( + stream, + stream_name, + state, + evolve, + codec, + events, + timeout, + event, + ) + Ok(kurrentdb_erlang.ReadMessage(_)) -> + receive_stream( + stream, + stream_name, + state, + evolve, + codec, + events, + revision, + timeout, + ) + } +} + +fn receive_stream_event( + stream: kurrentdb_erlang.Stream, + stream_name: String, + state: state, + evolve: fn(state, event) -> state, + codec: EventCodec(event, decode_error), + events: List(Recorded(event)), + timeout: Int, + read_event: read_stream.ReadEvent, +) -> Result(LoadedStream(event, state), Error(domain_error, decode_error)) { + use recorded <- result.try(decode_recorded(read_event, codec)) + + receive_stream( + stream, + stream_name, + evolve(state, recorded.event), + evolve, + codec, + [recorded, ..events], + CurrentRevision(recorded.revision), + timeout, + ) +} + +fn decode_recorded( + read_event: read_stream.ReadEvent, + codec: EventCodec(event, decode_error), +) -> Result(Recorded(event), Error(domain_error, decode_error)) { + let recorded_event = case read_event { + read_stream.Recorded(event) -> event + read_stream.Resolved(event: event, ..) -> event + } + + let EventCodec(_, decode) = codec + use decoded <- result.try( + decode(recorded_event) + |> result.map_error(DecodeError), + ) + + let Decoded(event, type_, tags) = decoded + Ok(Recorded( + id: recorded_event.id, + stream: recorded_event.stream, + revision: recorded_event.revision, + position: SequencePosition( + recorded_event.commit_position, + recorded_event.prepare_position, + ), + metadata: recorded_event.metadata, + custom_metadata: recorded_event.custom_metadata, + data: recorded_event.data, + type_: type_, + tags: tags, + event: event, + )) +} + +fn query_to_read_all_filter(query: Query) -> read_all.Filter { + let types = query_event_type_names(query) + + case types { + [] -> read_all.NoFilter + [_, ..] -> read_all.EventTypePrefix(types, window: read_all.FilterMax(1000)) + } +} + +fn query_event_type_names(query: Query) -> List(String) { + case query { + AllEvents -> [] + Query(items) -> collect_type_names(items, []) + } +} + +fn collect_type_names( + items: List(QueryItem), + names: List(String), +) -> List(String) { + case items { + [] -> list.reverse(names) + [QueryItem(types, _), ..rest] -> + collect_type_names(rest, prepend_type_names(types, names)) + } +} + +fn prepend_type_names( + types: List(EventType), + names: List(String), +) -> List(String) { + case types { + [] -> names + [EventType(name), ..rest] -> prepend_type_names(rest, [name, ..names]) + } +} + +fn matches_item(recorded: Recorded(event), item: QueryItem) -> Bool { + let QueryItem(types, tags) = item + matches_types(recorded.type_, types) && matches_tags(recorded.tags, tags) +} + +fn matches_types(event_type: EventType, types: List(EventType)) -> Bool { + case types { + [] -> True + [_, ..] -> list.any(types, fn(required) { event_type == required }) + } +} + +fn matches_tags(event_tags: List(Tag), required_tags: List(Tag)) -> Bool { + case required_tags { + [] -> True + [_, ..] -> + list.all(required_tags, fn(required) { + list.any(event_tags, fn(tag) { tag == required }) + }) + } +} + +fn append_config(revision: Revision) -> append_to_stream.Configuration { + case revision { + NoEvents -> append_to_stream.configure() |> append_to_stream.no_stream + CurrentRevision(revision) -> + append_to_stream.configure() + |> append_to_stream.expected_revision(revision) + } +} diff --git a/stores/cqrs_postgres/.github/workflows/test.yml b/stores/cqrs_postgres/.github/workflows/test.yml deleted file mode 100644 index eb8b0e4..0000000 --- a/stores/cqrs_postgres/.github/workflows/test.yml +++ /dev/null @@ -1,23 +0,0 @@ -name: test - -on: - push: - branches: - - master - - main - pull_request: - -jobs: - test: - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v6 - - uses: erlef/setup-beam@v1 - with: - otp-version: "29" - gleam-version: "1.17.0" - rebar3-version: "3" - # elixir-version: "1" - - run: gleam deps download - - run: gleam test - - run: gleam format --check src test diff --git a/stores/cqrs_postgres/.gitignore b/stores/cqrs_postgres/.gitignore deleted file mode 100644 index 599be4e..0000000 --- a/stores/cqrs_postgres/.gitignore +++ /dev/null @@ -1,4 +0,0 @@ -*.beam -*.ez -/build -erl_crash.dump diff --git a/stores/cqrs_postgres/README.md b/stores/cqrs_postgres/README.md deleted file mode 100644 index 9a32a78..0000000 --- a/stores/cqrs_postgres/README.md +++ /dev/null @@ -1,24 +0,0 @@ -# cqrs_postgres - -[![Package Version](https://img.shields.io/hexpm/v/cqrs_postgres)](https://hex.pm/packages/cqrs_postgres) -[![Hex Docs](https://img.shields.io/badge/hex-docs-ffaff3)](https://hexdocs.pm/cqrs_postgres/) - -```sh -gleam add cqrs_postgres@1 -``` -```gleam -import cqrs_postgres - -pub fn main() -> Nil { - // TODO: An example of the project in use -} -``` - -Further documentation can be found at . - -## Development - -```sh -gleam run # Run the project -gleam test # Run the tests -``` diff --git a/stores/cqrs_postgres/compose.yml b/stores/cqrs_postgres/compose.yml deleted file mode 100644 index c80433f..0000000 --- a/stores/cqrs_postgres/compose.yml +++ /dev/null @@ -1,14 +0,0 @@ -services: - postgres: - image: postgres:16-alpine - ports: - - "54329:5432" - environment: - POSTGRES_DB: cqrs_test - POSTGRES_USER: cqrs - POSTGRES_PASSWORD: cqrs - healthcheck: - test: ["CMD-SHELL", "pg_isready -U cqrs -d cqrs_test"] - interval: 1s - timeout: 5s - retries: 30 diff --git a/stores/cqrs_postgres/gleam.toml b/stores/cqrs_postgres/gleam.toml deleted file mode 100644 index 10ef8b0..0000000 --- a/stores/cqrs_postgres/gleam.toml +++ /dev/null @@ -1,25 +0,0 @@ -name = "cqrs_postgres" -version = "1.0.0" - -# Fill out these fields if you intend to generate HTML documentation or publish -# your project to the Hex package manager. -# -# description = "" -# licences = ["Apache-2.0"] -# repository = { type = "github", user = "", repo = "" } -# links = [{ title = "Website", href = "" }] -# -# For a full reference of all the available options, you can have a look at -# https://gleam.run/writing-gleam/gleam-toml/. - -[dependencies] -gleam_stdlib = ">= 1.0.0 and < 2.0.0" -pog = ">= 4.1.0 and < 5.0.0" -gleam_json = ">= 3.1.0 and < 4.0.0" -cqrs = { path = "../.." } -gleam_erlang = ">= 1.3.0 and < 2.0.0" -gleam_otp = ">= 1.2.0 and < 2.0.0" - -[dev_dependencies] -gleeunit = ">= 1.11.0 and < 2.0.0" -global_value = ">= 1.0.0 and < 2.0.0" diff --git a/stores/cqrs_postgres/manifest.toml b/stores/cqrs_postgres/manifest.toml deleted file mode 100644 index c579397..0000000 --- a/stores/cqrs_postgres/manifest.toml +++ /dev/null @@ -1,35 +0,0 @@ -# Do not manually edit this file, it is managed by Gleam. -# -# This file locks the dependency versions used, to make your build -# deterministic and to prevent unexpected versions from being included -# in your application. -# -# You should check this file into your source control repository. - -packages = [ - { name = "backoff", version = "1.1.6", build_tools = ["rebar3"], requirements = [], otp_app = "backoff", source = "hex", outer_checksum = "CF0CFFF8995FB20562F822E5CC47D8CCF664C5ECDC26A684CBE85C225F9D7C39" }, - { name = "cqrs", version = "1.0.0", build_tools = ["gleam"], requirements = ["gleam_erlang", "gleam_json", "gleam_otp", "gleam_stdlib", "group_registry", "pog"], source = "local", path = "../.." }, - { name = "exception", version = "2.1.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "exception", source = "hex", outer_checksum = "329D269D5C2A314F7364BD2711372B6F2C58FA6F39981572E5CA68624D291F8C" }, - { 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 = "group_registry", version = "1.0.0", build_tools = ["gleam"], requirements = ["gleam_erlang", "gleam_otp", "gleam_stdlib"], otp_app = "group_registry", source = "hex", outer_checksum = "BC798A53D6F2406DB94E27CB45C57052CB56B32ACF7CC16EA20F6BAEC7E36B90" }, - { 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" }, -] - -[requirements] -cqrs = { path = "../.." } -gleam_erlang = { version = ">= 1.3.0 and < 2.0.0" } -gleam_json = { version = ">= 3.1.0 and < 4.0.0" } -gleam_otp = { version = ">= 1.2.0 and < 2.0.0" } -gleam_stdlib = { version = ">= 1.0.0 and < 2.0.0" } -gleeunit = { version = ">= 1.11.0 and < 2.0.0" } -global_value = { version = ">= 1.0.0 and < 2.0.0" } -pog = { version = ">= 4.1.0 and < 5.0.0" } diff --git a/stores/cqrs_postgres/src/cqrs/stores/postgres_store.gleam b/stores/cqrs_postgres/src/cqrs/stores/postgres_store.gleam deleted file mode 100644 index 62688e6..0000000 --- a/stores/cqrs_postgres/src/cqrs/stores/postgres_store.gleam +++ /dev/null @@ -1,362 +0,0 @@ -import cqrs -import gleam/dynamic/decode -import gleam/json -import gleam/list -import gleam/result -import pog - -const create_event_table_query: String = " -CREATE TABLE IF NOT EXISTS cqrs_event ( - global_position bigint PRIMARY KEY, - stream_id text NOT NULL, - stream_position bigint NOT NULL, - event_type text NOT NULL, - payload text NOT NULL, - metadata text NOT NULL, - recorded_at timestamptz NOT NULL DEFAULT now(), - UNIQUE (stream_id, stream_position) -); -" - -const lock_event_table_query: String = " -SELECT 1 -FROM pg_advisory_xact_lock(hashtext('cqrs_event')); -" - -const select_next_global_position_query: String = " -SELECT COALESCE(MAX(global_position), 0) + 1 -FROM cqrs_event; -" - -const select_next_stream_position_query: String = " -SELECT COALESCE(MAX(stream_position), 0) + 1 -FROM cqrs_event -WHERE stream_id = $1; -" - -const insert_event_query: String = " -INSERT INTO cqrs_event - (global_position, stream_id, stream_position, event_type, payload, metadata) -VALUES - ($1, $2, $3, $4, $5, $6); -" - -const load_all_query: String = " -SELECT global_position, stream_id, stream_position, event_type, payload, metadata -FROM cqrs_event -ORDER BY global_position ASC; -" - -const load_stream_query: String = " -SELECT global_position, stream_id, stream_position, event_type, payload, metadata -FROM cqrs_event -WHERE stream_id = $1 -ORDER BY stream_position ASC; -" - -pub opaque type PostgresStore(event) { - PostgresStore( - connection: pog.Connection, - event_to_json: fn(event) -> json.Json, - event_decoder: decode.Decoder(event), - ) -} - -pub type PostgresStoreError { - QueryFailed(reason: pog.QueryError) - TransactionFailed(reason: pog.QueryError) - PayloadCouldNotBeDecoded(event_type: String, reason: json.DecodeError) - MetadataCouldNotBeDecoded(reason: json.DecodeError) - ExpectedSingleRow(query: String) -} - -type EventRow { - EventRow( - global_position: Int, - stream_id: String, - stream_position: Int, - event_type: String, - payload: String, - metadata: String, - ) -} - -pub fn new( - connection: pog.Connection, - event_to_json: fn(event) -> json.Json, - event_decoder: decode.Decoder(event), -) -> cqrs.EventStore(PostgresStore(event), event, PostgresStoreError) { - cqrs.EventStore( - store: PostgresStore( - connection: connection, - event_to_json: event_to_json, - event_decoder: event_decoder, - ), - append: append, - load_all: load_all, - load_stream: load_stream, - ) -} - -pub fn create_table( - connection: pog.Connection, -) -> Result(Nil, PostgresStoreError) { - create_event_table_query - |> pog.query - |> pog.returning(nil_decoder()) - |> pog.execute(connection) - |> result.replace(Nil) - |> result.map_error(QueryFailed) -} - -fn append( - store: PostgresStore(event), - stream_id: String, - events: List(cqrs.PendingEvent(event)), - metadata: List(#(String, String)), -) -> Result( - #(PostgresStore(event), List(cqrs.RecordedEvent(event))), - PostgresStoreError, -) { - case - pog.transaction(store.connection, fn(transaction) { - use _ <- result.try(lock_event_table(transaction)) - use next_global_position <- result.try(next_global_position(transaction)) - use next_stream_position <- result.try(next_stream_position( - transaction, - stream_id, - )) - - let recorded_events = - record_events( - events, - stream_id, - metadata, - next_global_position, - next_stream_position, - ) - - use _ <- result.try(insert_events(transaction, store, recorded_events)) - Ok(recorded_events) - }) - { - Ok(recorded_events) -> Ok(#(store, recorded_events)) - Error(pog.TransactionQueryError(reason)) -> Error(TransactionFailed(reason)) - Error(pog.TransactionRolledBack(reason)) -> Error(reason) - } -} - -fn load_all( - store: PostgresStore(event), -) -> Result(List(cqrs.RecordedEvent(event)), PostgresStoreError) { - load_all_query - |> pog.query - |> pog.returning(event_row_decoder()) - |> pog.execute(store.connection) - |> result.map_error(QueryFailed) - |> result.try(fn(returned) { - let pog.Returned(_, rows) = returned - decode_rows(rows, store) - }) -} - -fn load_stream( - store: PostgresStore(event), - stream_id: String, -) -> Result(List(cqrs.RecordedEvent(event)), PostgresStoreError) { - load_stream_query - |> pog.query - |> pog.parameter(pog.text(stream_id)) - |> pog.returning(event_row_decoder()) - |> pog.execute(store.connection) - |> result.map_error(QueryFailed) - |> result.try(fn(returned) { - let pog.Returned(_, rows) = returned - decode_rows(rows, store) - }) -} - -fn lock_event_table( - connection: pog.Connection, -) -> Result(Nil, PostgresStoreError) { - lock_event_table_query - |> pog.query - |> pog.returning(int_row_decoder()) - |> pog.execute(connection) - |> result.map(fn(_) { Nil }) - |> result.map_error(QueryFailed) -} - -fn next_global_position( - connection: pog.Connection, -) -> Result(Int, PostgresStoreError) { - select_next_global_position_query - |> pog.query - |> pog.returning(int_row_decoder()) - |> pog.execute(connection) - |> result.map_error(QueryFailed) - |> result.try(fn(returned) { - single_row(returned, "select_next_global_position") - }) -} - -fn next_stream_position( - connection: pog.Connection, - stream_id: String, -) -> Result(Int, PostgresStoreError) { - select_next_stream_position_query - |> pog.query - |> pog.parameter(pog.text(stream_id)) - |> pog.returning(int_row_decoder()) - |> pog.execute(connection) - |> result.map_error(QueryFailed) - |> result.try(fn(returned) { - single_row(returned, "select_next_stream_position") - }) -} - -fn insert_events( - connection: pog.Connection, - store: PostgresStore(event), - events: List(cqrs.RecordedEvent(event)), -) -> Result(Nil, PostgresStoreError) { - list.try_each(events, fn(event) { - insert_event_query - |> pog.query - |> pog.parameter(pog.int(event.global_position)) - |> pog.parameter(pog.text(event.stream_id)) - |> pog.parameter(pog.int(event.stream_position)) - |> pog.parameter(pog.text(event.event_type)) - |> pog.parameter( - pog.text(json.to_string(store.event_to_json(event.payload))), - ) - |> pog.parameter(pog.text(metadata_to_json(event.metadata))) - |> pog.returning(nil_decoder()) - |> pog.execute(connection) - |> result.replace(Nil) - |> result.map_error(QueryFailed) - }) -} - -fn record_events( - events: List(cqrs.PendingEvent(event)), - stream_id: String, - metadata: List(#(String, String)), - next_global_position: Int, - next_stream_position: Int, -) -> List(cqrs.RecordedEvent(event)) { - let #(recorded_events, _, _) = - list.fold( - events, - #([], next_global_position, next_stream_position), - fn(accumulator, pending_event) { - let #(recorded_events, global_position, stream_position) = accumulator - let cqrs.PendingEvent(event_type, payload) = pending_event - - #( - [ - cqrs.RecordedEvent( - global_position: global_position, - stream_id: stream_id, - stream_position: stream_position, - event_type: event_type, - payload: payload, - metadata: metadata, - ), - ..recorded_events - ], - global_position + 1, - stream_position + 1, - ) - }, - ) - - list.reverse(recorded_events) -} - -fn decode_rows( - rows: List(EventRow), - store: PostgresStore(event), -) -> Result(List(cqrs.RecordedEvent(event)), PostgresStoreError) { - list.try_map(rows, fn(row) { decode_row(row, store) }) -} - -fn decode_row( - row: EventRow, - store: PostgresStore(event), -) -> Result(cqrs.RecordedEvent(event), PostgresStoreError) { - use payload <- result.try( - json.parse(row.payload, store.event_decoder) - |> result.map_error(fn(reason) { - PayloadCouldNotBeDecoded(event_type: row.event_type, reason: reason) - }), - ) - - use metadata <- result.try( - json.parse(row.metadata, metadata_decoder()) - |> result.map_error(MetadataCouldNotBeDecoded), - ) - - Ok(cqrs.RecordedEvent( - global_position: row.global_position, - stream_id: row.stream_id, - stream_position: row.stream_position, - event_type: row.event_type, - payload: payload, - metadata: metadata, - )) -} - -fn metadata_to_json(metadata: List(#(String, String))) -> String { - metadata - |> json.array(fn(pair) { - let #(key, value) = pair - json.object([#("key", json.string(key)), #("value", json.string(value))]) - }) - |> json.to_string -} - -fn metadata_decoder() -> decode.Decoder(List(#(String, String))) { - decode.list({ - use key <- decode.field("key", decode.string) - use value <- decode.field("value", decode.string) - decode.success(#(key, value)) - }) -} - -fn event_row_decoder() -> decode.Decoder(EventRow) { - use global_position <- decode.field(0, decode.int) - use stream_id <- decode.field(1, decode.string) - use stream_position <- decode.field(2, decode.int) - use event_type <- decode.field(3, decode.string) - use payload <- decode.field(4, decode.string) - use metadata <- decode.field(5, decode.string) - - decode.success(EventRow( - global_position: global_position, - stream_id: stream_id, - stream_position: stream_position, - event_type: event_type, - payload: payload, - metadata: metadata, - )) -} - -fn int_row_decoder() -> decode.Decoder(Int) { - decode.field(0, decode.int, decode.success) -} - -fn nil_decoder() -> decode.Decoder(Nil) { - decode.map(decode.dynamic, fn(_) { Nil }) -} - -fn single_row( - returned: pog.Returned(Int), - query: String, -) -> Result(Int, PostgresStoreError) { - case returned { - pog.Returned(_, [row]) -> Ok(row) - pog.Returned(_, []) -> Error(ExpectedSingleRow(query)) - pog.Returned(_, [_, ..]) -> Error(ExpectedSingleRow(query)) - } -} diff --git a/stores/cqrs_postgres/test/cqrs_postgres_test.gleam b/stores/cqrs_postgres/test/cqrs_postgres_test.gleam deleted file mode 100644 index afa37a9..0000000 --- a/stores/cqrs_postgres/test/cqrs_postgres_test.gleam +++ /dev/null @@ -1,337 +0,0 @@ -import cqrs -import cqrs/stores/postgres_store -import gleam/dynamic/decode -import gleam/erlang/process -import gleam/int -import gleam/json -import gleam/list -import gleam/result -import gleeunit -import global_value -import pog - -const database_url: String = "postgres://cqrs:cqrs@localhost:54329/cqrs_test" - -const stress_appends: Int = 300 - -const stress_queries: Int = 300 - -pub fn main() -> Nil { - gleeunit.main() -} - -type TestEvent { - UserRegistered(id: String, email: String) -} - -type TestGlobalData { - TestGlobalData(connection: pog.Connection) -} - -pub fn appends_and_loads_all_events_test() { - let system = start_system() - - let assert Ok(events) = - cqrs.append( - system, - "append-user-1", - [cqrs.PendingEvent("user_registered", UserRegistered("user-1", "a@b.c"))], - [#("request_id", "request-1")], - ) - - assert events - == [ - cqrs.RecordedEvent( - global_position: events |> first_global_position, - stream_id: "append-user-1", - stream_position: 1, - event_type: "user_registered", - payload: UserRegistered("user-1", "a@b.c"), - metadata: [#("request_id", "request-1")], - ), - ] - - let assert Ok(stored_events) = cqrs.load_stream(system, "append-user-1") - assert stored_events == events - cqrs.stop(system) -} - -pub fn load_stream_only_returns_matching_stream_test() { - let system = start_system() - - let assert Ok(_) = - cqrs.append( - system, - "load-user-1", - [cqrs.PendingEvent("user_registered", UserRegistered("user-1", "a@b.c"))], - [], - ) - - let assert Ok(_) = - cqrs.append( - system, - "load-user-2", - [cqrs.PendingEvent("user_registered", UserRegistered("user-2", "b@c.d"))], - [], - ) - - let assert Ok(events) = cqrs.load_stream(system, "load-user-2") - - assert events - == [ - cqrs.RecordedEvent( - global_position: events |> first_global_position, - stream_id: "load-user-2", - stream_position: 1, - event_type: "user_registered", - payload: UserRegistered("user-2", "b@c.d"), - metadata: [], - ), - ] - - cqrs.stop(system) -} - -pub fn works_with_cqrs_system_test() { - let system = start_system() - - let assert Ok(events) = - cqrs.append( - system, - "system-user-3", - [cqrs.PendingEvent("user_registered", UserRegistered("user-3", "c@d.e"))], - [], - ) - - assert events - == [ - cqrs.RecordedEvent( - global_position: events |> first_global_position, - stream_id: "system-user-3", - stream_position: 1, - event_type: "user_registered", - payload: UserRegistered("user-3", "c@d.e"), - metadata: [], - ), - ] - - let assert Ok(stored_events) = cqrs.load_stream(system, "system-user-3") - assert stored_events == events - cqrs.stop(system) -} - -pub fn simultaneous_appends_and_queries_stress_test() { - let system = start_system() - let results = process.new_subject() - - int.range(0, stress_appends, Nil, fn(_, index) { - let _pid = - process.spawn(fn() { - let result = - cqrs.append( - system, - "stress-stream", - [ - cqrs.PendingEvent( - "user_registered", - UserRegistered( - "stress-" <> int.to_string(index), - "stress-" <> int.to_string(index) <> "@example.com", - ), - ), - ], - [#("index", int.to_string(index))], - ) - - process.send(results, AppendCompleted(result)) - }) - - Nil - }) - - int.range(0, stress_queries, Nil, fn(_, _index) { - let _pid = - process.spawn(fn() { - process.send(results, QueryCompleted(cqrs.load_all(system))) - }) - - Nil - }) - - let outcomes = - receive_stress_outcomes(results, stress_appends + stress_queries, []) - let append_results = append_results(outcomes) - let query_results = query_results(outcomes) - - assert list.length(append_results) == stress_appends - assert list.length(query_results) == stress_queries - assert ok_count(append_results) == stress_appends - assert ok_count(query_results) == stress_queries - - let assert Ok(stored_events) = cqrs.load_stream(system, "stress-stream") - assert list.length(stored_events) == stress_appends - assert positions(stored_events, fn(event) { event.stream_position }) - == expected_positions(stress_appends) - cqrs.stop(system) -} - -fn global_data() -> TestGlobalData { - global_value.create_with_unique_name("cqrs_postgres_test.global_data", fn() { - let pool_name = process.new_name("cqrs_postgres_test_pool") - let assert Ok(config) = pog.url_config(pool_name, database_url) - let assert Ok(started) = - config - |> pog.pool_size(20) - |> pog.start - - process.unlink(started.pid) - let connection = started.data - let assert Ok(Nil) = postgres_store.create_table(connection) - reset_database(connection) - TestGlobalData(connection:) - }) -} - -type StressOutcome { - AppendCompleted( - Result( - List(cqrs.RecordedEvent(TestEvent)), - cqrs.Error(Nil, postgres_store.PostgresStoreError), - ), - ) - QueryCompleted( - Result( - List(cqrs.RecordedEvent(TestEvent)), - cqrs.Error(Nil, postgres_store.PostgresStoreError), - ), - ) -} - -fn receive_stress_outcomes( - results: process.Subject(StressOutcome), - remaining: Int, - outcomes: List(StressOutcome), -) -> List(StressOutcome) { - case remaining { - 0 -> outcomes - _ -> { - let assert Ok(outcome) = process.receive(results, 30_000) - receive_stress_outcomes(results, remaining - 1, [outcome, ..outcomes]) - } - } -} - -fn append_results( - outcomes: List(StressOutcome), -) -> List( - Result( - List(cqrs.RecordedEvent(TestEvent)), - cqrs.Error(Nil, postgres_store.PostgresStoreError), - ), -) { - outcomes - |> list.filter_map(fn(outcome) { - case outcome { - AppendCompleted(result) -> Ok(result) - QueryCompleted(_) -> Error(Nil) - } - }) -} - -fn query_results( - outcomes: List(StressOutcome), -) -> List( - Result( - List(cqrs.RecordedEvent(TestEvent)), - cqrs.Error(Nil, postgres_store.PostgresStoreError), - ), -) { - outcomes - |> list.filter_map(fn(outcome) { - case outcome { - AppendCompleted(_) -> Error(Nil) - QueryCompleted(result) -> Ok(result) - } - }) -} - -fn ok_count(results: List(Result(a, b))) -> Int { - results - |> list.filter(result.is_ok) - |> list.length -} - -fn positions( - events: List(cqrs.RecordedEvent(event)), - get_position: fn(cqrs.RecordedEvent(event)) -> Int, -) -> List(Int) { - events - |> list.map(get_position) - |> list.sort(by: int.compare) -} - -fn expected_positions(count: Int) -> List(Int) { - int.range(1, count + 1, [], list.prepend) - |> list.reverse -} - -fn start_system() -> cqrs.System(TestEvent, postgres_store.PostgresStoreError) { - let globals = global_data() - let assert Ok(started) = - cqrs.start( - process.new_name("cqrs_postgres_registry"), - timeout: 30_000, - event_store: store(globals.connection), - ) - - started.data -} - -fn first_global_position(events: List(cqrs.RecordedEvent(event))) -> Int { - let assert [event] = events - event.global_position -} - -fn reset_database(connection: pog.Connection) -> Nil { - let decoder = decode.map(decode.dynamic, fn(_) { Nil }) - let assert Ok(_) = - "TRUNCATE TABLE cqrs_event;" - |> pog.query - |> pog.returning(decoder) - |> pog.execute(connection) - - Nil -} - -fn store( - connection: pog.Connection, -) -> cqrs.EventStore( - postgres_store.PostgresStore(TestEvent), - TestEvent, - postgres_store.PostgresStoreError, -) { - postgres_store.new(connection, event_to_json, event_decoder()) -} - -fn event_to_json(event: TestEvent) -> json.Json { - case event { - UserRegistered(id, email) -> - json.object([ - #("type", json.string("user_registered")), - #("id", json.string(id)), - #("email", json.string(email)), - ]) - } -} - -fn event_decoder() -> decode.Decoder(TestEvent) { - use event_type <- decode.field("type", decode.string) - case event_type { - "user_registered" -> { - use id <- decode.field("id", decode.string) - use email <- decode.field("email", decode.string) - decode.success(UserRegistered(id, email)) - } - _ -> decode.failure(UserRegistered("", ""), expected: "TestEvent") - } -} diff --git a/test/cqrs_test.gleam b/test/cqrs_test.gleam deleted file mode 100644 index d534c8c..0000000 --- a/test/cqrs_test.gleam +++ /dev/null @@ -1,292 +0,0 @@ -import cqrs -import gleam/erlang/process -import gleam/otp/actor -import gleam/otp/static_supervisor -import gleam/result -import gleeunit - -pub fn main() -> Nil { - gleeunit.main() -} - -type Command { - RegisterUser(id: String, email: String) - RejectRegistration(reason: String) -} - -type Event { - UserRegistered(id: String, email: String) -} - -type DomainError { - RegistrationRejected(reason: String) -} - -pub fn execute_maps_command_to_events_test() { - let assert Ok(started) = start_system() - let system = started.data - - let assert Ok(events) = - cqrs.execute( - system, - "user-2", - RegisterUser("user-2", "b@c.d"), - [], - handle_command, - ) - - assert events - == [ - cqrs.RecordedEvent( - global_position: 1, - stream_id: "user-2", - stream_position: 1, - event_type: "user_registered", - payload: UserRegistered("user-2", "b@c.d"), - metadata: [], - ), - ] - - cqrs.stop(system) -} - -pub fn execute_returns_domain_error_without_committing_test() { - let assert Ok(started) = start_system() - let system = started.data - - let assert Error(cqrs.CommandRejected(RegistrationRejected("closed"))) = - cqrs.execute( - system, - "user-3", - RejectRegistration("closed"), - [], - handle_command, - ) - - let assert Ok(events) = cqrs.load_all(system) - assert events == [] - cqrs.stop(system) -} - -pub fn load_stream_filters_recorded_events_test() { - let assert Ok(started) = start_system() - let system = started.data - - let assert Ok(_) = - cqrs.append( - system, - "user-4", - [ - cqrs.PendingEvent("user_registered", UserRegistered("user-4", "d@e.f")), - ], - [], - ) - - let assert Ok(_) = - cqrs.append( - system, - "user-5", - [ - cqrs.PendingEvent("user_registered", UserRegistered("user-5", "e@f.g")), - ], - [], - ) - - let assert Ok(events) = cqrs.load_stream(system, "user-5") - - assert events - == [ - cqrs.RecordedEvent( - global_position: 2, - stream_id: "user-5", - stream_position: 1, - event_type: "user_registered", - payload: UserRegistered("user-5", "e@f.g"), - metadata: [], - ), - ] - - cqrs.stop(system) -} - -pub fn subscribers_receive_committed_events_test() { - let assert Ok(started) = start_system() - let system = started.data - let all_events = cqrs.subscribe(system, cqrs.AllEvents) - let stream_events = cqrs.subscribe(system, cqrs.StreamEvents("user-6")) - - let assert Ok(committed_events) = - cqrs.append( - system, - "user-6", - [cqrs.PendingEvent("user_registered", UserRegistered("user-6", "f@g.h"))], - [], - ) - - assert process.receive(all_events, 1000) - == Ok(cqrs.EventsCommitted(committed_events)) - assert process.receive(stream_events, 1000) - == Ok(cqrs.EventsCommitted(committed_events)) - - cqrs.stop(system) -} - -pub fn custom_event_store_is_used_for_persistence_test() { - let assert Ok(started) = - cqrs.start( - process.new_name("cqrs_registry"), - timeout: cqrs.default_timeout, - event_store: failing_load_store(), - ) - - let system = started.data - - let assert Ok(_) = - cqrs.append( - system, - "user-7", - [cqrs.PendingEvent("user_registered", UserRegistered("user-7", "g@h.i"))], - [], - ) - - let assert Error(cqrs.EventStoreFailed("load_all failed")) = - cqrs.load_all(system) - cqrs.stop(system) -} - -pub fn started_event_handler_receives_committed_events_test() { - let assert Ok(started) = start_system() - let system = started.data - let received = process.new_subject() - let assert Ok(handler) = - cqrs.start_handler(system, "projection", cqrs.AllEvents, fn(events) { - process.send(received, events) - Ok(Nil) - }) - - let assert Ok(committed_events) = - cqrs.append( - system, - "user-8", - [cqrs.PendingEvent("user_registered", UserRegistered("user-8", "h@i.j"))], - [], - ) - - assert process.receive(received, 1000) == Ok(committed_events) - cqrs.stop_handler(handler.data) - cqrs.stop(system) -} - -pub fn event_handler_failures_are_observable_test() { - let assert Ok(started) = start_system() - let system = started.data - let assert Ok(handler) = - cqrs.start_handler(system, "projection", cqrs.AllEvents, fn(_events) { - Error("projection failed") - }) - - let assert Ok(_) = - cqrs.append( - system, - "user-9", - [cqrs.PendingEvent("user_registered", UserRegistered("user-9", "i@j.k"))], - [], - ) - - process.sleep(50) - - assert cqrs.handler_failures(handler.data, timeout: cqrs.default_timeout) - == Ok([ - cqrs.HandlerFailure(handler: "projection", reason: "projection failed"), - ]) - - cqrs.stop_handler(handler.data) - cqrs.stop(system) -} - -pub fn supervised_system_can_run_under_static_supervisor_test() { - let spec = - cqrs.supervised( - registry_name: process.new_name("cqrs_registry"), - timeout: cqrs.default_timeout, - event_store: cqrs.memory_store(), - ) - - let assert Ok(started) = spec.start() - let system = started.data - - let assert Ok(events) = - cqrs.append( - system, - "supervised-user", - [ - cqrs.PendingEvent( - "user_registered", - UserRegistered("supervised-user", "supervised@example.com"), - ), - ], - [], - ) - - assert events - == [ - cqrs.RecordedEvent( - global_position: 1, - stream_id: "supervised-user", - stream_position: 1, - event_type: "user_registered", - payload: UserRegistered("supervised-user", "supervised@example.com"), - metadata: [], - ), - ] - - let supervised_spec = - cqrs.supervised( - registry_name: process.new_name("cqrs_supervised_registry"), - timeout: cqrs.default_timeout, - event_store: cqrs.memory_store(), - ) - - let assert Ok(started) = - static_supervisor.new(static_supervisor.OneForOne) - |> static_supervisor.add(supervised_spec) - |> static_supervisor.start - - assert process.is_alive(started.pid) -} - -fn start_system() -> actor.StartResult(cqrs.System(Event, Nil)) { - cqrs.start( - process.new_name("cqrs_registry"), - timeout: cqrs.default_timeout, - event_store: cqrs.memory_store(), - ) -} - -fn handle_command( - command: Command, -) -> Result(List(cqrs.PendingEvent(Event)), DomainError) { - case command { - RegisterUser(id, email) -> - Ok([cqrs.PendingEvent("user_registered", UserRegistered(id, email))]) - RejectRegistration(reason) -> Error(RegistrationRejected(reason)) - } -} - -fn failing_load_store() -> cqrs.EventStore( - cqrs.MemoryStore(Event), - Event, - String, -) { - let memory_store = cqrs.memory_store() - - cqrs.EventStore( - store: memory_store.store, - append: fn(store, stream_id, events, metadata) { - memory_store.append(store, stream_id, events, metadata) - |> result.map_error(fn(_reason) { "append failed" }) - }, - load_all: fn(_store) { Error("load_all failed") }, - load_stream: fn(_store, _stream_id) { Error("load_stream failed") }, - ) -} diff --git a/test/factos_test.gleam b/test/factos_test.gleam new file mode 100644 index 0000000..d0e1b8d --- /dev/null +++ b/test/factos_test.gleam @@ -0,0 +1,155 @@ +import factos +import gleam/int +import gleeunit +import youid/uuid + +pub fn main() -> Nil { + gleeunit.main() +} + +type Event { + UsernameReserved(username: String) + UserRegistered(username: String) + DisplayNameChanged(name: String) +} + +type State { + UsernameAvailable + UsernameTaken +} + +pub fn fold_replays_recorded_events_test() { + let events = [ + recorded( + UsernameReserved("renata"), + [factos.tag("username:renata")], + revision: 0, + ), + recorded( + UserRegistered("renata"), + [factos.tag("username:renata")], + revision: 1, + ), + ] + + assert factos.fold(UsernameAvailable, events, evolve) == UsernameTaken +} + +pub fn query_matches_by_type_and_tags_test() { + let username_query = + factos.query([ + factos.query_item( + types: [ + factos.event_type("UsernameReserved"), + factos.event_type("UserRegistered"), + ], + tags: [factos.tag("username:renata")], + ), + ]) + + let matching = + recorded( + UsernameReserved("renata"), + [factos.tag("username:renata")], + revision: 0, + ) + let wrong_tag = + recorded( + UsernameReserved("lucy"), + [factos.tag("username:lucy")], + revision: 1, + ) + let wrong_type = + recorded( + DisplayNameChanged("Renata"), + [factos.tag("username:renata")], + revision: 2, + ) + + assert factos.matches_query(matching, username_query) + assert !factos.matches_query(wrong_tag, username_query) + assert !factos.matches_query(wrong_type, username_query) +} + +pub fn query_items_are_or_combined_test() { + let query = + factos.query([ + factos.query_item(types: [factos.event_type("UserRegistered")], tags: [ + factos.tag("user:1"), + ]), + factos.query_item(types: [factos.event_type("DisplayNameChanged")], tags: [ + factos.tag("user:2"), + ]), + ]) + + let event = + recorded(DisplayNameChanged("R"), [factos.tag("user:2")], revision: 0) + + assert factos.matches_query(event, query) +} + +pub fn empty_query_matches_all_events_test() { + let query = factos.query([]) + let event = recorded(DisplayNameChanged("R"), [], revision: 0) + + assert factos.matches_query(event, query) +} + +pub fn highest_position_keeps_later_position_test() { + let early = factos.SequencePosition(commit_position: 10, prepare_position: 10) + let later = factos.SequencePosition(commit_position: 11, prepare_position: 0) + + assert factos.highest_position(early, later) == later + assert factos.highest_position(factos.NoPosition, early) == early +} + +pub fn revision_models_empty_and_loaded_streams_test() { + assert revision_label(factos.NoEvents) == "empty" + assert revision_label(factos.CurrentRevision(2)) == "revision:2" +} + +fn evolve(state: State, event: Event) -> State { + case state, event { + UsernameAvailable, UsernameReserved(_) -> UsernameTaken + UsernameAvailable, UserRegistered(_) -> UsernameTaken + UsernameAvailable, DisplayNameChanged(_) -> state + UsernameTaken, UsernameReserved(_) -> state + UsernameTaken, UserRegistered(_) -> state + UsernameTaken, DisplayNameChanged(_) -> state + } +} + +fn recorded( + event: Event, + tags tags: List(factos.Tag), + revision revision: Int, +) -> factos.Recorded(Event) { + let type_ = case event { + UsernameReserved(_) -> factos.event_type("UsernameReserved") + UserRegistered(_) -> factos.event_type("UserRegistered") + DisplayNameChanged(_) -> factos.event_type("DisplayNameChanged") + } + + factos.Recorded( + id: uuid.v7(), + stream: "user-1", + revision: revision, + position: factos.SequencePosition( + commit_position: revision, + prepare_position: revision, + ), + metadata: [], + custom_metadata: <<>>, + data: <<>>, + type_: type_, + tags: tags, + event: event, + ) +} + +fn revision_label(revision: factos.Revision) -> String { + case revision { + factos.NoEvents -> "empty" + factos.CurrentRevision(revision) -> "revision:" <> int.to_string(revision) + } +}