diff --git a/README.md b/README.md index 64001c1..a29555c 100644 --- a/README.md +++ b/README.md @@ -32,7 +32,8 @@ One dependency is optional: ## A minimal example -Define a router that registers a service and the module that handles it: +Define a router that registers a service and the module that handles it, +and register the services (typically from your application's `start/2`). ```elixir defmodule MyApp.Router do @@ -51,14 +52,13 @@ defmodule MyApp.Greeter do {:reply, Message.ok(message, "Hello, #{name}!")} end end + +:ok = MyApp.Router.register_services() ``` -Register the services (typically from your application's `start/2`) and dispatch a -message to the service by name: +Dispatch a message to the service by name: ```elixir -:ok = MyApp.Router.register_services() - :greet |> X3m.System.Message.new(raw_request: %{"name" => "Ada"}) |> X3m.System.Dispatcher.dispatch() @@ -78,6 +78,30 @@ flowchart LR No aggregates or event store are involved here — any module registered through a router can be a dispatch target. +## Dispatch accross the cluster + +If you want to try dispatch accross the cluster, run 2 iex sessions: + +```bash +iex --sname x3m_1@localhost -S mix +``` + +and from the other terminal + +```bash +iex --sname x3m_2@localhost -S mix +``` + +Define your router and message handler (Greeter here) and register your services in first iex session (like in example from above), +and then from the second one, connect to the first node: + +```elixir +Node.connect :"x3m_1@localhost" +#=> true +``` + +... and then dispatch message the same way you did in previous example. Result will be the same. + ## Guides - [Getting started](guides/getting-started.md) — install, optional deps, your first dispatch. @@ -86,6 +110,12 @@ router can be a dispatch target. - [Distribution](guides/distribution.md) — service discovery across nodes, choosing the node, and forwarding. - [Scheduling](guides/scheduling.md) — persistable, future-dated message delivery with `Scheduler`. +## Example + +The [Banking example](examples/bank/) is a complete poncho project demonstrating the +full CQRS/ES flow — HTTP API, command aggregates, event store, listener-driven read +model, and cross-node dispatch. See its README for setup and curl walkthrough. + ## License Released under the MIT License. See the [LICENSE](https://github.com/x3m-ex/system/blob/master/LICENSE) file. diff --git a/examples/bank/.gitignore b/examples/bank/.gitignore new file mode 100644 index 0000000..8fe9c19 --- /dev/null +++ b/examples/bank/.gitignore @@ -0,0 +1,9 @@ +_volumes/pg_data/* +!_volumes/pg_data/.gitkeep +_volumes/es_data/* +!_volumes/es_data/.gitkeep +_build/ +deps/ +*.beam +*.ez +.elixir_ls/ diff --git a/examples/bank/README.md b/examples/bank/README.md new file mode 100644 index 0000000..ef80456 --- /dev/null +++ b/examples/bank/README.md @@ -0,0 +1,248 @@ +# Banking Example + +A reference application demonstrating [x3m_system](../../README.md)'s full CQRS/ES flow. + +Poncho project with five independent apps communicating via `X3m.System.Dispatcher`: + +| App | Purpose | +|-----|---------| +| `banking` | Shared library (Identity, logger config, mix bless) | +| `banking_event_store` | Extreme TCP wrapper for EventStore | +| `banking_core` | Account aggregate (open, deposit, withdraw, close) | +| `banking_listeners` | Event listener → PG read model + query services | +| `banking_api` | Bandit + Plug HTTP API | + +## Architecture + +Each app runs as a separate Erlang node. The Dispatcher crosses node boundaries +transparently — the API node doesn't know (or care) which node hosts a given service. + +```mermaid +graph TB + subgraph node_api["node: banking_api@localhost"] + Router["Plug.Router"] + SI["ServiceInvoker"] + D["Dispatcher"] + end + + subgraph node_core["node: banking_core@localhost"] + CR["Core.Router"] + MH["MessageHandler"] + Agg["Account Aggregate"] + end + + subgraph node_listeners["node: banking_listeners@localhost"] + LR["Listeners.Router"] + Listener["Extreme.Listener"] + Den["Denormalizer"] + QS["Query Services"] + end + + subgraph infra["infrastructure (Docker)"] + ES[(EventStore)] + PG[(PostgreSQL)] + end + + Router --> SI + SI --> D + D -.->|"cross-node call"| CR + D -.->|"cross-node call"| LR + CR -->|authorize + invoke| MH + MH --> Agg + Agg -->|events| ES + ES -->|subscription| Listener + Listener --> Den + Den -->|upsert| PG + LR -->|authorize + invoke| QS + QS -->|read| PG +``` + +## Command flow (write) + +A command enters through HTTP, crosses node boundaries via the Dispatcher, +and reaches the aggregate which emits events persisted to EventStore. + +```mermaid +sequenceDiagram + participant HTTP as curl / client + + box rgb(40,60,90) banking_api@localhost + participant API as Api.Router + participant D as Dispatcher + end + + box rgb(60,40,90) banking_core@localhost + participant CR as Core.Router + participant MH as MessageHandler + participant Agg as Account Aggregate + end + + participant ES as EventStore + + HTTP->>API: POST /accounts (JSON) + API->>D: Dispatcher.dispatch(message) + D->>CR: find :open_account service (cross-node) + CR->>CR: authorize/1 + CR->>MH: invoke open_account + MH->>Agg: handle_msg (validate → execute) + Agg->>Agg: emit Opened event + Agg-->>MH: {:block, msg, state} + MH->>ES: persist events + ES-->>MH: {:ok, version} + MH->>Agg: apply_event (update state) + MH-->>D: {:created, id, version} + D-->>API: response + API-->>HTTP: 201 {"id": "..."} +``` + +## Query flow (read) + +Queries dispatch to listener services on a different node that read from +the PostgreSQL projection — completely separate from the aggregate path. + +```mermaid +sequenceDiagram + participant HTTP as curl / client + + box rgb(40,60,90) banking_api@localhost + participant API as Api.Router + participant D as Dispatcher + end + + box rgb(40,90,60) banking_listeners@localhost + participant LR as Listeners.Router + participant QS as GetAccount + end + + participant PG as PostgreSQL + + HTTP->>API: GET /accounts/:id + API->>D: Dispatcher.dispatch(message) + D->>LR: find :get_account service (cross-node) + LR->>LR: authorize/1 + LR->>QS: invoke get_account + QS->>PG: Repo.get(Account, id) + PG-->>QS: row + QS-->>D: {:ok, data} + D-->>API: response + API-->>HTTP: 200 {"id": "...", "balance": 70, ...} +``` + +## Prerequisites + +- Elixir >= 1.17 +- Docker & Docker Compose + +## Setup + +Start infrastructure: + + docker compose up -d + +Install dependencies and compile each app: + + for app in core listeners api; do + (cd apps/$app && mix deps.get) + done + +## Running + +Start each app in a separate terminal. Order doesn't matter. + +**Terminal 1 — Core (aggregates):** + + cd apps/core && ./run.sh + +**Terminal 2 — Listeners (read model):** + + cd apps/listeners && ./run.sh + +**Terminal 3 — API (HTTP):** + + cd apps/api && ./run.sh + +The apps auto-connect via the shared `apps/.iex.exs` (Erlang distribution with `--sname`). + +Browse persisted events in the EventStore web UI at +[http://localhost:2113/web/index.html#/streams/$ce-accounts](http://localhost:2113/web/index.html#/streams/$ce-accounts). + +## Usage (curl) + +Open an account and capture the returned id: + + export ACC=$(curl -s -X POST http://localhost:4001/accounts \ + -H "Content-Type: application/json" \ + -d "{\"owner_id\": \"user-1\", \"invoked_by\": {\"user_id\": \"user-1\"}}" \ + | grep -o '"id":"[^"]*"' | cut -d'"' -f4) + + # → 201 {"id": ""} + +### Deposit + + curl -sw "\n%{http_code}" -X PUT "http://localhost:4001/accounts/$ACC/deposit" \ + -H "Content-Type: application/json" \ + -d '{"amount": 100, "invoked_by": {"user_id": "user-1"}}' + + # → 204 + +### Withdraw + + curl -sw "\n%{http_code}" -X PUT "http://localhost:4001/accounts/$ACC/withdraw" \ + -H "Content-Type: application/json" \ + -d '{"amount": 30, "invoked_by": {"user_id": "user-1"}}' + + # → 204 + +### Get account + + curl -sw "\n%{http_code}" "http://localhost:4001/accounts/$ACC" + + # → 200 {"id": "", "owner_id": "user-1", "balance": 70, "status": "open"} + +### List accounts (admin only) + + curl -sw "\n%{http_code}" "http://localhost:4001/accounts?admin=true" + + # → 200 [{"id": "", "owner_id": "user-1", "balance": 70, "status": "open"}, ...] + +### Close account + + curl -sw "\n%{http_code}" -X PUT "http://localhost:4001/accounts/$ACC/close" \ + -H "Content-Type: application/json" \ + -d '{"invoked_by": {"admin?": true}}' + + # → 204 + # If balance was > 0, a Withdrawn event is emitted first, then Closed. + +### Error responses + +Withdraw more than the balance: + + curl -sw "\n%{http_code}" -X PUT "http://localhost:4001/accounts/$ACC/withdraw" \ + -H "Content-Type: application/json" \ + -d '{"amount": 9999, "invoked_by": {"user_id": "user-1"}}' + + # → 422 {"errors": {"amount": ["insufficient funds"]}} + +Wrong user: + + curl -sw "\n%{http_code}" -X PUT "http://localhost:4001/accounts/$ACC/deposit" \ + -H "Content-Type: application/json" \ + -d '{"amount": 10, "invoked_by": {"user_id": "other-user"}}' + + # → 403 {"error": "forbidden"} + +Close an already-closed account: + + curl -sw "\n%{http_code}" -X PUT "http://localhost:4001/accounts/$ACC/close" \ + -H "Content-Type: application/json" \ + -d '{"invoked_by": {"user_id": "user-1"}}' + + # → 409 {"error": "account is closed"} + +Missing required field: + + curl -sw "\n%{http_code}" -X POST http://localhost:4001/accounts \ + -H "Content-Type: application/json" -d '{}' + + # → 422 {"errors": {"owner_id": ["can't be blank"]}} diff --git a/examples/bank/_volumes/es_data/.gitkeep b/examples/bank/_volumes/es_data/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/examples/bank/_volumes/pg_data/.gitkeep b/examples/bank/_volumes/pg_data/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/examples/bank/apps/.iex.exs b/examples/bank/apps/.iex.exs new file mode 100644 index 0000000..c1efd7c --- /dev/null +++ b/examples/bank/apps/.iex.exs @@ -0,0 +1,34 @@ +defmodule BankingClusterBoot do + @moduledoc false + + def connect_to_peers do + {:ok, hostname} = :inet.gethostname() + self_node = Node.self() + + peers = + [:banking_core_1, :banking_listeners_1, :banking_api_1] + |> Enum.map(&:"#{&1}@#{hostname}") + |> Enum.reject(&(&1 == self_node)) + + _wait_for_peers(peers, _attempts = 30) + end + + defp _wait_for_peers(_peers, 0) do + IO.puts("[cluster] timed out waiting for peers; Node.list/0 = #{inspect(Node.list())}") + end + + defp _wait_for_peers(peers, attempts) do + connected = + peers + |> Enum.filter(&(Node.connect(&1) == true)) + + if connected != [] do + IO.puts("[cluster] connected to #{inspect(connected)}; Node.list/0 = #{inspect(Node.list())}") + else + Process.sleep(500) + _wait_for_peers(peers, attempts - 1) + end + end +end + +spawn(fn -> BankingClusterBoot.connect_to_peers() end) diff --git a/examples/bank/apps/api/.formatter.exs b/examples/bank/apps/api/.formatter.exs new file mode 100644 index 0000000..af4e262 --- /dev/null +++ b/examples/bank/apps/api/.formatter.exs @@ -0,0 +1,4 @@ +[ + import_deps: [:plug, :x3m_system], + inputs: ["{config,lib,test}/**/*.{ex,exs}"] +] diff --git a/examples/bank/apps/api/config/config.exs b/examples/bank/apps/api/config/config.exs new file mode 100644 index 0000000..9ed628a --- /dev/null +++ b/examples/bank/apps/api/config/config.exs @@ -0,0 +1,7 @@ +import Config + +import_config "../../banking/config/config.exs" + +config :banking_api, port: 4001 + +import_config "#{Mix.env()}.exs" diff --git a/examples/bank/apps/api/config/dev.exs b/examples/bank/apps/api/config/dev.exs new file mode 100644 index 0000000..becde76 --- /dev/null +++ b/examples/bank/apps/api/config/dev.exs @@ -0,0 +1 @@ +import Config diff --git a/examples/bank/apps/api/config/test.exs b/examples/bank/apps/api/config/test.exs new file mode 100644 index 0000000..becde76 --- /dev/null +++ b/examples/bank/apps/api/config/test.exs @@ -0,0 +1 @@ +import Config diff --git a/examples/bank/apps/api/lib/application.ex b/examples/bank/apps/api/lib/application.ex new file mode 100644 index 0000000..07dbbc1 --- /dev/null +++ b/examples/bank/apps/api/lib/application.ex @@ -0,0 +1,19 @@ +defmodule Banking.Api.Application do + @moduledoc false + use Application + + def start(_type, _args) do + Banking.TelemetryLogger.setup() + + port = + System.get_env("BANKING_API_PORT", "4001") + |> String.to_integer() + + children = [ + {Bandit, plug: Banking.Api.Router, port: port} + ] + + opts = [strategy: :one_for_one, name: Banking.Api.MainSupervisor] + Supervisor.start_link(children, opts) + end +end diff --git a/examples/bank/apps/api/lib/router.ex b/examples/bank/apps/api/lib/router.ex new file mode 100644 index 0000000..b663345 --- /dev/null +++ b/examples/bank/apps/api/lib/router.ex @@ -0,0 +1,55 @@ +defmodule Banking.Api.Router do + @moduledoc !""" + HTTP router — Plug.Router with all account endpoints. + """ + use Plug.Router + alias Banking.Api.ServiceInvoker + + plug Plug.RequestId + plug Plug.Logger + plug :match + + plug Plug.Parsers, + parsers: [:json], + json_decoder: Jason + + plug :dispatch + + # Commands + post "/accounts" do + ServiceInvoker.invoke(conn, :open_account, conn.body_params) + end + + put "/accounts/:id/deposit" do + conn.body_params + |> Map.put("id", id) + |> then(&ServiceInvoker.invoke(conn, :deposit, &1)) + end + + put "/accounts/:id/withdraw" do + conn.body_params + |> Map.put("id", id) + |> then(&ServiceInvoker.invoke(conn, :withdraw, &1)) + end + + put "/accounts/:id/close" do + conn.body_params + |> Map.put("id", id) + |> then(&ServiceInvoker.invoke(conn, :close_account, &1)) + end + + # Queries + get "/accounts/:id" do + ServiceInvoker.invoke(conn, :get_account, %{"id" => id}) + end + + get "/accounts" do + ServiceInvoker.invoke(conn, :list_accounts, conn.query_params) + end + + match _ do + conn + |> put_resp_content_type("application/json") + |> send_resp(404, Jason.encode!(%{error: "not_found"})) + end +end diff --git a/examples/bank/apps/api/lib/service_invoker.ex b/examples/bank/apps/api/lib/service_invoker.ex new file mode 100644 index 0000000..7a4c095 --- /dev/null +++ b/examples/bank/apps/api/lib/service_invoker.ex @@ -0,0 +1,100 @@ +defmodule Banking.Api.ServiceInvoker do + @moduledoc !""" + Builds an X3m.System.Message from an HTTP request, dispatches it, + and sends the JSON response. + """ + alias X3m.System.Message, as: SysMsg + alias X3m.System.Dispatcher + alias Banking.Identity + alias Plug.Conn + + @spec invoke(conn :: Conn.t(), service_name :: atom(), params :: map()) :: Conn.t() + def invoke(%Conn{} = conn, service_name, params) do + req_id = + conn + |> Conn.get_resp_header("x-request-id") + |> List.first() + + identity = _build_identity(params) + raw_request = Map.delete(params, "invoked_by") + + service_name + |> SysMsg.new(raw_request: raw_request, id: req_id) + |> SysMsg.assign(:invoked_by, identity) + |> Dispatcher.dispatch() + |> _respond(conn) + end + + defp _build_identity(%{"invoked_by" => %{"user_id" => uid} = invoked_by}) do + %Identity{ + user_id: uid, + admin?: Map.get(invoked_by, "admin?", false) + } + end + + defp _build_identity(%{"invoked_by" => %{"admin?" => true}}) do + %Identity{admin?: true} + end + + defp _build_identity(%{"admin" => "true"}), do: %Identity{admin?: true} + + defp _build_identity(_), do: %Identity{} + + # Success + + defp _respond(%SysMsg{response: {:created, id, _version}}, conn), + do: _json(conn, 201, %{id: id}) + + defp _respond(%SysMsg{response: {:ok, version}}, conn) when is_integer(version) do + conn + |> Conn.put_resp_content_type("application/json") + |> Conn.send_resp(204, "") + end + + defp _respond(%SysMsg{response: {:ok, data}}, conn), + do: _json(conn, 200, data) + + # Validation + + defp _respond(%SysMsg{response: {:validation_error, changeset}}, conn) do + errors = Banking.ErrorHelpers.traverse_errors(changeset) + _json(conn, 422, %{errors: errors}) + end + + # Domain errors + + defp _respond(%SysMsg{response: {:error, :not_found}}, conn), + do: _json(conn, 404, %{error: "not_found"}) + + defp _respond(%SysMsg{response: {:error, :forbidden}}, conn), + do: _json(conn, 403, %{error: "forbidden"}) + + defp _respond(%SysMsg{response: {:error, {:conflict, message}}}, conn), + do: _json(conn, 409, %{error: message}) + + defp _respond(%SysMsg{response: {:error, :key_already_registered, _}}, conn), + do: _json(conn, 409, %{error: "aggregate already exists"}) + + defp _respond(%SysMsg{response: {:error, {:error, :wrong_expected_version, _}}}, conn), + do: _json(conn, 409, %{error: "aggregate already exists"}) + + defp _respond(%SysMsg{response: {:error, reason}}, conn) when is_atom(reason), + do: _json(conn, 422, %{error: to_string(reason)}) + + defp _respond(%SysMsg{response: {:error, reason}}, conn), + do: _json(conn, 422, %{error: inspect(reason)}) + + # Infrastructure + + defp _respond(%SysMsg{response: {:service_unavailable, _}}, conn), + do: _json(conn, 503, %{error: "service_unavailable"}) + + defp _respond(%SysMsg{halted?: true, response: response}, conn), + do: _json(conn, 422, %{error: inspect(response)}) + + defp _json(conn, status, body) do + conn + |> Conn.put_resp_content_type("application/json") + |> Conn.send_resp(status, Jason.encode!(body)) + end +end diff --git a/examples/bank/apps/api/mix.exs b/examples/bank/apps/api/mix.exs new file mode 100644 index 0000000..0ef3928 --- /dev/null +++ b/examples/bank/apps/api/mix.exs @@ -0,0 +1,39 @@ +defmodule Banking.Api.MixProject do + use Mix.Project + + def project do + [ + app: :banking_api, + version: "0.1.0", + elixir: "~> 1.17", + start_permanent: Mix.env() == :prod, + deps: _deps(), + elixirc_paths: ["lib"], + dialyzer: [plt_add_deps: :apps_direct] + ] + end + + def cli do + [preferred_envs: [bless: :test]] + end + + def application do + [ + extra_applications: [:logger], + mod: {Banking.Api.Application, []} + ] + end + + defp _deps do + [ + {:bandit, "~> 1.0"}, + {:plug, "~> 1.16"}, + {:jason, "~> 1.4"}, + {:x3m_system, path: "../../../.."}, + {:dialyxir, "~> 1.0", only: :dev, runtime: false}, + + # local + {:banking, path: "../banking"} + ] + end +end diff --git a/examples/bank/apps/api/mix.lock b/examples/bank/apps/api/mix.lock new file mode 100644 index 0000000..f961111 --- /dev/null +++ b/examples/bank/apps/api/mix.lock @@ -0,0 +1,17 @@ +%{ + "bandit": {:hex, :bandit, "1.12.0", "6c5214daa2469644ac4ab0113b98abc24f75e348378e6a974c6343b3e5da22ef", [:mix], [{:hpax, "~> 1.0", [hex: :hpax, repo: "hexpm", optional: false]}, {:plug, "~> 1.18", [hex: :plug, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}, {:thousand_island, "~> 1.5", [hex: :thousand_island, repo: "hexpm", optional: false]}, {:websock, "~> 0.5", [hex: :websock, repo: "hexpm", optional: false]}], "hexpm", "45dac82dc86f45cf4a196dee9cc5a8b791d9c9469d996055f055e6ee36c66e20"}, + "db_connection": {:hex, :db_connection, "2.10.1", "d5465f6bcc125c1b8981c1dbf23c193ca16f446ec0b25832dc174f74f18be510", [:mix], [{:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "18ed94c6e627b4bf452dbd4df61b69a35a1e768525140bc1917b7a685026a6a3"}, + "decimal": {:hex, :decimal, "3.1.1", "430d87b04011ce6cbd4fd205be758311a81f87d552d40904abd00f015935b1d0", [:mix], [], "hexpm", "c5f25f2ced74a0587d03e6023f595db8e924c9d3922c8c8ffd9edfc4498cf1f6"}, + "dialyxir": {:hex, :dialyxir, "1.4.7", "dda948fcee52962e4b6c5b4b16b2d8fa7d50d8645bbae8b8685c3f9ecb7f5f4d", [:mix], [{:erlex, ">= 0.2.8", [hex: :erlex, repo: "hexpm", optional: false]}], "hexpm", "b34527202e6eb8cee198efec110996c25c5898f43a4094df157f8d28f27d9efe"}, + "ecto": {:hex, :ecto, "3.14.0", "2fa64521eebfcb2670d907a86e4ad947290e9933706bb315e6fb5c21b172cb26", [:mix], [{:decimal, "~> 3.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "130d69ffb4285f9ce4792b65dfbb994fd13ea4cbc3cbea2524b199aa3de84af3"}, + "ecto_sql": {:hex, :ecto_sql, "3.14.0", "06446ab8410d2f85bfbb80857ee224ab3b693700cbb38f6535d507449a627b2e", [:mix], [{:db_connection, "~> 2.9", [hex: :db_connection, repo: "hexpm", optional: false]}, {:decimal, "~> 3.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:ecto, "~> 3.14.0", [hex: :ecto, repo: "hexpm", optional: false]}, {:myxql, "~> 0.8", [hex: :myxql, repo: "hexpm", optional: true]}, {:postgrex, "~> 0.19 or ~> 1.0", [hex: :postgrex, repo: "hexpm", optional: true]}, {:tds, "~> 2.1.1 or ~> 2.2", [hex: :tds, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4.0 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "f4d8d36faf294c9417b5a37ec7ac8217ee2abdef5fcf197ba690f361548d3949"}, + "erlex": {:hex, :erlex, "0.2.9", "7debbbaa9f4f368b8cd648983e0f1d7963028508e9c59e9d4ed504e94ef52a55", [:mix], [], "hexpm", "8cfffc0ec7159e6d73de2ab28a588064de80f88b2798d5cbe4482cbbc200178b"}, + "hpax": {:hex, :hpax, "1.0.3", "ed67ef51ad4df91e75cc6a1494f851850c0bd98ebc0be6e81b026e765ee535aa", [:mix], [], "hexpm", "8eab6e1cfa8d5918c2ce4ba43588e894af35dbd8e91e6e55c817bca5847df34a"}, + "jason": {:hex, :jason, "1.4.5", "2e3a008590b0b8d7388c20293e9dcc9cf3e5d642fd2a114e4cbbb52e595d940a", [:mix], [{:decimal, "~> 1.0 or ~> 2.0 or ~> 3.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "b0c823996102bcd0239b3c2444eb00409b72f6a140c1950bc8b457d836b30684"}, + "mime": {:hex, :mime, "2.0.7", "b8d739037be7cd402aee1ba0306edfdef982687ee7e9859bee6198c1e7e2f128", [:mix], [], "hexpm", "6171188e399ee16023ffc5b76ce445eb6d9672e2e241d2df6050f3c771e80ccd"}, + "plug": {:hex, :plug, "1.19.2", "e4950525b22c6789dfb38a3f95d47171ba159da3fc5a33be9643b43d5e8adb98", [:mix], [{:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:plug_crypto, "~> 1.1.1 or ~> 1.2 or ~> 2.0", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "b6fce20a56af5e60fa5dfecf3f907bb98ec981be43c79a3809a499bc3d133de0"}, + "plug_crypto": {:hex, :plug_crypto, "2.1.1", "19bda8184399cb24afa10be734f84a16ea0a2bc65054e23a62bb10f06bc89491", [:mix], [], "hexpm", "6470bce6ffe41c8bd497612ffde1a7e4af67f36a15eea5f921af71cf3e11247c"}, + "telemetry": {:hex, :telemetry, "1.4.2", "a0cb522801dffb1c49fe6e30561badffc7b6d0e180db1300df759faa22062855", [:rebar3], [], "hexpm", "928f6495066506077862c0d1646609eed891a4326bee3126ba54b60af61febb1"}, + "thousand_island": {:hex, :thousand_island, "1.5.0", "f50a213cac97262b6d5ebb85745aa2c00fec1413191e6e66834788d45425cecb", [:mix], [{:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "708923d40523e43cf99041ab37a0d4b0ec426ac6438fa3716ab23d919eaeb412"}, + "websock": {:hex, :websock, "0.5.3", "2f69a6ebe810328555b6fe5c831a851f485e303a7c8ce6c5f675abeb20ebdadc", [:mix], [], "hexpm", "6105453d7fac22c712ad66fab1d45abdf049868f253cf719b625151460b8b453"}, +} diff --git a/examples/bank/apps/api/run.sh b/examples/bank/apps/api/run.sh new file mode 100755 index 0000000..b833475 --- /dev/null +++ b/examples/bank/apps/api/run.sh @@ -0,0 +1,10 @@ +#!/bin/bash + +if [ "$1" ] ; then + num="$1" +else + num=1 +fi + +BANKING_API_PORT=400$num \ +iex --dot-iex ../.iex.exs --sname banking_api_$num -S mix diff --git a/examples/bank/apps/banking/.formatter.exs b/examples/bank/apps/banking/.formatter.exs new file mode 100644 index 0000000..044c4d9 --- /dev/null +++ b/examples/bank/apps/banking/.formatter.exs @@ -0,0 +1,4 @@ +[ + import_deps: [:ecto], + inputs: ["{config,lib,test}/**/*.{ex,exs}"] +] diff --git a/examples/bank/apps/banking/config/config.exs b/examples/bank/apps/banking/config/config.exs new file mode 100644 index 0000000..2047596 --- /dev/null +++ b/examples/bank/apps/banking/config/config.exs @@ -0,0 +1,8 @@ +import Config + +config :logger, :console, + format: "$date $time $metadata[$level] $message\n", + level: :debug, + metadata: [:request_id, :corr_id, :pid] + +import_config "#{Mix.env()}.exs" diff --git a/examples/bank/apps/banking/config/dev.exs b/examples/bank/apps/banking/config/dev.exs new file mode 100644 index 0000000..becde76 --- /dev/null +++ b/examples/bank/apps/banking/config/dev.exs @@ -0,0 +1 @@ +import Config diff --git a/examples/bank/apps/banking/config/test.exs b/examples/bank/apps/banking/config/test.exs new file mode 100644 index 0000000..63787d4 --- /dev/null +++ b/examples/bank/apps/banking/config/test.exs @@ -0,0 +1,3 @@ +import Config + +config :logger, level: :warning diff --git a/examples/bank/apps/banking/lib/banking.ex b/examples/bank/apps/banking/lib/banking.ex new file mode 100644 index 0000000..131a83d --- /dev/null +++ b/examples/bank/apps/banking/lib/banking.ex @@ -0,0 +1,5 @@ +defmodule Banking do + @moduledoc """ + Shared library for the Banking example application. + """ +end diff --git a/examples/bank/apps/banking/lib/error_helpers.ex b/examples/bank/apps/banking/lib/error_helpers.ex new file mode 100644 index 0000000..496925e --- /dev/null +++ b/examples/bank/apps/banking/lib/error_helpers.ex @@ -0,0 +1,10 @@ +defmodule Banking.ErrorHelpers do + @moduledoc !""" + Converts Ecto changesets into JSON-friendly error maps. + """ + + @spec traverse_errors(changeset :: Ecto.Changeset.t()) :: %{atom() => [String.t()]} + def traverse_errors(%Ecto.Changeset{} = changeset) do + Ecto.Changeset.traverse_errors(changeset, fn {msg, _opts} -> msg end) + end +end diff --git a/examples/bank/apps/banking/lib/identity.ex b/examples/bank/apps/banking/lib/identity.ex new file mode 100644 index 0000000..fcf157a --- /dev/null +++ b/examples/bank/apps/banking/lib/identity.ex @@ -0,0 +1,12 @@ +defmodule Banking.Identity do + @moduledoc """ + Represents the caller's identity, passed through message assigns. + """ + + @type t() :: %__MODULE__{ + user_id: String.t() | nil, + admin?: boolean() + } + + defstruct [:user_id, admin?: false] +end diff --git a/examples/bank/apps/banking/lib/migrations.ex b/examples/bank/apps/banking/lib/migrations.ex new file mode 100644 index 0000000..ff35d14 --- /dev/null +++ b/examples/bank/apps/banking/lib/migrations.ex @@ -0,0 +1,52 @@ +defmodule Banking.Migrations do + @moduledoc !""" + Creates and migrates Ecto repos on application start. + """ + require Logger + + @spec create(app :: atom()) :: :ok + def create(app) do + for repo <- _repos(app) do + repo.__adapter__ + |> apply(:storage_up, [repo.config]) + |> case do + :ok -> + Logger.info(fn -> "The database for #{inspect(repo)} has been created" end) + + {:error, :already_up} -> + Logger.debug(fn -> "The database for #{inspect(repo)} has already been created" end) + + {:error, term} when is_binary(term) -> + Logger.warning(fn -> + "The database for #{inspect(repo)} couldn't be created: #{term}" + end) + + {:error, term} -> + Logger.warning(fn -> + "The database for #{inspect(repo)} couldn't be created: #{inspect(term)}" + end) + end + end + + :ok + end + + @spec up(app :: atom()) :: :ok + def up(app) do + for repo <- _repos(app) do + {:ok, _, _} = + Ecto.Migrator + |> apply(:with_repo, [repo, &_run_migrator(&1, :up, all: true)]) + end + + :ok + end + + defp _repos(app) do + Application.load(app) + Application.fetch_env!(app, :ecto_repos) + end + + defp _run_migrator(repo, direction, opts), + do: apply(Ecto.Migrator, :run, [repo, direction, opts]) +end diff --git a/examples/bank/apps/banking/lib/mix/tasks/bless.ex b/examples/bank/apps/banking/lib/mix/tasks/bless.ex new file mode 100644 index 0000000..7417f2f --- /dev/null +++ b/examples/bank/apps/banking/lib/mix/tasks/bless.ex @@ -0,0 +1,22 @@ +defmodule Mix.Tasks.Bless do + @moduledoc !""" + Runs all checks required to push project to repo. + """ + use Mix.Task + + @shortdoc "Runs all checks (compile, format, test, dialyzer)" + def run(_) do + [ + {"compile", ["--force"]}, + {"format", ["--check-formatted"]}, + {"test", []}, + {"dialyzer", []} + ] + |> Enum.each(fn {task, args} -> + IO.ANSI.format([:cyan, "Running #{task} with args #{inspect(args)}"]) + |> IO.puts() + + Mix.Task.run(task, args) + end) + end +end diff --git a/examples/bank/apps/banking/lib/telemetry_logger.ex b/examples/bank/apps/banking/lib/telemetry_logger.ex new file mode 100644 index 0000000..ce703fe --- /dev/null +++ b/examples/bank/apps/banking/lib/telemetry_logger.ex @@ -0,0 +1,117 @@ +defmodule Banking.TelemetryLogger do + @moduledoc !""" + Attaches to x3m_system telemetry events and logs them. + """ + require Logger + + @spec setup() :: :ok + def setup do + events = [ + # cluster + [:x3m, :system, :node_joined], + [:x3m, :system, :node_left], + + # service discovery + dispatch + [:x3m, :system, :discovering_service], + [:x3m, :system, :service_found], + [:x3m, :system, :service_not_found], + [:x3m, :system, :service_request_received], + [:x3m, :system, :invoking_service], + [:x3m, :system, :service_responded], + [:x3m, :system, :executing_service], + [:x3m, :system, :execution_finished], + + # aggregate + [:x3m, :system, :handle_msg], + [:x3m, :system, :new_aggr_spawned], + [:x3m, :system, :aggregate_commit_timeout] + ] + + :telemetry.attach_many("banking-logger", events, &__MODULE__.handle_event/4, nil) + :ok + end + + # Cluster + + def handle_event([:x3m, :system, :node_joined], _, %{node: node}, _), + do: Logger.info(fn -> "[Node] #{node} joined" end) + + def handle_event([:x3m, :system, :node_left], _, %{node: node}, _), + do: Logger.info(fn -> "[Node] #{node} left" end) + + # Service discovery + dispatch + + def handle_event([:x3m, :system, :discovering_service], _, %{message: msg}, _), + do: Logger.debug(fn -> "[Dispatch] Discovering #{msg.service_name}" end) + + def handle_event([:x3m, :system, :service_found], measurements, %{message: msg, service_node: node}, _) do + duration = _format_duration(measurements[:duration]) + Logger.debug(fn -> "[Dispatch] #{msg.service_name} found on #{node} in #{duration}" end) + end + + def handle_event([:x3m, :system, :service_not_found], measurements, %{message: msg}, _) do + duration = _format_duration(measurements[:duration]) + Logger.warning(fn -> "[Dispatch] #{msg.service_name} not found in #{duration}" end) + end + + def handle_event([:x3m, :system, :invoking_service], _, %{message: msg, service_node: node}, _), + do: Logger.info(fn -> "[Dispatch] Invoking #{msg.service_name} on #{node}" end) + + def handle_event([:x3m, :system, :service_responded], measurements, %{message: msg}, _) do + duration = _format_duration(measurements[:duration]) + Logger.info(fn -> "[Dispatch] #{msg.service_name} responded in #{duration}" end) + end + + # Service execution (on the receiving node) + + def handle_event([:x3m, :system, :service_request_received], _, %{service: service, origin_node: origin}, _), + do: Logger.info(fn -> "[Service] Received #{service} from #{origin}" end) + + def handle_event([:x3m, :system, :executing_service], _, %{service: service}, _), + do: Logger.debug(fn -> "[Service] Executing #{service}" end) + + def handle_event([:x3m, :system, :execution_finished], measurements, %{message: msg}, _) do + duration = _format_duration(measurements[:duration]) + Logger.info(fn -> "[Service] #{msg.service_name} finished in #{duration}" end) + end + + # Aggregate + + def handle_event([:x3m, :system, :handle_msg], _, %{aggregate: aggregate, message: message}, _), + do: Logger.info(fn -> "[Aggregate] Handling #{inspect(message)} on #{inspect(aggregate)}" end) + + def handle_event([:x3m, :system, :new_aggr_spawned], _, %{id: id}, _), + do: Logger.info(fn -> "[Aggregate] New process spawned for #{id}" end) + + def handle_event( + [:x3m, :system, :aggregate_commit_timeout], + %{timeout_after: timeout}, + %{aggregate: aggregate, message: message}, + _ + ), + do: + Logger.error(fn -> + "[Aggregate] Commit timeout for #{inspect(aggregate)}.#{inspect(message)} after #{timeout}ms" + end) + + # Duration formatting + + defp _format_duration(nil), do: "?" + + defp _format_duration(duration) do + cond do + duration < 1_000 -> + "#{duration}ns" + + duration < 1_000_000 -> + duration + |> System.convert_time_unit(:native, :microsecond) + |> then(&"#{&1}µs") + + true -> + duration + |> System.convert_time_unit(:native, :millisecond) + |> then(&"#{&1}ms") + end + end +end diff --git a/examples/bank/apps/banking/mix.exs b/examples/bank/apps/banking/mix.exs new file mode 100644 index 0000000..7c02dff --- /dev/null +++ b/examples/bank/apps/banking/mix.exs @@ -0,0 +1,26 @@ +defmodule Banking.MixProject do + use Mix.Project + + def project do + [ + app: :banking, + version: "0.1.0", + elixir: "~> 1.17", + start_permanent: false, + deps: _deps(), + elixirc_paths: ["lib"] + ] + end + + def application do + [extra_applications: [:logger]] + end + + defp _deps do + [ + {:ecto, "~> 3.12"}, + {:ecto_sql, "~> 3.12"}, + {:telemetry, "~> 1.0"} + ] + end +end diff --git a/examples/bank/apps/banking/mix.lock b/examples/bank/apps/banking/mix.lock new file mode 100644 index 0000000..f13123b --- /dev/null +++ b/examples/bank/apps/banking/mix.lock @@ -0,0 +1,7 @@ +%{ + "db_connection": {:hex, :db_connection, "2.10.1", "d5465f6bcc125c1b8981c1dbf23c193ca16f446ec0b25832dc174f74f18be510", [:mix], [{:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "18ed94c6e627b4bf452dbd4df61b69a35a1e768525140bc1917b7a685026a6a3"}, + "decimal": {:hex, :decimal, "3.1.1", "430d87b04011ce6cbd4fd205be758311a81f87d552d40904abd00f015935b1d0", [:mix], [], "hexpm", "c5f25f2ced74a0587d03e6023f595db8e924c9d3922c8c8ffd9edfc4498cf1f6"}, + "ecto": {:hex, :ecto, "3.14.0", "2fa64521eebfcb2670d907a86e4ad947290e9933706bb315e6fb5c21b172cb26", [:mix], [{:decimal, "~> 3.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "130d69ffb4285f9ce4792b65dfbb994fd13ea4cbc3cbea2524b199aa3de84af3"}, + "ecto_sql": {:hex, :ecto_sql, "3.14.0", "06446ab8410d2f85bfbb80857ee224ab3b693700cbb38f6535d507449a627b2e", [:mix], [{:db_connection, "~> 2.9", [hex: :db_connection, repo: "hexpm", optional: false]}, {:decimal, "~> 3.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:ecto, "~> 3.14.0", [hex: :ecto, repo: "hexpm", optional: false]}, {:myxql, "~> 0.8", [hex: :myxql, repo: "hexpm", optional: true]}, {:postgrex, "~> 0.19 or ~> 1.0", [hex: :postgrex, repo: "hexpm", optional: true]}, {:tds, "~> 2.1.1 or ~> 2.2", [hex: :tds, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4.0 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "f4d8d36faf294c9417b5a37ec7ac8217ee2abdef5fcf197ba690f361548d3949"}, + "telemetry": {:hex, :telemetry, "1.4.2", "a0cb522801dffb1c49fe6e30561badffc7b6d0e180db1300df759faa22062855", [:rebar3], [], "hexpm", "928f6495066506077862c0d1646609eed891a4326bee3126ba54b60af61febb1"}, +} diff --git a/examples/bank/apps/core/.formatter.exs b/examples/bank/apps/core/.formatter.exs new file mode 100644 index 0000000..88715a2 --- /dev/null +++ b/examples/bank/apps/core/.formatter.exs @@ -0,0 +1,4 @@ +[ + import_deps: [:ecto, :x3m_system], + inputs: ["{config,lib,test}/**/*.{ex,exs}"] +] diff --git a/examples/bank/apps/core/config/config.exs b/examples/bank/apps/core/config/config.exs new file mode 100644 index 0000000..66615e4 --- /dev/null +++ b/examples/bank/apps/core/config/config.exs @@ -0,0 +1,13 @@ +import Config + +import_config "../../banking/config/config.exs" + +config :banking_core, Banking.Core.EventStore, + db_type: "node", + host: "localhost", + port: "1113", + username: "admin", + password: "changeit", + connection_name: "banking_core" + +import_config "#{Mix.env()}.exs" diff --git a/examples/bank/apps/core/config/dev.exs b/examples/bank/apps/core/config/dev.exs new file mode 100644 index 0000000..becde76 --- /dev/null +++ b/examples/bank/apps/core/config/dev.exs @@ -0,0 +1 @@ +import Config diff --git a/examples/bank/apps/core/config/test.exs b/examples/bank/apps/core/config/test.exs new file mode 100644 index 0000000..becde76 --- /dev/null +++ b/examples/bank/apps/core/config/test.exs @@ -0,0 +1 @@ +import Config diff --git a/examples/bank/apps/core/lib/aggregate_repo.ex b/examples/bank/apps/core/lib/aggregate_repo.ex new file mode 100644 index 0000000..df0ab05 --- /dev/null +++ b/examples/bank/apps/core/lib/aggregate_repo.ex @@ -0,0 +1,7 @@ +defmodule Banking.Core.AggregateRepo do + @moduledoc !""" + Aggregate event persistence via EventStore. + """ + use Banking.EventStore.AggregateRepo, + extreme: Banking.Core.EventStore +end diff --git a/examples/bank/apps/core/lib/aggregates/account/aggregate.ex b/examples/bank/apps/core/lib/aggregates/account/aggregate.ex new file mode 100644 index 0000000..0ad861c --- /dev/null +++ b/examples/bank/apps/core/lib/aggregates/account/aggregate.ex @@ -0,0 +1,103 @@ +defmodule Banking.Core.Aggregates.Account.Aggregate do + @moduledoc """ + Account aggregate — handles commands and applies events. + + All commands require the caller to be either the account owner or an admin + (instance authorization via `admin_or_owner?/2` guard). + """ + use X3m.System.Aggregate + alias X3m.System.Message, as: SysMsg + alias Banking.Core.Aggregates.Account.{State, Commands, Events} + + defguardp admin_or_owner?(assigns, owner_id) + when assigns.invoked_by.admin? == true or + assigns.invoked_by.user_id == owner_id + + @impl X3m.System.Aggregate + def initial_state, do: %State{} + + handle_msg :open_account, &Commands.Open.new/2, fn + %SysMsg{request: cmd} = msg, %State{} = state -> + event = %Events.Opened{id: cmd.id, owner_id: cmd.owner_id} + + msg = + msg + |> SysMsg.add_event(event) + |> SysMsg.created(cmd.id) + + {:block, msg, state} + end + + handle_msg :deposit, &Commands.Deposit.new/2, fn + %SysMsg{assigns: assigns, request: cmd} = msg, %State{owner_id: owner_id} = state + when admin_or_owner?(assigns, owner_id) -> + event = %Events.Deposited{id: cmd.id, amount: cmd.amount} + + msg = + msg + |> SysMsg.add_event(event) + |> SysMsg.ok() + + {:block, msg, state} + + %SysMsg{} = msg, %State{} = state -> + {:noblock, SysMsg.error(msg, :forbidden), state} + end + + handle_msg :withdraw, &Commands.Withdraw.new/2, fn + %SysMsg{assigns: assigns, request: cmd} = msg, + %State{owner_id: owner_id} = state + when admin_or_owner?(assigns, owner_id) -> + event = %Events.Withdrawn{id: cmd.id, amount: cmd.amount} + + msg = + msg + |> SysMsg.add_event(event) + |> SysMsg.ok() + + {:block, msg, state} + + %SysMsg{} = msg, %State{} = state -> + {:noblock, SysMsg.error(msg, :forbidden), state} + end + + handle_msg :close_account, fn + %SysMsg{} = msg, %State{closed?: true} = state -> + {:noblock, SysMsg.error(msg, {:conflict, "account is closed"}), state} + + %SysMsg{assigns: assigns} = msg, %State{owner_id: owner_id} = state + when admin_or_owner?(assigns, owner_id) -> + _close(msg, state) + + %SysMsg{} = msg, %State{} = state -> + {:noblock, SysMsg.error(msg, :forbidden), state} + end + + def processed_message_id(%{message_id: id}), do: id + + def apply_event(%Events.Opened{} = e, %State{} = state), + do: %State{state | id: e.id, owner_id: e.owner_id} + + def apply_event(%Events.Deposited{} = e, %State{} = state), + do: %State{state | balance: state.balance + e.amount} + + def apply_event(%Events.Withdrawn{} = e, %State{} = state), + do: %State{state | balance: state.balance - e.amount} + + def apply_event(%Events.Closed{}, %State{} = state), + do: %State{state | closed?: true} + + defp _close(%SysMsg{} = msg, %State{id: id, balance: balance} = state) do + events = + if balance > 0, + do: [%Events.Withdrawn{id: id, amount: balance}, %Events.Closed{id: id}], + else: [%Events.Closed{id: id}] + + msg = + events + |> Enum.reduce(msg, fn event, acc -> SysMsg.add_event(acc, event) end) + |> SysMsg.ok() + + {:block, msg, state} + end +end diff --git a/examples/bank/apps/core/lib/aggregates/account/command_helpers.ex b/examples/bank/apps/core/lib/aggregates/account/command_helpers.ex new file mode 100644 index 0000000..3482c3c --- /dev/null +++ b/examples/bank/apps/core/lib/aggregates/account/command_helpers.ex @@ -0,0 +1,16 @@ +defmodule Banking.Core.Aggregates.Account.CommandHelpers do + @moduledoc !""" + Shared helper for converting Ecto changesets into message requests. + """ + alias X3m.System.Message, as: SysMsg + + @spec put_request(changeset :: Ecto.Changeset.t(), msg :: SysMsg.t()) :: SysMsg.t() + def put_request(%Ecto.Changeset{valid?: false} = changeset, %SysMsg{} = msg), + do: SysMsg.put_request(changeset, msg) + + def put_request(%Ecto.Changeset{valid?: true} = changeset, %SysMsg{} = msg) do + changeset + |> Ecto.Changeset.apply_changes() + |> SysMsg.put_request(msg) + end +end diff --git a/examples/bank/apps/core/lib/aggregates/account/commands/deposit.ex b/examples/bank/apps/core/lib/aggregates/account/commands/deposit.ex new file mode 100644 index 0000000..397d686 --- /dev/null +++ b/examples/bank/apps/core/lib/aggregates/account/commands/deposit.ex @@ -0,0 +1,31 @@ +defmodule Banking.Core.Aggregates.Account.Commands.Deposit do + @moduledoc !""" + Validates the deposit command. Rejects if account is closed. + """ + use Ecto.Schema + import Ecto.Changeset + alias X3m.System.Message, as: SysMsg + alias Banking.Core.Aggregates.Account.{CommandHelpers, State} + + @primary_key false + embedded_schema do + field :id, :string + field :amount, :integer + end + + @spec new(msg :: SysMsg.t(), state :: State.t()) :: SysMsg.t() + def new(%SysMsg{} = msg, %State{} = state) do + %__MODULE__{} + |> cast(msg.raw_request, [:id, :amount]) + |> validate_required([:id, :amount]) + |> validate_number(:amount, greater_than: 0) + |> _reject_if_closed(state) + |> CommandHelpers.put_request(msg) + end + + defp _reject_if_closed(changeset, %State{closed?: true}), + do: add_error(changeset, :id, "account is closed") + + defp _reject_if_closed(changeset, %State{}), + do: changeset +end diff --git a/examples/bank/apps/core/lib/aggregates/account/commands/open.ex b/examples/bank/apps/core/lib/aggregates/account/commands/open.ex new file mode 100644 index 0000000..f75ea1a --- /dev/null +++ b/examples/bank/apps/core/lib/aggregates/account/commands/open.ex @@ -0,0 +1,23 @@ +defmodule Banking.Core.Aggregates.Account.Commands.Open do + @moduledoc !""" + Validates the open_account command. + """ + use Ecto.Schema + import Ecto.Changeset + alias X3m.System.Message, as: SysMsg + alias Banking.Core.Aggregates.Account.CommandHelpers + + @primary_key false + embedded_schema do + field :id, :string + field :owner_id, :string + end + + @spec new(msg :: SysMsg.t(), state :: term()) :: SysMsg.t() + def new(%SysMsg{} = msg, _state) do + %__MODULE__{} + |> cast(msg.raw_request, [:id, :owner_id]) + |> validate_required([:id, :owner_id]) + |> CommandHelpers.put_request(msg) + end +end diff --git a/examples/bank/apps/core/lib/aggregates/account/commands/withdraw.ex b/examples/bank/apps/core/lib/aggregates/account/commands/withdraw.ex new file mode 100644 index 0000000..65ecd85 --- /dev/null +++ b/examples/bank/apps/core/lib/aggregates/account/commands/withdraw.ex @@ -0,0 +1,43 @@ +defmodule Banking.Core.Aggregates.Account.Commands.Withdraw do + @moduledoc !""" + Validates the withdraw command. Rejects if account is closed. + """ + use Ecto.Schema + import Ecto.Changeset + alias X3m.System.Message, as: SysMsg + alias Banking.Core.Aggregates.Account.{CommandHelpers, State} + + @primary_key false + embedded_schema do + field :id, :string + field :amount, :integer + end + + @spec new(msg :: SysMsg.t(), state :: State.t()) :: SysMsg.t() + def new(%SysMsg{} = msg, %State{} = state) do + %__MODULE__{} + |> cast(msg.raw_request, [:id, :amount]) + |> validate_required([:id, :amount]) + |> validate_number(:amount, greater_than: 0) + |> _reject_if_closed(state) + |> _reject_if_insufficient(state) + |> CommandHelpers.put_request(msg) + end + + defp _reject_if_closed(changeset, %State{closed?: true}), + do: add_error(changeset, :id, "account is closed") + + defp _reject_if_closed(changeset, %State{}), + do: changeset + + defp _reject_if_insufficient(%Ecto.Changeset{valid?: false} = changeset, _state), + do: changeset + + defp _reject_if_insufficient(changeset, %State{balance: balance}) do + amount = get_change(changeset, :amount, 0) + + if amount > balance, + do: add_error(changeset, :amount, "insufficient funds"), + else: changeset + end +end diff --git a/examples/bank/apps/core/lib/aggregates/account/events/closed.ex b/examples/bank/apps/core/lib/aggregates/account/events/closed.ex new file mode 100644 index 0000000..f997a96 --- /dev/null +++ b/examples/bank/apps/core/lib/aggregates/account/events/closed.ex @@ -0,0 +1,6 @@ +defmodule Banking.Core.Aggregates.Account.Events.Closed do + @moduledoc !""" + Emitted when an account is closed. + """ + defstruct [:id] +end diff --git a/examples/bank/apps/core/lib/aggregates/account/events/deposited.ex b/examples/bank/apps/core/lib/aggregates/account/events/deposited.ex new file mode 100644 index 0000000..042405d --- /dev/null +++ b/examples/bank/apps/core/lib/aggregates/account/events/deposited.ex @@ -0,0 +1,6 @@ +defmodule Banking.Core.Aggregates.Account.Events.Deposited do + @moduledoc !""" + Emitted when funds are deposited into an account. + """ + defstruct [:id, :amount] +end diff --git a/examples/bank/apps/core/lib/aggregates/account/events/opened.ex b/examples/bank/apps/core/lib/aggregates/account/events/opened.ex new file mode 100644 index 0000000..42c2280 --- /dev/null +++ b/examples/bank/apps/core/lib/aggregates/account/events/opened.ex @@ -0,0 +1,6 @@ +defmodule Banking.Core.Aggregates.Account.Events.Opened do + @moduledoc !""" + Emitted when a new account is created. + """ + defstruct [:id, :owner_id] +end diff --git a/examples/bank/apps/core/lib/aggregates/account/events/withdrawn.ex b/examples/bank/apps/core/lib/aggregates/account/events/withdrawn.ex new file mode 100644 index 0000000..3c88939 --- /dev/null +++ b/examples/bank/apps/core/lib/aggregates/account/events/withdrawn.ex @@ -0,0 +1,6 @@ +defmodule Banking.Core.Aggregates.Account.Events.Withdrawn do + @moduledoc !""" + Emitted when funds are withdrawn from an account. + """ + defstruct [:id, :amount] +end diff --git a/examples/bank/apps/core/lib/aggregates/account/state.ex b/examples/bank/apps/core/lib/aggregates/account/state.ex new file mode 100644 index 0000000..83e3b34 --- /dev/null +++ b/examples/bank/apps/core/lib/aggregates/account/state.ex @@ -0,0 +1,14 @@ +defmodule Banking.Core.Aggregates.Account.State do + @moduledoc """ + Account aggregate state — rebuilt from the event stream. + """ + + @type t() :: %__MODULE__{ + id: String.t() | nil, + owner_id: String.t() | nil, + balance: non_neg_integer(), + closed?: boolean() + } + + defstruct [:id, :owner_id, balance: 0, closed?: false] +end diff --git a/examples/bank/apps/core/lib/application.ex b/examples/bank/apps/core/lib/application.ex new file mode 100644 index 0000000..8926211 --- /dev/null +++ b/examples/bank/apps/core/lib/application.ex @@ -0,0 +1,30 @@ +defmodule Banking.Core.Application do + @moduledoc false + use Application + + def start(_type, _args) do + Banking.TelemetryLogger.setup() + :ok = Banking.Core.Router.register_services() + + children = _get_children() + + opts = [strategy: :one_for_one, name: Banking.Core.MainSupervisor] + Supervisor.start_link(children, opts) + end + + case Mix.env() do + :test -> + defp _get_children, do: [] + + _ -> + defp _get_children do + event_store_config = + Application.get_env(:banking_core, Banking.Core.EventStore) + + [ + {Banking.Core.EventStore, event_store_config}, + {X3m.System.LocalAggregatesSupervision, [Banking.Core.LocalAggregates, Banking.Core]} + ] + end + end +end diff --git a/examples/bank/apps/core/lib/event_store.ex b/examples/bank/apps/core/lib/event_store.ex new file mode 100644 index 0000000..1cf8208 --- /dev/null +++ b/examples/bank/apps/core/lib/event_store.ex @@ -0,0 +1,6 @@ +defmodule Banking.Core.EventStore do + @moduledoc !""" + Extreme TCP connection to EventStore for the core aggregate app. + """ + use Extreme, otp_app: :banking_core +end diff --git a/examples/bank/apps/core/lib/local_aggregates.ex b/examples/bank/apps/core/lib/local_aggregates.ex new file mode 100644 index 0000000..97fd779 --- /dev/null +++ b/examples/bank/apps/core/lib/local_aggregates.ex @@ -0,0 +1,8 @@ +defmodule Banking.Core.LocalAggregates do + @moduledoc !""" + Registers aggregate supervision trees for this node. + """ + use X3m.System.LocalAggregates, [ + Banking.Core.Aggregates.Account.Aggregate + ] +end diff --git a/examples/bank/apps/core/lib/message_handler.ex b/examples/bank/apps/core/lib/message_handler.ex new file mode 100644 index 0000000..ce1f717 --- /dev/null +++ b/examples/bank/apps/core/lib/message_handler.ex @@ -0,0 +1,16 @@ +defmodule Banking.Core.MessageHandler do + @moduledoc !""" + Wires commands to the Account aggregate via x3m_system's MessageHandler macro. + """ + use X3m.System.MessageHandler, + aggregate_mod: Banking.Core.Aggregates.Account.Aggregate, + aggregate_repo: Banking.Core.AggregateRepo, + stream: "accounts", + pid_facade_mod: X3m.System.AggregatePidFacade, + event_metadata: %{app_version: "0.1.0"} + + on_new_aggregate :open_account + on_aggregate :deposit + on_aggregate :withdraw + on_aggregate :close_account +end diff --git a/examples/bank/apps/core/lib/router.ex b/examples/bank/apps/core/lib/router.ex new file mode 100644 index 0000000..4365d76 --- /dev/null +++ b/examples/bank/apps/core/lib/router.ex @@ -0,0 +1,14 @@ +defmodule Banking.Core.Router do + @moduledoc !""" + Registers command services and routes messages to the account message handler. + """ + use X3m.System.Router + alias Banking.Core.MessageHandler, as: Account + + service :open_account, Account + service :deposit, Account + service :withdraw, Account + service :close_account, Account + + def authorize(%X3m.System.Message{}), do: :ok +end diff --git a/examples/bank/apps/core/mix.exs b/examples/bank/apps/core/mix.exs new file mode 100644 index 0000000..41b22e8 --- /dev/null +++ b/examples/bank/apps/core/mix.exs @@ -0,0 +1,41 @@ +defmodule Banking.Core.MixProject do + use Mix.Project + + def project do + [ + app: :banking_core, + version: "0.1.0", + elixir: "~> 1.17", + start_permanent: Mix.env() == :prod, + deps: _deps(), + elixirc_paths: _elixirc_paths(Mix.env()), + dialyzer: [plt_add_deps: :apps_direct] + ] + end + + def cli do + [preferred_envs: [bless: :test]] + end + + def application do + [ + extra_applications: [:logger], + mod: {Banking.Core.Application, []} + ] + end + + defp _elixirc_paths(:test), do: ["lib", "test/support"] + defp _elixirc_paths(_), do: ["lib"] + + defp _deps do + [ + {:x3m_system, path: "../../../.."}, + {:ecto, "~> 3.12"}, + {:dialyxir, "~> 1.0", only: :dev, runtime: false}, + + # local + {:banking_event_store, path: "../event_store"}, + {:banking, path: "../banking"} + ] + end +end diff --git a/examples/bank/apps/core/mix.lock b/examples/bank/apps/core/mix.lock new file mode 100644 index 0000000..547623d --- /dev/null +++ b/examples/bank/apps/core/mix.lock @@ -0,0 +1,14 @@ +%{ + "db_connection": {:hex, :db_connection, "2.10.1", "d5465f6bcc125c1b8981c1dbf23c193ca16f446ec0b25832dc174f74f18be510", [:mix], [{:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "18ed94c6e627b4bf452dbd4df61b69a35a1e768525140bc1917b7a685026a6a3"}, + "decimal": {:hex, :decimal, "3.1.1", "430d87b04011ce6cbd4fd205be758311a81f87d552d40904abd00f015935b1d0", [:mix], [], "hexpm", "c5f25f2ced74a0587d03e6023f595db8e924c9d3922c8c8ffd9edfc4498cf1f6"}, + "dialyxir": {:hex, :dialyxir, "1.4.7", "dda948fcee52962e4b6c5b4b16b2d8fa7d50d8645bbae8b8685c3f9ecb7f5f4d", [:mix], [{:erlex, ">= 0.2.8", [hex: :erlex, repo: "hexpm", optional: false]}], "hexpm", "b34527202e6eb8cee198efec110996c25c5898f43a4094df157f8d28f27d9efe"}, + "ecto": {:hex, :ecto, "3.14.0", "2fa64521eebfcb2670d907a86e4ad947290e9933706bb315e6fb5c21b172cb26", [:mix], [{:decimal, "~> 3.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "130d69ffb4285f9ce4792b65dfbb994fd13ea4cbc3cbea2524b199aa3de84af3"}, + "ecto_sql": {:hex, :ecto_sql, "3.14.0", "06446ab8410d2f85bfbb80857ee224ab3b693700cbb38f6535d507449a627b2e", [:mix], [{:db_connection, "~> 2.9", [hex: :db_connection, repo: "hexpm", optional: false]}, {:decimal, "~> 3.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:ecto, "~> 3.14.0", [hex: :ecto, repo: "hexpm", optional: false]}, {:myxql, "~> 0.8", [hex: :myxql, repo: "hexpm", optional: true]}, {:postgrex, "~> 0.19 or ~> 1.0", [hex: :postgrex, repo: "hexpm", optional: true]}, {:tds, "~> 2.1.1 or ~> 2.2", [hex: :tds, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4.0 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "f4d8d36faf294c9417b5a37ec7ac8217ee2abdef5fcf197ba690f361548d3949"}, + "elixir_uuid": {:hex, :elixir_uuid, "1.2.1", "dce506597acb7e6b0daeaff52ff6a9043f5919a4c3315abb4143f0b00378c097", [:mix], [], "hexpm", "f7eba2ea6c3555cea09706492716b0d87397b88946e6380898c2889d68585752"}, + "erlex": {:hex, :erlex, "0.2.9", "7debbbaa9f4f368b8cd648983e0f1d7963028508e9c59e9d4ed504e94ef52a55", [:mix], [], "hexpm", "8cfffc0ec7159e6d73de2ab28a588064de80f88b2798d5cbe4482cbbc200178b"}, + "exprotobuf": {:hex, :exprotobuf, "1.2.17", "3003937da617f588a8fb63ebdd7b127a18d78d6502623c272076fd54c07c4de1", [:mix], [{:gpb, "~> 4.0", [hex: :gpb, repo: "hexpm", optional: false]}], "hexpm", "e07ec1e5ae6f8c1c8521450d5f6b658c8c700b1f34c70356e91ece0766f4361a"}, + "extreme": {:hex, :extreme, "1.1.4", "4a69821922dcfd083aabcd34bdacd7b93f5a1bc119f666454d55a28dc6467654", [:mix], [{:elixir_uuid, "~> 1.2", [hex: :elixir_uuid, repo: "hexpm", optional: false]}, {:exprotobuf, "~> 1.2.9", [hex: :exprotobuf, repo: "hexpm", optional: false]}, {:jason, "~> 1.1", [hex: :jason, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "88d1ac4c46a7cf65fc64cf9695d44a7f8b8b426f710d0bbdc65723c28d72e495"}, + "gpb": {:hex, :gpb, "4.21.7", "34320bee1d98811f731f1dbc815b5fc13d8d33f83f95c43f77dbfb532091fff8", [:make, :rebar3], [], "hexpm", "9b3c3adfea1a631fd1da6b2b08ec8f59b412982393950e0fd1b306def79a676a"}, + "jason": {:hex, :jason, "1.4.5", "2e3a008590b0b8d7388c20293e9dcc9cf3e5d642fd2a114e4cbbb52e595d940a", [:mix], [{:decimal, "~> 1.0 or ~> 2.0 or ~> 3.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "b0c823996102bcd0239b3c2444eb00409b72f6a140c1950bc8b457d836b30684"}, + "telemetry": {:hex, :telemetry, "1.4.2", "a0cb522801dffb1c49fe6e30561badffc7b6d0e180db1300df759faa22062855", [:rebar3], [], "hexpm", "928f6495066506077862c0d1646609eed891a4326bee3126ba54b60af61febb1"}, +} diff --git a/examples/bank/apps/core/run.sh b/examples/bank/apps/core/run.sh new file mode 100755 index 0000000..4971ea3 --- /dev/null +++ b/examples/bank/apps/core/run.sh @@ -0,0 +1,9 @@ +#!/bin/bash + +if [ "$1" ] ; then + num="$1" +else + num=1 +fi + +iex --dot-iex ../.iex.exs --sname banking_core_$num -S mix diff --git a/examples/bank/apps/event_store/.formatter.exs b/examples/bank/apps/event_store/.formatter.exs new file mode 100644 index 0000000..88715a2 --- /dev/null +++ b/examples/bank/apps/event_store/.formatter.exs @@ -0,0 +1,4 @@ +[ + import_deps: [:ecto, :x3m_system], + inputs: ["{config,lib,test}/**/*.{ex,exs}"] +] diff --git a/examples/bank/apps/event_store/lib/aggregate_repo.ex b/examples/bank/apps/event_store/lib/aggregate_repo.ex new file mode 100644 index 0000000..70f3ff1 --- /dev/null +++ b/examples/bank/apps/event_store/lib/aggregate_repo.ex @@ -0,0 +1,219 @@ +defmodule Banking.EventStore.AggregateRepo do + @moduledoc !""" + Macro providing EventStore operations for aggregate repositories. + + Consumer apps use this macro with their Extreme connection module: + + use Banking.EventStore.AggregateRepo, extreme: Banking.Core.EventStore + """ + + defmodule Query do + @moduledoc !""" + Builds Extreme protocol buffer messages for EventStore operations. + """ + alias Extreme.Messages, as: ExMsg + alias X3m.System.Message, as: SysMsg + + @spec write_events(stream :: String.t(), message :: SysMsg.t(), write_metadata :: map()) :: + ExMsg.WriteEvents.t() + def write_events(stream, %SysMsg{} = message, write_metadata) do + emitted_at = + DateTime.utc_now() + |> DateTime.to_iso8601() + + metadata = fn -> + %{ + "$correlationId": message.correlation_id, + "$causationId": message.id, + emitted_at: emitted_at + } + |> Map.merge(write_metadata) + end + + proto_events = + Enum.map(message.events, fn event -> + ExMsg.NewEvent.new( + event_id: Extreme.Tools.generate_uuid(), + event_type: to_string(event.__struct__), + data_content_type: 1, + metadata_content_type: 1, + data: Jason.encode!(Map.from_struct(event)), + metadata: Jason.encode!(metadata.()) + ) + end) + + ExMsg.WriteEvents.new( + event_stream_id: stream, + expected_version: message.aggregate_meta.version, + events: proto_events, + require_master: false + ) + end + + @spec read_events( + stream :: String.t(), + start_at :: non_neg_integer(), + per_page :: pos_integer() + ) :: + ExMsg.ReadStreamEvents.t() + def read_events(stream, start_at, per_page) do + ExMsg.ReadStreamEvents.new( + event_stream_id: stream, + from_event_number: start_at, + max_count: per_page, + resolve_link_tos: true, + require_master: false + ) + end + + @spec delete_stream( + stream :: String.t(), + hard_delete? :: boolean(), + expected_version :: integer() + ) :: + ExMsg.DeleteStream.t() + def delete_stream(stream, hard_delete?, expected_version) do + ExMsg.DeleteStream.new( + event_stream_id: stream, + expected_version: expected_version, + require_master: false, + hard_delete: hard_delete? + ) + end + end + + defmacro __using__(opts) do + quote do + use X3m.System.Aggregate.Repo + require Logger + alias Banking.EventStore.AggregateRepo.Query + + @extreme Keyword.fetch!(unquote(opts), :extreme) + + @impl X3m.System.Aggregate.Repo + @spec has?(stream_name :: String.t()) :: boolean() + def has?(stream) do + stream + |> Query.read_events(0, 1) + |> @extreme.execute() + |> case do + {:ok, _} -> true + _ -> false + end + end + + @impl X3m.System.Aggregate.Repo + @spec delete_stream( + stream_name :: String.t(), + hard_delete? :: boolean(), + expected_version :: integer() + ) :: :ok + def delete_stream(stream, hard_delete?, expected_version) do + stream + |> Query.delete_stream(hard_delete?, expected_version) + |> @extreme.execute() + |> case do + {:ok, _} -> :ok + {:error, :NoStream, _} -> :ok + end + end + + @impl X3m.System.Aggregate.Repo + @spec save_events( + stream_name :: String.t(), + message :: X3m.System.Message.t(), + events_metadata :: map() + ) :: + {:ok, last_event_number :: integer()} + | {:error, :wrong_expected_version, expected_version :: integer()} + | {:error, reason :: any()} + def save_events(_stream, %X3m.System.Message{events: []} = message, _), + do: {:ok, message.aggregate_meta.version} + + def save_events(stream, %X3m.System.Message{} = message, events_metadata) do + stream + |> Query.write_events(message, events_metadata) + |> @extreme.execute() + |> case do + {:ok, result} -> + {:ok, result.last_event_number} + + {:error, :WrongExpectedVersion, _} -> + {:error, :wrong_expected_version, message.aggregate_meta.version} + + other -> + {:error, other} + end + end + + @impl X3m.System.Aggregate.Repo + def stream_events(stream, start_at \\ 0, per_page \\ 500) do + Stream.resource( + fn -> _fetch_stream_events({stream, start_at, per_page, false}) end, + &_return_stream_events/1, + fn x -> x end + ) + end + + defp _fetch_stream_events({stream, start_at, per_page, _is_completed}) do + Logger.debug(fn -> + "Taking #{per_page} items starting from #{start_at} for stream: #{inspect(stream)}" + end) + + {events, is_end_of_stream} = + stream + |> Query.read_events(start_at, per_page) + |> @extreme.execute() + |> case do + {:ok, response} -> + events = + Enum.map(response.events, fn e -> + event_payload = Jason.decode!(e.event.data, keys: :atoms) + event_metadata = Jason.decode!(e.event.metadata) + + event = + e.event.event_type + |> String.to_atom() + |> _create_event(event_payload) + + {event, e.event.event_number, event_metadata} + end) + + {events, response.is_end_of_stream} + + {:error, :NoStream, _} -> + {[], true} + + {:error, :no_stream, _} -> + {[], true} + end + + {events, {stream, start_at + per_page, per_page, is_end_of_stream}} + end + + defp _return_stream_events({[], {_, _, _, is_completed} = params}) when is_completed, + do: {:halt, params} + + defp _return_stream_events({[], params}) do + {result, next} = _fetch_stream_events(params) + {result, {[], next}} + end + + defp _return_stream_events({events, params}), + do: {events, {[], params}} + + defp _create_event(event_type, event_payload) when is_atom(event_type) do + try do + struct(event_type, event_payload) + rescue + UndefinedFunctionError -> + Logger.warning( + "Event struct type #{inspect(event_type)} missing, returning payload as tuple" + ) + + {event_type, event_payload} + end + end + end + end +end diff --git a/examples/bank/apps/event_store/lib/db.ex b/examples/bank/apps/event_store/lib/db.ex new file mode 100644 index 0000000..1e1f3d5 --- /dev/null +++ b/examples/bank/apps/event_store/lib/db.ex @@ -0,0 +1,53 @@ +defmodule Banking.EventStore.DB do + @moduledoc !""" + Catch-up tracking for event listeners. + Stores the last acknowledged event number per listener module and stream. + """ + import Ecto.Query + alias Banking.EventStore.EventsCatchup + + @spec in_transaction(repo :: module(), fun :: (-> term())) :: :ok | X3m.System.Message.t() + def in_transaction(repo, fun) do + repo.transaction(fun) + |> case do + {:ok, %X3m.System.Message{} = msg} -> msg + {:ok, _} -> :ok + end + end + + @spec ack_event( + repo :: module(), + module_name :: String.t(), + stream :: String.t(), + event_number :: integer() + ) :: :ok + def ack_event(repo, module_name, stream, event_number) do + query = + from(e in EventsCatchup, + where: e.module_name == ^module_name and e.stream == ^stream + ) + + {1, _} = repo.update_all(query, set: [ver: event_number]) + :ok + end + + @spec get_last_event(repo :: module(), module_name :: String.t(), stream :: String.t()) :: + integer() + def get_last_event(repo, module_name, stream) do + query = + from(e in EventsCatchup, + where: e.module_name == ^module_name and e.stream == ^stream + ) + + repo + |> apply(:one, [query]) + |> case do + nil -> + repo.insert(%EventsCatchup{module_name: module_name, stream: stream, ver: -1}) + -1 + + rec -> + rec.ver + end + end +end diff --git a/examples/bank/apps/event_store/lib/events_catchup.ex b/examples/bank/apps/event_store/lib/events_catchup.ex new file mode 100644 index 0000000..1fd1401 --- /dev/null +++ b/examples/bank/apps/event_store/lib/events_catchup.ex @@ -0,0 +1,18 @@ +defmodule Banking.EventStore.EventsCatchup do + @moduledoc !""" + Ecto schema for tracking listener catch-up position per stream. + """ + use Ecto.Schema + + @type t() :: %__MODULE__{ + module_name: String.t(), + stream: String.t(), + ver: integer() + } + + schema "events_catchup" do + field :module_name, :string + field :stream, :string + field :ver, :integer + end +end diff --git a/examples/bank/apps/event_store/lib/listener.ex b/examples/bank/apps/event_store/lib/listener.ex new file mode 100644 index 0000000..2976c25 --- /dev/null +++ b/examples/bank/apps/event_store/lib/listener.ex @@ -0,0 +1,166 @@ +defmodule Banking.EventStore.Listener do + @moduledoc !""" + Macro for defining event listeners that subscribe to EventStore streams. + + Provides the `on_event/2` macro to map event types to dispatch service names: + + use Banking.EventStore.Listener, + stream: "$ce-accounts", + db_repo: Banking.Listeners.Repo, + module_name: "Banking.Listeners.Accounts.Listener" + + on_event Banking.Core.Aggregates.Account.Events.Opened, dispatch: :account_opened! + """ + + defmacro on_event(event_type, opts \\ []) do + service_name = Keyword.fetch!(opts, :dispatch) + + quote do + def process_event(unquote(event_type), %X3m.System.Message{} = msg) do + msg + |> X3m.System.Message.to_service(unquote(service_name)) + |> Map.put(:logger_metadata, Logger.metadata()) + |> X3m.System.Dispatcher.dispatch() + |> case do + %X3m.System.Message{response: :ok} = msg -> + msg + + %X3m.System.Message{response: {:ok, ver}} = msg when is_integer(ver) -> + msg + + %X3m.System.Message{response: {:created, id}} = msg when is_binary(id) -> + msg + + %X3m.System.Message{response: {:created, id, ver}} = msg + when is_binary(id) and is_integer(ver) -> + msg + end + end + end + end + + defmacro __using__(opts) do + db_repo = Keyword.fetch!(opts, :db_repo) + stream = Keyword.fetch!(opts, :stream) + + quote location: :keep do + use Extreme.Listener + require Logger + require Banking.EventStore.Listener + import Banking.EventStore.Listener + alias X3m.System.Message, as: SysMsg + + @module_name Keyword.get(unquote(opts), :module_name, __MODULE__) |> to_string() + + def child_spec([extreme | opts]) do + %{ + id: __MODULE__, + start: {__MODULE__, :start_link, [extreme, unquote(stream), opts]} + } + end + + @spec get_last_event(stream_name :: String.t()) :: integer() + defp get_last_event(stream_name) do + unquote(db_repo) + |> Process.whereis() + |> case do + nil -> + Process.sleep(100) + get_last_event(stream_name) + + _pid -> + Banking.EventStore.DB.get_last_event( + unquote(db_repo), + @module_name, + stream_name + ) + end + end + + @spec process_push(push :: map(), stream_name :: String.t()) :: + {:ok, event_number :: integer()} + defp process_push( + %{event: %{data: nil}, link: %{event_number: link_event_number}}, + stream_name + ) do + :ok = + Banking.EventStore.DB.ack_event( + unquote(db_repo), + @module_name, + stream_name, + link_event_number + ) + + {:ok, link_event_number} + end + + defp process_push(push, stream_name) do + link_event_number = push.link.event_number + event_number = push.event.event_number + event_type = String.to_atom(push.event.event_type) + + data = Jason.decode!(push.event.data, keys: :atoms) + metadata = _get_metadata(push.event.metadata) + correlation_id = Map.get(metadata, :"$correlationId") + causation_id = Map.get(metadata, :"$id") + emitted_at = Map.get(metadata, :emitted_at) + + logger_metadata = + Logger.metadata() + |> Keyword.put(:corr_id, correlation_id) + |> Keyword.put(:caus_id, causation_id) + + Logger.metadata(logger_metadata) + + msg = + :on_event + |> SysMsg.new( + raw_request: data, + correlation_id: correlation_id, + causation_id: causation_id, + logger_metadata: logger_metadata + ) + |> SysMsg.assign(:on_event, event_type) + |> SysMsg.assign(:event_number, event_number) + |> SysMsg.assign(:link_event_number, link_event_number) + |> SysMsg.assign(:emitted_at, emitted_at) + |> SysMsg.assign(:event_data, data) + |> SysMsg.assign(:invoked_by, %{system?: true}) + + Banking.EventStore.DB.in_transaction(unquote(db_repo), fn -> + %SysMsg{} = process_event(event_type, msg) + + :ok = + Banking.EventStore.DB.ack_event( + unquote(db_repo), + @module_name, + stream_name, + link_event_number + ) + + Logger.debug(fn -> "#{__MODULE__} processed event ##{event_number}" end) + end) + + {:ok, link_event_number} + end + + defp _get_metadata(nil), do: %{} + + defp _get_metadata(metadata), + do: Jason.decode!(metadata, keys: :atoms) + + @before_compile Banking.EventStore.Listener + end + end + + defmacro __before_compile__(_env) do + quote do + @spec process_event(event_type :: atom(), msg :: X3m.System.Message.t()) :: + X3m.System.Message.t() + def process_event(event_type, msg) do + Logger.debug(fn -> "#{__MODULE__} skipping event #{event_type}" end) + X3m.System.Message.return(msg, :ok) + end + end + end +end diff --git a/examples/bank/apps/event_store/mix.exs b/examples/bank/apps/event_store/mix.exs new file mode 100644 index 0000000..9e04156 --- /dev/null +++ b/examples/bank/apps/event_store/mix.exs @@ -0,0 +1,32 @@ +defmodule Banking.EventStore.MixProject do + use Mix.Project + + def project do + [ + app: :banking_event_store, + version: "0.1.0", + elixir: "~> 1.17", + start_permanent: false, + deps: _deps(), + elixirc_paths: ["lib"] + ] + end + + def application do + [extra_applications: [:logger]] + end + + defp _deps do + [ + {:jason, "~> 1.4"}, + {:extreme, "~> 1.0"}, + {:x3m_system, path: "../../../.."}, + {:postgrex, "~> 0.19", optional: true}, + {:ecto, "~> 3.12", optional: true}, + {:ecto_sql, "~> 3.12", optional: true}, + + # local + {:banking, path: "../banking"} + ] + end +end diff --git a/examples/bank/apps/event_store/mix.lock b/examples/bank/apps/event_store/mix.lock new file mode 100644 index 0000000..a03b8ab --- /dev/null +++ b/examples/bank/apps/event_store/mix.lock @@ -0,0 +1,13 @@ +%{ + "db_connection": {:hex, :db_connection, "2.10.1", "d5465f6bcc125c1b8981c1dbf23c193ca16f446ec0b25832dc174f74f18be510", [:mix], [{:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "18ed94c6e627b4bf452dbd4df61b69a35a1e768525140bc1917b7a685026a6a3"}, + "decimal": {:hex, :decimal, "3.1.1", "430d87b04011ce6cbd4fd205be758311a81f87d552d40904abd00f015935b1d0", [:mix], [], "hexpm", "c5f25f2ced74a0587d03e6023f595db8e924c9d3922c8c8ffd9edfc4498cf1f6"}, + "ecto": {:hex, :ecto, "3.14.0", "2fa64521eebfcb2670d907a86e4ad947290e9933706bb315e6fb5c21b172cb26", [:mix], [{:decimal, "~> 3.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "130d69ffb4285f9ce4792b65dfbb994fd13ea4cbc3cbea2524b199aa3de84af3"}, + "ecto_sql": {:hex, :ecto_sql, "3.14.0", "06446ab8410d2f85bfbb80857ee224ab3b693700cbb38f6535d507449a627b2e", [:mix], [{:db_connection, "~> 2.9", [hex: :db_connection, repo: "hexpm", optional: false]}, {:decimal, "~> 3.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:ecto, "~> 3.14.0", [hex: :ecto, repo: "hexpm", optional: false]}, {:myxql, "~> 0.8", [hex: :myxql, repo: "hexpm", optional: true]}, {:postgrex, "~> 0.19 or ~> 1.0", [hex: :postgrex, repo: "hexpm", optional: true]}, {:tds, "~> 2.1.1 or ~> 2.2", [hex: :tds, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4.0 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "f4d8d36faf294c9417b5a37ec7ac8217ee2abdef5fcf197ba690f361548d3949"}, + "elixir_uuid": {:hex, :elixir_uuid, "1.2.1", "dce506597acb7e6b0daeaff52ff6a9043f5919a4c3315abb4143f0b00378c097", [:mix], [], "hexpm", "f7eba2ea6c3555cea09706492716b0d87397b88946e6380898c2889d68585752"}, + "exprotobuf": {:hex, :exprotobuf, "1.2.17", "3003937da617f588a8fb63ebdd7b127a18d78d6502623c272076fd54c07c4de1", [:mix], [{:gpb, "~> 4.0", [hex: :gpb, repo: "hexpm", optional: false]}], "hexpm", "e07ec1e5ae6f8c1c8521450d5f6b658c8c700b1f34c70356e91ece0766f4361a"}, + "extreme": {:hex, :extreme, "1.1.4", "4a69821922dcfd083aabcd34bdacd7b93f5a1bc119f666454d55a28dc6467654", [:mix], [{:elixir_uuid, "~> 1.2", [hex: :elixir_uuid, repo: "hexpm", optional: false]}, {:exprotobuf, "~> 1.2.9", [hex: :exprotobuf, repo: "hexpm", optional: false]}, {:jason, "~> 1.1", [hex: :jason, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "88d1ac4c46a7cf65fc64cf9695d44a7f8b8b426f710d0bbdc65723c28d72e495"}, + "gpb": {:hex, :gpb, "4.21.7", "34320bee1d98811f731f1dbc815b5fc13d8d33f83f95c43f77dbfb532091fff8", [:make, :rebar3], [], "hexpm", "9b3c3adfea1a631fd1da6b2b08ec8f59b412982393950e0fd1b306def79a676a"}, + "jason": {:hex, :jason, "1.4.5", "2e3a008590b0b8d7388c20293e9dcc9cf3e5d642fd2a114e4cbbb52e595d940a", [:mix], [{:decimal, "~> 1.0 or ~> 2.0 or ~> 3.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "b0c823996102bcd0239b3c2444eb00409b72f6a140c1950bc8b457d836b30684"}, + "postgrex": {:hex, :postgrex, "0.22.2", "4aec14df2a72722aee92492566edbeeb44e233ecb86b1915d03136297ef1385d", [:mix], [{:db_connection, "~> 2.9", [hex: :db_connection, repo: "hexpm", optional: false]}, {:decimal, "~> 1.5 or ~> 2.0 or ~> 3.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:table, "~> 0.1.0", [hex: :table, repo: "hexpm", optional: true]}], "hexpm", "8946382ddb06294f56026ac4278b3cc212bac8a2c82ed68b4087819ed1abc53b"}, + "telemetry": {:hex, :telemetry, "1.4.2", "a0cb522801dffb1c49fe6e30561badffc7b6d0e180db1300df759faa22062855", [:rebar3], [], "hexpm", "928f6495066506077862c0d1646609eed891a4326bee3126ba54b60af61febb1"}, +} diff --git a/examples/bank/apps/listeners/.formatter.exs b/examples/bank/apps/listeners/.formatter.exs new file mode 100644 index 0000000..4d4cce1 --- /dev/null +++ b/examples/bank/apps/listeners/.formatter.exs @@ -0,0 +1,4 @@ +[ + import_deps: [:ecto, :ecto_sql, :x3m_system], + inputs: ["{config,lib,test}/**/*.{ex,exs}", "priv/repo/migrations/**/*.{ex,exs}"] +] diff --git a/examples/bank/apps/listeners/config/config.exs b/examples/bank/apps/listeners/config/config.exs new file mode 100644 index 0000000..8f57c60 --- /dev/null +++ b/examples/bank/apps/listeners/config/config.exs @@ -0,0 +1,23 @@ +import Config + +import_config "../../banking/config/config.exs" + +config :banking_listeners, ecto_repos: [Banking.Listeners.Repo] + +config :banking_listeners, Banking.Listeners.EventStore, + db_type: "node", + host: "localhost", + port: "1113", + username: "admin", + password: "changeit", + connection_name: "banking_listeners" + +config :banking_listeners, Banking.Listeners.Repo, + adapter: Ecto.Adapters.Postgres, + username: "postgres", + password: "postgres", + database: "banking", + hostname: "localhost", + pool_size: 5 + +import_config "#{Mix.env()}.exs" diff --git a/examples/bank/apps/listeners/config/dev.exs b/examples/bank/apps/listeners/config/dev.exs new file mode 100644 index 0000000..becde76 --- /dev/null +++ b/examples/bank/apps/listeners/config/dev.exs @@ -0,0 +1 @@ +import Config diff --git a/examples/bank/apps/listeners/config/test.exs b/examples/bank/apps/listeners/config/test.exs new file mode 100644 index 0000000..3664f7c --- /dev/null +++ b/examples/bank/apps/listeners/config/test.exs @@ -0,0 +1,3 @@ +import Config + +config :banking_listeners, Banking.Listeners.Repo, pool: Ecto.Adapters.SQL.Sandbox diff --git a/examples/bank/apps/listeners/lib/accounts/denormalizer.ex b/examples/bank/apps/listeners/lib/accounts/denormalizer.ex new file mode 100644 index 0000000..2155234 --- /dev/null +++ b/examples/bank/apps/listeners/lib/accounts/denormalizer.ex @@ -0,0 +1,59 @@ +defmodule Banking.Listeners.Accounts.Denormalizer do + @moduledoc !""" + Upserts account events into the read model. + """ + import Ecto.Query + alias Banking.Listeners.Repo + alias Banking.Listeners.Accounts.Model + alias X3m.System.Message, as: SysMsg + alias Banking.Core.Aggregates.Account.Events + + @spec denormalize(msg :: SysMsg.t()) :: {:reply, SysMsg.t()} + def denormalize( + %SysMsg{ + assigns: %{on_event: Events.Opened}, + raw_request: %{id: id, owner_id: owner_id} + } = msg + ) do + %Model{id: id, owner_id: owner_id, balance: 0, status: "open"} + |> Repo.insert!(on_conflict: :nothing) + + {:reply, SysMsg.ok(msg)} + end + + def denormalize( + %SysMsg{ + assigns: %{on_event: Events.Deposited}, + raw_request: %{id: id, amount: amount} + } = msg + ) do + from(a in Model, where: a.id == ^id) + |> Repo.update_all(inc: [balance: amount]) + + {:reply, SysMsg.ok(msg)} + end + + def denormalize( + %SysMsg{ + assigns: %{on_event: Events.Withdrawn}, + raw_request: %{id: id, amount: amount} + } = msg + ) do + from(a in Model, where: a.id == ^id) + |> Repo.update_all(inc: [balance: -amount]) + + {:reply, SysMsg.ok(msg)} + end + + def denormalize( + %SysMsg{ + assigns: %{on_event: Events.Closed}, + raw_request: %{id: id} + } = msg + ) do + from(a in Model, where: a.id == ^id) + |> Repo.update_all(set: [status: "closed"]) + + {:reply, SysMsg.ok(msg)} + end +end diff --git a/examples/bank/apps/listeners/lib/accounts/listener.ex b/examples/bank/apps/listeners/lib/accounts/listener.ex new file mode 100644 index 0000000..acef98f --- /dev/null +++ b/examples/bank/apps/listeners/lib/accounts/listener.ex @@ -0,0 +1,16 @@ +defmodule Banking.Listeners.Accounts.Listener do + @moduledoc !""" + Subscribes to the accounts category stream and dispatches events to denormalizers. + """ + use Banking.EventStore.Listener, + stream: "$ce-accounts", + db_repo: Banking.Listeners.Repo, + module_name: "Banking.Listeners.Accounts.Listener" + + alias Banking.Core.Aggregates.Account.Events + + on_event(Events.Opened, dispatch: :account_opened!) + on_event(Events.Deposited, dispatch: :account_deposited!) + on_event(Events.Withdrawn, dispatch: :account_withdrawn!) + on_event(Events.Closed, dispatch: :account_closed!) +end diff --git a/examples/bank/apps/listeners/lib/accounts/model.ex b/examples/bank/apps/listeners/lib/accounts/model.ex new file mode 100644 index 0000000..569f58d --- /dev/null +++ b/examples/bank/apps/listeners/lib/accounts/model.ex @@ -0,0 +1,16 @@ +defmodule Banking.Listeners.Accounts.Model do + @moduledoc !""" + Ecto schema for the accounts read model table. + """ + use Ecto.Schema + + @primary_key {:id, :string, autogenerate: false} + + schema "accounts" do + field :owner_id, :string + field :balance, :integer, default: 0 + field :status, :string, default: "open" + + timestamps() + end +end diff --git a/examples/bank/apps/listeners/lib/application.ex b/examples/bank/apps/listeners/lib/application.ex new file mode 100644 index 0000000..565ff6f --- /dev/null +++ b/examples/bank/apps/listeners/lib/application.ex @@ -0,0 +1,36 @@ +defmodule Banking.Listeners.Application do + @moduledoc false + use Application + + def start(_type, _args) do + Banking.TelemetryLogger.setup() + Banking.Migrations.create(:banking_listeners) + Banking.Migrations.up(:banking_listeners) + + :ok = Banking.Listeners.Router.register_services() + + children = _get_children() + + opts = [strategy: :one_for_one, name: Banking.Listeners.MainSupervisor] + Supervisor.start_link(children, opts) + end + + case Mix.env() do + :test -> + defp _get_children do + [Banking.Listeners.Repo] + end + + _ -> + defp _get_children do + event_store_config = + Application.get_env(:banking_listeners, Banking.Listeners.EventStore) + + [ + Banking.Listeners.Repo, + {Banking.Listeners.EventStore, event_store_config}, + {Banking.Listeners.Accounts.Listener, [Banking.Listeners.EventStore]} + ] + end + end +end diff --git a/examples/bank/apps/listeners/lib/event_store.ex b/examples/bank/apps/listeners/lib/event_store.ex new file mode 100644 index 0000000..8ebdad2 --- /dev/null +++ b/examples/bank/apps/listeners/lib/event_store.ex @@ -0,0 +1,6 @@ +defmodule Banking.Listeners.EventStore do + @moduledoc !""" + Extreme TCP connection to EventStore for the listeners app. + """ + use Extreme, otp_app: :banking_listeners +end diff --git a/examples/bank/apps/listeners/lib/repo.ex b/examples/bank/apps/listeners/lib/repo.ex new file mode 100644 index 0000000..0414fff --- /dev/null +++ b/examples/bank/apps/listeners/lib/repo.ex @@ -0,0 +1,8 @@ +defmodule Banking.Listeners.Repo do + @moduledoc !""" + Ecto repository for the denormalized read model. + """ + use Ecto.Repo, + otp_app: :banking_listeners, + adapter: Ecto.Adapters.Postgres +end diff --git a/examples/bank/apps/listeners/lib/router.ex b/examples/bank/apps/listeners/lib/router.ex new file mode 100644 index 0000000..9e591c3 --- /dev/null +++ b/examples/bank/apps/listeners/lib/router.ex @@ -0,0 +1,26 @@ +defmodule Banking.Listeners.Router do + @moduledoc !""" + Registers denormalization and read services. + """ + use X3m.System.Router + alias Banking.Listeners.Accounts.Denormalizer + alias Banking.Listeners.Services + + # Read services (dispatched by API) + service :get_account, Services.GetAccount, :handle + service :list_accounts, Services.ListAccounts, :handle + + # Denormalization services (dispatched by listener) + servicep :account_opened!, Denormalizer, :denormalize + servicep :account_deposited!, Denormalizer, :denormalize + servicep :account_withdrawn!, Denormalizer, :denormalize + servicep :account_closed!, Denormalizer, :denormalize + + def authorize(%X3m.System.Message{service_name: :list_accounts, assigns: assigns}) do + if assigns[:invoked_by] && assigns.invoked_by.admin?, + do: :ok, + else: :forbidden + end + + def authorize(%X3m.System.Message{}), do: :ok +end diff --git a/examples/bank/apps/listeners/lib/services/get_account.ex b/examples/bank/apps/listeners/lib/services/get_account.ex new file mode 100644 index 0000000..9605eb3 --- /dev/null +++ b/examples/bank/apps/listeners/lib/services/get_account.ex @@ -0,0 +1,29 @@ +defmodule Banking.Listeners.Services.GetAccount do + @moduledoc !""" + Query service — returns a single account by ID from the read model. + """ + alias Banking.Listeners.Repo + alias Banking.Listeners.Accounts.Model + alias X3m.System.Message, as: SysMsg + + @spec handle(msg :: SysMsg.t()) :: {:reply, SysMsg.t()} + def handle(%SysMsg{raw_request: %{"id" => id}} = msg) do + Repo.get(Model, id) + |> case do + nil -> + {:reply, SysMsg.error(msg, :not_found)} + + %Model{} = account -> + {:reply, SysMsg.return(msg, {:ok, _to_map(account)})} + end + end + + defp _to_map(%Model{} = account) do + %{ + id: account.id, + owner_id: account.owner_id, + balance: account.balance, + status: account.status + } + end +end diff --git a/examples/bank/apps/listeners/lib/services/list_accounts.ex b/examples/bank/apps/listeners/lib/services/list_accounts.ex new file mode 100644 index 0000000..cefc04a --- /dev/null +++ b/examples/bank/apps/listeners/lib/services/list_accounts.ex @@ -0,0 +1,27 @@ +defmodule Banking.Listeners.Services.ListAccounts do + @moduledoc !""" + Query service — returns all accounts from the read model. + """ + alias Banking.Listeners.Repo + alias Banking.Listeners.Accounts.Model + alias X3m.System.Message, as: SysMsg + + @spec handle(msg :: SysMsg.t()) :: {:reply, SysMsg.t()} + def handle(%SysMsg{} = msg) do + accounts = + Model + |> Repo.all() + |> Enum.map(&_to_map/1) + + {:reply, SysMsg.return(msg, {:ok, accounts})} + end + + defp _to_map(%Model{} = account) do + %{ + id: account.id, + owner_id: account.owner_id, + balance: account.balance, + status: account.status + } + end +end diff --git a/examples/bank/apps/listeners/mix.exs b/examples/bank/apps/listeners/mix.exs new file mode 100644 index 0000000..314144c --- /dev/null +++ b/examples/bank/apps/listeners/mix.exs @@ -0,0 +1,45 @@ +defmodule Banking.Listeners.MixProject do + use Mix.Project + + def project do + [ + app: :banking_listeners, + version: "0.1.0", + elixir: "~> 1.17", + start_permanent: Mix.env() == :prod, + deps: _deps(), + elixirc_paths: _elixirc_paths(Mix.env()), + dialyzer: [ + plt_add_deps: :apps_direct, + plt_add_apps: [:ecto, :extreme, :jason, :x3m_system] + ] + ] + end + + def cli do + [preferred_envs: [bless: :test]] + end + + def application do + [ + extra_applications: [:logger], + mod: {Banking.Listeners.Application, []} + ] + end + + defp _elixirc_paths(:test), do: ["lib", "test/support"] + defp _elixirc_paths(_), do: ["lib"] + + defp _deps do + [ + {:postgrex, "~> 0.19"}, + {:ecto_sql, "~> 3.12"}, + {:x3m_system, path: "../../../.."}, + {:dialyxir, "~> 1.0", only: :dev, runtime: false}, + + # local + {:banking_event_store, path: "../event_store"}, + {:banking, path: "../banking"} + ] + end +end diff --git a/examples/bank/apps/listeners/mix.lock b/examples/bank/apps/listeners/mix.lock new file mode 100644 index 0000000..eda7914 --- /dev/null +++ b/examples/bank/apps/listeners/mix.lock @@ -0,0 +1,15 @@ +%{ + "db_connection": {:hex, :db_connection, "2.10.1", "d5465f6bcc125c1b8981c1dbf23c193ca16f446ec0b25832dc174f74f18be510", [:mix], [{:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "18ed94c6e627b4bf452dbd4df61b69a35a1e768525140bc1917b7a685026a6a3"}, + "decimal": {:hex, :decimal, "3.1.1", "430d87b04011ce6cbd4fd205be758311a81f87d552d40904abd00f015935b1d0", [:mix], [], "hexpm", "c5f25f2ced74a0587d03e6023f595db8e924c9d3922c8c8ffd9edfc4498cf1f6"}, + "dialyxir": {:hex, :dialyxir, "1.4.7", "dda948fcee52962e4b6c5b4b16b2d8fa7d50d8645bbae8b8685c3f9ecb7f5f4d", [:mix], [{:erlex, ">= 0.2.8", [hex: :erlex, repo: "hexpm", optional: false]}], "hexpm", "b34527202e6eb8cee198efec110996c25c5898f43a4094df157f8d28f27d9efe"}, + "ecto": {:hex, :ecto, "3.14.0", "2fa64521eebfcb2670d907a86e4ad947290e9933706bb315e6fb5c21b172cb26", [:mix], [{:decimal, "~> 3.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "130d69ffb4285f9ce4792b65dfbb994fd13ea4cbc3cbea2524b199aa3de84af3"}, + "ecto_sql": {:hex, :ecto_sql, "3.14.0", "06446ab8410d2f85bfbb80857ee224ab3b693700cbb38f6535d507449a627b2e", [:mix], [{:db_connection, "~> 2.9", [hex: :db_connection, repo: "hexpm", optional: false]}, {:decimal, "~> 3.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:ecto, "~> 3.14.0", [hex: :ecto, repo: "hexpm", optional: false]}, {:myxql, "~> 0.8", [hex: :myxql, repo: "hexpm", optional: true]}, {:postgrex, "~> 0.19 or ~> 1.0", [hex: :postgrex, repo: "hexpm", optional: true]}, {:tds, "~> 2.1.1 or ~> 2.2", [hex: :tds, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4.0 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "f4d8d36faf294c9417b5a37ec7ac8217ee2abdef5fcf197ba690f361548d3949"}, + "elixir_uuid": {:hex, :elixir_uuid, "1.2.1", "dce506597acb7e6b0daeaff52ff6a9043f5919a4c3315abb4143f0b00378c097", [:mix], [], "hexpm", "f7eba2ea6c3555cea09706492716b0d87397b88946e6380898c2889d68585752"}, + "erlex": {:hex, :erlex, "0.2.9", "7debbbaa9f4f368b8cd648983e0f1d7963028508e9c59e9d4ed504e94ef52a55", [:mix], [], "hexpm", "8cfffc0ec7159e6d73de2ab28a588064de80f88b2798d5cbe4482cbbc200178b"}, + "exprotobuf": {:hex, :exprotobuf, "1.2.17", "3003937da617f588a8fb63ebdd7b127a18d78d6502623c272076fd54c07c4de1", [:mix], [{:gpb, "~> 4.0", [hex: :gpb, repo: "hexpm", optional: false]}], "hexpm", "e07ec1e5ae6f8c1c8521450d5f6b658c8c700b1f34c70356e91ece0766f4361a"}, + "extreme": {:hex, :extreme, "1.1.4", "4a69821922dcfd083aabcd34bdacd7b93f5a1bc119f666454d55a28dc6467654", [:mix], [{:elixir_uuid, "~> 1.2", [hex: :elixir_uuid, repo: "hexpm", optional: false]}, {:exprotobuf, "~> 1.2.9", [hex: :exprotobuf, repo: "hexpm", optional: false]}, {:jason, "~> 1.1", [hex: :jason, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "88d1ac4c46a7cf65fc64cf9695d44a7f8b8b426f710d0bbdc65723c28d72e495"}, + "gpb": {:hex, :gpb, "4.21.7", "34320bee1d98811f731f1dbc815b5fc13d8d33f83f95c43f77dbfb532091fff8", [:make, :rebar3], [], "hexpm", "9b3c3adfea1a631fd1da6b2b08ec8f59b412982393950e0fd1b306def79a676a"}, + "jason": {:hex, :jason, "1.4.5", "2e3a008590b0b8d7388c20293e9dcc9cf3e5d642fd2a114e4cbbb52e595d940a", [:mix], [{:decimal, "~> 1.0 or ~> 2.0 or ~> 3.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "b0c823996102bcd0239b3c2444eb00409b72f6a140c1950bc8b457d836b30684"}, + "postgrex": {:hex, :postgrex, "0.22.2", "4aec14df2a72722aee92492566edbeeb44e233ecb86b1915d03136297ef1385d", [:mix], [{:db_connection, "~> 2.9", [hex: :db_connection, repo: "hexpm", optional: false]}, {:decimal, "~> 1.5 or ~> 2.0 or ~> 3.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:table, "~> 0.1.0", [hex: :table, repo: "hexpm", optional: true]}], "hexpm", "8946382ddb06294f56026ac4278b3cc212bac8a2c82ed68b4087819ed1abc53b"}, + "telemetry": {:hex, :telemetry, "1.4.2", "a0cb522801dffb1c49fe6e30561badffc7b6d0e180db1300df759faa22062855", [:rebar3], [], "hexpm", "928f6495066506077862c0d1646609eed891a4326bee3126ba54b60af61febb1"}, +} diff --git a/examples/bank/apps/listeners/priv/repo/migrations/20260613000001_create_accounts.exs b/examples/bank/apps/listeners/priv/repo/migrations/20260613000001_create_accounts.exs new file mode 100644 index 0000000..c2ab932 --- /dev/null +++ b/examples/bank/apps/listeners/priv/repo/migrations/20260613000001_create_accounts.exs @@ -0,0 +1,22 @@ +defmodule Banking.Listeners.Repo.Migrations.CreateAccounts do + use Ecto.Migration + + def change do + create table(:accounts, primary_key: false) do + add :id, :string, primary_key: true + add :owner_id, :string, null: false + add :balance, :integer, null: false, default: 0 + add :status, :string, null: false, default: "open" + + timestamps() + end + + create table(:events_catchup) do + add :module_name, :string, null: false + add :stream, :string, null: false + add :ver, :integer, null: false, default: -1 + end + + create unique_index(:events_catchup, [:module_name, :stream]) + end +end diff --git a/examples/bank/apps/listeners/run.sh b/examples/bank/apps/listeners/run.sh new file mode 100755 index 0000000..61c995d --- /dev/null +++ b/examples/bank/apps/listeners/run.sh @@ -0,0 +1,9 @@ +#!/bin/bash + +if [ "$1" ] ; then + num="$1" +else + num=1 +fi + +iex --dot-iex ../.iex.exs --sname banking_listeners_$num -S mix diff --git a/examples/bank/docker-compose.yml b/examples/bank/docker-compose.yml new file mode 100644 index 0000000..ad34cb4 --- /dev/null +++ b/examples/bank/docker-compose.yml @@ -0,0 +1,36 @@ +services: + postgres: + image: postgres:17 + environment: + POSTGRES_USER: postgres + POSTGRES_PASSWORD: postgres + POSTGRES_DB: banking + PGDATA: /var/lib/postgresql/data/pgdata + ports: + - "5432:5432" + volumes: + - ./_volumes/pg_data:/var/lib/postgresql/data + healthcheck: + test: ["CMD-SHELL", "pg_isready -U postgres"] + interval: 5s + timeout: 5s + retries: 5 + + eventstore: + image: eventstore/eventstore:21.10.5-buster-slim + # ARM64 (Mac M-series): + # image: ghcr.io/eventstore/eventstore:21.10.5-alpha-arm64v8 + environment: + - EVENTSTORE_CLUSTER_SIZE=1 + - EVENTSTORE_RUN_PROJECTIONS=All + - EVENTSTORE_START_STANDARD_PROJECTIONS=true + - EVENTSTORE_EXT_TCP_PORT=1113 + - EVENTSTORE_HTTP_PORT=2113 + - EVENTSTORE_INSECURE=true + - EVENTSTORE_ENABLE_EXTERNAL_TCP=true + - EVENTSTORE_ENABLE_ATOM_PUB_OVER_HTTP=true + ports: + - "1113:1113" + - "2113:2113" + volumes: + - ./_volumes/es_data:/var/lib/eventstore