diff --git a/lib/commanded/opentelemetry.ex b/lib/commanded/opentelemetry.ex index 6b31257e..30d4a4ba 100644 --- a/lib/commanded/opentelemetry.ex +++ b/lib/commanded/opentelemetry.ex @@ -40,6 +40,7 @@ defmodule Commanded.OpenTelemetry do alias Commanded.OpenTelemetry.AggregatePopulate alias Commanded.OpenTelemetry.Application, as: OTelApplication alias Commanded.OpenTelemetry.EventHandler + alias Commanded.OpenTelemetry.EventStore @typedoc """ Determines how event handler spans relate to command dispatch spans. @@ -85,6 +86,12 @@ defmodule Commanded.OpenTelemetry do ]}, default: [], doc: "Event handler tracing configuration. Use `:disabled` to disable." + ], + event_store: [ + type: {:in, [:disabled, []]}, + default: :disabled, + doc: + "Event store tracing configuration. Disabled by default. Use `[]` to enable." ] ) @@ -99,7 +106,7 @@ defmodule Commanded.OpenTelemetry do ## Examples - # Default setup (enables all tracing) + # Default setup (enables all tracing except event_store) Commanded.OpenTelemetry.setup() # Disable aggregate tracing @@ -114,6 +121,9 @@ defmodule Commanded.OpenTelemetry do # Disable aggregate populate tracing Commanded.OpenTelemetry.setup(aggregate_populate: :disabled) + # Enable event store tracing (disabled by default) + Commanded.OpenTelemetry.setup(event_store: []) + # Use parent-child relationships for event handlers Commanded.OpenTelemetry.setup(event_handler: [span_relationship: :child]) @@ -142,6 +152,11 @@ defmodule Commanded.OpenTelemetry do config -> EventHandler.setup(config) end + case opts[:event_store] do + :disabled -> :ok + _config -> EventStore.setup() + end + :ok end end diff --git a/lib/commanded/opentelemetry/event_store.ex b/lib/commanded/opentelemetry/event_store.ex new file mode 100644 index 00000000..7d444d51 --- /dev/null +++ b/lib/commanded/opentelemetry/event_store.ex @@ -0,0 +1,194 @@ +defmodule Commanded.OpenTelemetry.EventStore do + @moduledoc false + + alias Commanded.OpenTelemetry.CommandedAttributes + alias Commanded.OpenTelemetry.Helpers + alias OpenTelemetry.SemConv.ErrorAttributes + alias OpenTelemetry.SemConv.Incubating.CodeAttributes + alias OpenTelemetry.SemConv.Incubating.MessagingAttributes + alias OpenTelemetry.Span + + @tracer_id __MODULE__ + + @events ~w( + ack_event + append_to_stream + delete_snapshot + delete_subscription + read_snapshot + record_snapshot + stream_forward + subscribe + subscribe_to + unsubscribe + )a + + def setup do + for event <- @events do + :ok = + :telemetry.attach_many( + {__MODULE__, event}, + [ + [:commanded, :event_store, event, :start], + [:commanded, :event_store, event, :stop], + [:commanded, :event_store, event, :exception] + ], + &__MODULE__.handle_telemetry_event/4, + %{} + ) + end + + :ok + end + + def handle_telemetry_event( + [:commanded, :event_store, action, :start], + _measurements, + meta, + _config + ) do + operation_type = operation_type_for(action) + action_name = to_string(action) + source_uuid = extract_source_uuid(meta) + + attributes = + [ + {MessagingAttributes.messaging_system(), "commanded"}, + {MessagingAttributes.messaging_operation_name(), action_name}, + {CodeAttributes.code_function(), action_name}, + {CommandedAttributes.commanded_application(), meta[:application]} + ] + |> maybe_add_operation_type(operation_type) + |> maybe_add_stream_uuid(meta[:stream_uuid]) + |> maybe_add_expected_version(meta[:expected_version]) + |> maybe_add_subscription_name(meta[:subscription_name]) + |> maybe_add_source_uuid(source_uuid) + |> maybe_add_start_from(meta[:start_from]) + + # Use application as destination to maintain consistency with other Commanded modules + # and provide useful grouping in multi-application deployments. + application_name = Helpers.module_name(meta[:application]) + + span_name = + case application_name do + nil -> action_name + name -> "#{action_name} #{name}" + end + + OpentelemetryTelemetry.start_telemetry_span( + @tracer_id, + span_name, + meta, + %{ + kind: :internal, + attributes: attributes + } + ) + end + + def handle_telemetry_event( + [:commanded, :event_store, _action, :stop], + _measurements, + meta, + _config + ) do + ctx = OpentelemetryTelemetry.set_current_telemetry_span(@tracer_id, meta) + + if error = meta[:error] do + Span.set_attribute( + ctx, + ErrorAttributes.error_type(), + Helpers.to_error_type(error, @tracer_id) + ) + + Span.set_status(ctx, OpenTelemetry.status(:error, Helpers.format_error(error))) + end + + OpentelemetryTelemetry.end_telemetry_span(@tracer_id, meta) + end + + def handle_telemetry_event( + [:commanded, :event_store, _action, :exception], + _measurements, + %{kind: kind, reason: reason, stacktrace: stacktrace} = meta, + _config + ) do + ctx = OpentelemetryTelemetry.set_current_telemetry_span(@tracer_id, meta) + + # TODO: follow up with OTEL team to add exception.kind SemConv attribute + Span.set_attribute(ctx, :"erlang.exception.kind", kind) + + exception = Exception.normalize(kind, reason, stacktrace) + + Span.set_attribute( + ctx, + ErrorAttributes.error_type(), + Helpers.to_error_type(exception, @tracer_id) + ) + + Span.record_exception(ctx, exception, stacktrace) + + Span.set_status( + ctx, + OpenTelemetry.status(:error, Exception.format_banner(kind, reason, stacktrace)) + ) + + OpentelemetryTelemetry.end_telemetry_span(@tracer_id, meta) + end + + defp operation_type_for(:append_to_stream), do: :publish + defp operation_type_for(:stream_forward), do: :receive + defp operation_type_for(:subscribe), do: :receive + defp operation_type_for(:subscribe_to), do: :receive + defp operation_type_for(:ack_event), do: :settle + defp operation_type_for(:delete_snapshot), do: nil + defp operation_type_for(:delete_subscription), do: nil + defp operation_type_for(:read_snapshot), do: :receive + defp operation_type_for(:record_snapshot), do: :publish + defp operation_type_for(:unsubscribe), do: nil + defp operation_type_for(_), do: nil + + defp maybe_add_operation_type(attrs, nil), do: attrs + + defp maybe_add_operation_type(attrs, type), + do: [{MessagingAttributes.messaging_operation_type(), type} | attrs] + + defp maybe_add_stream_uuid(attrs, nil), do: attrs + + defp maybe_add_stream_uuid(attrs, stream_uuid), + do: [{CommandedAttributes.commanded_stream_uuid(), stream_uuid} | attrs] + + defp maybe_add_expected_version(attrs, nil), do: attrs + + defp maybe_add_expected_version(attrs, expected_version), + do: [{CommandedAttributes.commanded_expected_version(), expected_version} | attrs] + + defp maybe_add_subscription_name(attrs, nil), do: attrs + + defp maybe_add_subscription_name(attrs, name) do + [ + {MessagingAttributes.messaging_destination_subscription_name(), name}, + {CommandedAttributes.commanded_subscription_name(), name} + | attrs + ] + end + + defp maybe_add_source_uuid(attrs, nil), do: attrs + + defp maybe_add_source_uuid(attrs, source_uuid), + do: [{CommandedAttributes.commanded_source_uuid(), source_uuid} | attrs] + + defp maybe_add_start_from(attrs, nil), do: attrs + + defp maybe_add_start_from(attrs, start_from), + do: [{CommandedAttributes.commanded_start_from(), to_start_from_attr(start_from)} | attrs] + + # Extract source_uuid from metadata or nested snapshot struct (record_snapshot operation) + defp extract_source_uuid(%{source_uuid: uuid}) when is_binary(uuid), do: uuid + defp extract_source_uuid(%{snapshot: %{source_uuid: uuid}}) when is_binary(uuid), do: uuid + defp extract_source_uuid(_), do: nil + + defp to_start_from_attr(nil), do: nil + defp to_start_from_attr(atom) when is_atom(atom), do: to_string(atom) + defp to_start_from_attr(other), do: other +end diff --git a/test/opentelemetry/aggregate_test.exs b/test/opentelemetry/aggregate_test.exs index 4775a62c..de419a26 100644 --- a/test/opentelemetry/aggregate_test.exs +++ b/test/opentelemetry/aggregate_test.exs @@ -17,6 +17,7 @@ defmodule Commanded.OpenTelemetry.AggregateTest do start_supervised!(DefaultApp) detach_handlers() + detach_event_store_handlers() Aggregate.setup() :ok @@ -619,4 +620,27 @@ defmodule Commanded.OpenTelemetry.AggregateTest do end end end + + defp detach_event_store_handlers do + event_store_events = ~w( + ack_event + append_to_stream + delete_snapshot + delete_subscription + read_snapshot + record_snapshot + stream_forward + subscribe + subscribe_to + unsubscribe + )a + + for event <- event_store_events do + for suffix <- [:start, :stop, :exception] do + for handler <- :telemetry.list_handlers([:commanded, :event_store, event, suffix]) do + :telemetry.detach(handler.id) + end + end + end + end end diff --git a/test/opentelemetry/event_store_test.exs b/test/opentelemetry/event_store_test.exs new file mode 100644 index 00000000..0880d178 --- /dev/null +++ b/test/opentelemetry/event_store_test.exs @@ -0,0 +1,591 @@ +defmodule Commanded.OpenTelemetry.EventStoreTest do + @moduledoc """ + Tests for EventStore OpenTelemetry instrumentation. + + Follows the same patterns as ApplicationTest, AggregateTest, etc. + """ + + use Commanded.OpenTelemetryCase, async: false + + import Commanded.TestSupport.Factory + + alias Commanded.OpenTelemetry.EventStore, as: OTelEventStore + alias Commanded.UUID + + @events ~w( + ack_event + append_to_stream + delete_snapshot + delete_subscription + read_snapshot + record_snapshot + stream_forward + subscribe + subscribe_to + unsubscribe + )a + + setup do + detach_handlers() + OTelEventStore.setup() + + :ok + end + + describe "setup/0" do + test "attaches telemetry handlers for event store operations" do + detach_handlers() + + OTelEventStore.setup() + + for event <- @events do + for suffix <- [:start, :stop, :exception] do + handlers = :telemetry.list_handlers([:commanded, :event_store, event, suffix]) + + assert Enum.any?( + handlers, + &match?(%{id: {OTelEventStore, ^event}}, &1) + ), + "Expected handler for event [:commanded, :event_store, #{event}, #{suffix}]" + end + end + end + + test "calling setup twice raises MatchError (fail fast)" do + detach_handlers() + + :ok = OTelEventStore.setup() + + handlers = :telemetry.list_handlers([:commanded, :event_store, :append_to_stream, :start]) + assert length(handlers) == 1 + + assert_raise MatchError, fn -> + OTelEventStore.setup() + end + end + end + + describe "append_to_stream spans" do + setup do + detach_handlers() + OTelEventStore.setup() + :ok + end + + test "creates span with correct attributes" do + stream_uuid = UUID.uuid4() + + meta = + build_event_store_append_to_stream_metadata( + application: TestApp, + stream_uuid: stream_uuid, + expected_version: 0 + ) + + :telemetry.span([:commanded, :event_store, :append_to_stream], meta, fn -> + {:ok, meta} + end) + + assert_receive {:span, + span( + name: "append_to_stream TestApp", + kind: :internal, + attributes: attributes + )}, + 1000 + + assert :otel_attributes.map(attributes) == %{ + "messaging.system": "commanded", + "messaging.operation.type": :publish, + "messaging.operation.name": "append_to_stream", + "code.function": "append_to_stream", + "commanded.application": TestApp, + "commanded.stream.uuid": stream_uuid, + "commanded.expected_version": 0 + } + end + end + + describe "stream_forward spans" do + setup do + detach_handlers() + OTelEventStore.setup() + :ok + end + + test "creates span with correct attributes" do + stream_uuid = UUID.uuid4() + + meta = + build_event_store_stream_forward_metadata( + application: TestApp, + stream_uuid: stream_uuid, + start_version: 0, + read_batch_size: 1000 + ) + + :telemetry.span([:commanded, :event_store, :stream_forward], meta, fn -> + {:ok, meta} + end) + + assert_receive {:span, + span( + name: "stream_forward TestApp", + kind: :internal, + attributes: attributes + )}, + 1000 + + assert :otel_attributes.map(attributes) == %{ + "messaging.system": "commanded", + "messaging.operation.type": :receive, + "messaging.operation.name": "stream_forward", + "code.function": "stream_forward", + "commanded.application": TestApp, + "commanded.stream.uuid": stream_uuid + } + end + end + + describe "subscribe_to spans" do + setup do + detach_handlers() + OTelEventStore.setup() + :ok + end + + test "creates span with subscription attributes" do + meta = + build_event_store_subscribe_to_metadata( + application: TestApp, + stream_uuid: :all, + subscription_name: "TestSubscription", + subscriber: self(), + start_from: :origin + ) + + :telemetry.span([:commanded, :event_store, :subscribe_to], meta, fn -> + {:ok, meta} + end) + + assert_receive {:span, + span( + name: "subscribe_to TestApp", + kind: :internal, + attributes: attributes + )}, + 1000 + + assert :otel_attributes.map(attributes) == %{ + "messaging.system": "commanded", + "messaging.operation.type": :receive, + "messaging.operation.name": "subscribe_to", + "messaging.destination.subscription.name": "TestSubscription", + "code.function": "subscribe_to", + "commanded.application": TestApp, + "commanded.stream.uuid": :all, + "commanded.subscription.name": "TestSubscription", + "commanded.start_from": "origin" + } + end + end + + describe "ack_event spans" do + setup do + detach_handlers() + OTelEventStore.setup() + :ok + end + + test "creates span with settle operation type" do + meta = + build_event_store_ack_event_metadata( + application: TestApp, + subscription: self(), + event: %{} + ) + + :telemetry.span([:commanded, :event_store, :ack_event], meta, fn -> + {:ok, meta} + end) + + assert_receive {:span, + span( + name: "ack_event TestApp", + kind: :internal, + attributes: attributes + )}, + 1000 + + assert :otel_attributes.map(attributes) == %{ + "messaging.system": "commanded", + "messaging.operation.type": :settle, + "messaging.operation.name": "ack_event", + "code.function": "ack_event", + "commanded.application": TestApp + } + end + end + + describe "error handling" do + setup do + detach_handlers() + OTelEventStore.setup() + :ok + end + + test "sets error status when stop includes error" do + stream_uuid = UUID.uuid4() + + meta = + build_event_store_append_to_stream_metadata( + application: TestApp, + stream_uuid: stream_uuid, + expected_version: 0 + ) + + :telemetry.execute([:commanded, :event_store, :append_to_stream, :start], %{}, meta) + + stop_meta = Map.put(meta, :error, :stream_not_found) + + :telemetry.execute( + [:commanded, :event_store, :append_to_stream, :stop], + %{duration: 1000}, + stop_meta + ) + + assert_receive {:span, + span( + name: "append_to_stream TestApp", + status: {:status, :error, error_message}, + attributes: span_attrs + )}, + 1000 + + assert error_message == ":stream_not_found" + + assert :otel_attributes.map(span_attrs) == %{ + "messaging.system": "commanded", + "messaging.operation.type": :publish, + "messaging.operation.name": "append_to_stream", + "code.function": "append_to_stream", + "commanded.application": TestApp, + "commanded.stream.uuid": stream_uuid, + "commanded.expected_version": 0, + "error.type": "stream_not_found" + } + end + + test "handles RuntimeError exception" do + stream_uuid = UUID.uuid4() + + meta = + build_event_store_append_to_stream_metadata( + application: TestApp, + stream_uuid: stream_uuid, + expected_version: 0 + ) + + :telemetry.execute([:commanded, :event_store, :append_to_stream, :start], %{}, meta) + + exception_meta = + Map.merge(meta, %{ + kind: :error, + reason: %RuntimeError{message: "connection failed"}, + stacktrace: [] + }) + + :telemetry.execute( + [:commanded, :event_store, :append_to_stream, :exception], + %{duration: 100}, + exception_meta + ) + + assert_receive {:span, + span( + name: "append_to_stream TestApp", + status: {:status, :error, error_msg}, + attributes: span_attrs + )}, + 1000 + + assert error_msg == "** (RuntimeError) connection failed" + + assert :otel_attributes.map(span_attrs) == %{ + "messaging.system": "commanded", + "messaging.operation.type": :publish, + "messaging.operation.name": "append_to_stream", + "code.function": "append_to_stream", + "commanded.application": TestApp, + "commanded.stream.uuid": stream_uuid, + "commanded.expected_version": 0, + "erlang.exception.kind": :error, + "error.type": "Elixir.RuntimeError" + } + end + + test "records exception event with proper attributes" do + stream_uuid = UUID.uuid4() + + meta = + build_event_store_append_to_stream_metadata( + application: TestApp, + stream_uuid: stream_uuid, + expected_version: 0 + ) + + :telemetry.execute([:commanded, :event_store, :append_to_stream, :start], %{}, meta) + + exception_meta = + Map.merge(meta, %{ + kind: :error, + reason: %RuntimeError{message: "failed"}, + stacktrace: [] + }) + + :telemetry.execute( + [:commanded, :event_store, :append_to_stream, :exception], + %{duration: 100}, + exception_meta + ) + + assert_receive {:span, + span( + name: "append_to_stream TestApp", + events: events + )}, + 1000 + + events_list = :otel_events.list(events) + exception_event = Enum.find(events_list, fn event -> elem(event, 2) == :exception end) + {:event, _timestamp, :exception, attrs_tuple} = exception_event + {:attributes, _, _, _, attrs_map} = attrs_tuple + + assert attrs_map[:"exception.type"] == "Elixir.RuntimeError" + assert attrs_map[:"exception.message"] == "failed" + end + end + + describe "record_snapshot spans" do + setup do + detach_handlers() + OTelEventStore.setup() + :ok + end + + test "creates span with correct attributes including source_uuid" do + source_uuid = UUID.uuid4() + + meta = + build_event_store_record_snapshot_metadata( + application: TestApp, + source_uuid: source_uuid + ) + + :telemetry.span([:commanded, :event_store, :record_snapshot], meta, fn -> + {:ok, meta} + end) + + assert_receive {:span, + span( + name: "record_snapshot TestApp", + kind: :internal, + attributes: attributes + )}, + 1000 + + assert :otel_attributes.map(attributes) == %{ + "messaging.system": "commanded", + "messaging.operation.type": :publish, + "messaging.operation.name": "record_snapshot", + "code.function": "record_snapshot", + "commanded.application": TestApp, + "commanded.source.uuid": source_uuid + } + end + end + + describe "read_snapshot spans" do + setup do + detach_handlers() + OTelEventStore.setup() + :ok + end + + test "creates span with correct attributes including source_uuid" do + source_uuid = UUID.uuid4() + + meta = + build_event_store_read_snapshot_metadata( + application: TestApp, + source_uuid: source_uuid + ) + + :telemetry.span([:commanded, :event_store, :read_snapshot], meta, fn -> + {:ok, meta} + end) + + assert_receive {:span, + span( + name: "read_snapshot TestApp", + kind: :internal, + attributes: attributes + )}, + 1000 + + assert :otel_attributes.map(attributes) == %{ + "messaging.system": "commanded", + "messaging.operation.type": :receive, + "messaging.operation.name": "read_snapshot", + "code.function": "read_snapshot", + "commanded.application": TestApp, + "commanded.source.uuid": source_uuid + } + end + end + + describe "delete_snapshot spans" do + setup do + detach_handlers() + OTelEventStore.setup() + :ok + end + + test "creates span with correct attributes including source_uuid" do + source_uuid = UUID.uuid4() + + meta = + build_event_store_delete_snapshot_metadata( + application: TestApp, + source_uuid: source_uuid + ) + + :telemetry.span([:commanded, :event_store, :delete_snapshot], meta, fn -> + {:ok, meta} + end) + + assert_receive {:span, + span( + name: "delete_snapshot TestApp", + kind: :internal, + attributes: attributes + )}, + 1000 + + assert :otel_attributes.map(attributes) == %{ + "messaging.system": "commanded", + "messaging.operation.name": "delete_snapshot", + "code.function": "delete_snapshot", + "commanded.application": TestApp, + "commanded.source.uuid": source_uuid + } + end + end + + describe "subscribe spans" do + setup do + detach_handlers() + OTelEventStore.setup() + :ok + end + + test "creates span with correct attributes including stream_uuid" do + stream_uuid = UUID.uuid4() + + meta = + build_event_store_subscribe_metadata( + application: TestApp, + stream_uuid: stream_uuid + ) + + :telemetry.span([:commanded, :event_store, :subscribe], meta, fn -> + {:ok, meta} + end) + + assert_receive {:span, + span( + name: "subscribe TestApp", + kind: :internal, + attributes: attributes + )}, + 1000 + + assert :otel_attributes.map(attributes) == %{ + "messaging.system": "commanded", + "messaging.operation.type": :receive, + "messaging.operation.name": "subscribe", + "code.function": "subscribe", + "commanded.application": TestApp, + "commanded.stream.uuid": stream_uuid + } + end + end + + describe "unsubscribe spans" do + setup do + detach_handlers() + OTelEventStore.setup() + :ok + end + + test "creates span with correct attributes" do + meta = + build_event_store_unsubscribe_metadata( + application: TestApp, + subscription: self() + ) + + :telemetry.span([:commanded, :event_store, :unsubscribe], meta, fn -> + {:ok, meta} + end) + + assert_receive {:span, + span( + name: "unsubscribe TestApp", + kind: :internal, + attributes: attributes + )}, + 1000 + + assert :otel_attributes.map(attributes) == %{ + "messaging.system": "commanded", + "messaging.operation.name": "unsubscribe", + "code.function": "unsubscribe", + "commanded.application": TestApp + } + end + end + + describe "operations without stream_uuid" do + setup do + detach_handlers() + OTelEventStore.setup() + :ok + end + + test "creates span name without stream_uuid suffix" do + meta = build_event_store_delete_subscription_metadata(application: TestApp) + + :telemetry.span([:commanded, :event_store, :delete_subscription], meta, fn -> + {:ok, meta} + end) + + assert_receive {:span, + span( + name: "delete_subscription TestApp", + kind: :internal + )}, + 1000 + end + end + + defp detach_handlers do + for event <- @events do + for suffix <- [:start, :stop, :exception] do + for handler <- :telemetry.list_handlers([:commanded, :event_store, event, suffix]) do + :telemetry.detach(handler.id) + end + end + end + end +end diff --git a/test/support/factory.ex b/test/support/factory.ex index fcef045e..e657d7f5 100644 --- a/test/support/factory.ex +++ b/test/support/factory.ex @@ -706,6 +706,286 @@ defmodule Commanded.TestSupport.Factory do {event_name, measurements, metadata} end + def build_telemetry_event(:event_store_append_to_stream_start, opts) do + measurements = %{ + monotonic_time: Keyword.get(opts, :monotonic_time, -576_460_751_180_934_041), + system_time: Keyword.get(opts, :system_time, 1_769_132_756_438_728_125) + } + + metadata = build_event_store_append_to_stream_metadata(opts) + event_name = [:commanded, :event_store, :append_to_stream, :start] + + {event_name, measurements, metadata} + end + + def build_telemetry_event(:event_store_append_to_stream_stop, opts) do + measurements = %{ + monotonic_time: Keyword.get(opts, :monotonic_time, -576_460_751_102_678_041), + duration: Keyword.get(opts, :duration, 78_256_000) + } + + metadata = + opts + |> build_event_store_append_to_stream_metadata() + |> maybe_put(:error, opts) + + event_name = [:commanded, :event_store, :append_to_stream, :stop] + + {event_name, measurements, metadata} + end + + def build_telemetry_event(:event_store_stream_forward_start, opts) do + measurements = %{ + monotonic_time: Keyword.get(opts, :monotonic_time, -576_460_750_916_612_542), + system_time: Keyword.get(opts, :system_time, 1_769_137_874_780_173_542) + } + + metadata = build_event_store_stream_forward_metadata(opts) + event_name = [:commanded, :event_store, :stream_forward, :start] + + {event_name, measurements, metadata} + end + + def build_telemetry_event(:event_store_stream_forward_stop, opts) do + measurements = %{ + monotonic_time: Keyword.get(opts, :monotonic_time, -576_460_750_895_139_292), + duration: Keyword.get(opts, :duration, 21_473_250) + } + + metadata = + opts + |> build_event_store_stream_forward_metadata() + |> maybe_put(:error, opts) + + event_name = [:commanded, :event_store, :stream_forward, :stop] + + {event_name, measurements, metadata} + end + + def build_telemetry_event(:event_store_subscribe_to_start, opts) do + measurements = %{ + monotonic_time: Keyword.get(opts, :monotonic_time, -576_460_750_834_979_708), + system_time: Keyword.get(opts, :system_time, 1_769_132_779_441_574_667) + } + + metadata = build_event_store_subscribe_to_metadata(opts) + event_name = [:commanded, :event_store, :subscribe_to, :start] + + {event_name, measurements, metadata} + end + + def build_telemetry_event(:event_store_subscribe_to_stop, opts) do + measurements = %{ + monotonic_time: Keyword.get(opts, :monotonic_time, -576_460_750_766_990_583), + duration: Keyword.get(opts, :duration, 67_989_125) + } + + metadata = + opts + |> build_event_store_subscribe_to_metadata() + |> maybe_put(:error, opts) + + event_name = [:commanded, :event_store, :subscribe_to, :stop] + + {event_name, measurements, metadata} + end + + def build_telemetry_event(:event_store_record_snapshot_start, opts) do + measurements = %{ + monotonic_time: Keyword.get(opts, :monotonic_time, -576_460_750_928_468_041), + system_time: Keyword.get(opts, :system_time, 1_769_132_779_348_086_042) + } + + metadata = build_event_store_record_snapshot_metadata(opts) + event_name = [:commanded, :event_store, :record_snapshot, :start] + + {event_name, measurements, metadata} + end + + def build_telemetry_event(:event_store_record_snapshot_stop, opts) do + measurements = %{ + monotonic_time: Keyword.get(opts, :monotonic_time, -576_460_750_904_372_541), + duration: Keyword.get(opts, :duration, 24_095_500) + } + + metadata = + opts + |> build_event_store_record_snapshot_metadata() + |> maybe_put(:error, opts) + + event_name = [:commanded, :event_store, :record_snapshot, :stop] + + {event_name, measurements, metadata} + end + + def build_telemetry_event(:event_store_read_snapshot_start, opts) do + measurements = %{ + monotonic_time: Keyword.get(opts, :monotonic_time, -576_460_750_548_527_667), + system_time: Keyword.get(opts, :system_time, 1_769_137_875_148_258_375) + } + + metadata = build_event_store_read_snapshot_metadata(opts) + event_name = [:commanded, :event_store, :read_snapshot, :start] + + {event_name, measurements, metadata} + end + + def build_telemetry_event(:event_store_read_snapshot_stop, opts) do + measurements = %{ + monotonic_time: Keyword.get(opts, :monotonic_time, -576_460_750_534_117_958), + duration: Keyword.get(opts, :duration, 14_409_709) + } + + metadata = + opts + |> build_event_store_read_snapshot_metadata() + |> maybe_put(:error, opts) + + event_name = [:commanded, :event_store, :read_snapshot, :stop] + + {event_name, measurements, metadata} + end + + def build_telemetry_event(:event_store_ack_event_start, opts) do + measurements = %{ + monotonic_time: Keyword.get(opts, :monotonic_time, -576_460_750_664_247_583), + system_time: Keyword.get(opts, :system_time, 1_769_137_875_032_538_583) + } + + metadata = build_event_store_ack_event_metadata(opts) + event_name = [:commanded, :event_store, :ack_event, :start] + + {event_name, measurements, metadata} + end + + def build_telemetry_event(:event_store_ack_event_stop, opts) do + measurements = %{ + monotonic_time: Keyword.get(opts, :monotonic_time, -576_460_750_645_331_875), + duration: Keyword.get(opts, :duration, 18_915_708) + } + + metadata = + opts + |> build_event_store_ack_event_metadata() + |> maybe_put(:error, opts) + + event_name = [:commanded, :event_store, :ack_event, :stop] + + {event_name, measurements, metadata} + end + + def build_telemetry_event(:event_store_delete_snapshot_start, opts) do + measurements = %{ + monotonic_time: Keyword.get(opts, :monotonic_time, -576_460_750_736_021_125), + system_time: Keyword.get(opts, :system_time, 1_769_134_205_344_113_250) + } + + metadata = build_event_store_delete_snapshot_metadata(opts) + event_name = [:commanded, :event_store, :delete_snapshot, :start] + + {event_name, measurements, metadata} + end + + def build_telemetry_event(:event_store_delete_snapshot_stop, opts) do + measurements = %{ + monotonic_time: Keyword.get(opts, :monotonic_time, -576_460_750_706_717_459), + duration: Keyword.get(opts, :duration, 29_303_666) + } + + metadata = + opts + |> build_event_store_delete_snapshot_metadata() + |> maybe_put(:error, opts) + + event_name = [:commanded, :event_store, :delete_snapshot, :stop] + + {event_name, measurements, metadata} + end + + def build_telemetry_event(:event_store_subscribe_start, opts) do + measurements = %{ + monotonic_time: Keyword.get(opts, :monotonic_time, -576_460_750_920_081_042), + system_time: Keyword.get(opts, :system_time, 1_769_134_231_524_992_167) + } + + metadata = build_event_store_subscribe_metadata(opts) + event_name = [:commanded, :event_store, :subscribe, :start] + + {event_name, measurements, metadata} + end + + def build_telemetry_event(:event_store_subscribe_stop, opts) do + measurements = %{ + monotonic_time: Keyword.get(opts, :monotonic_time, -576_460_750_920_064_292), + duration: Keyword.get(opts, :duration, 16_750) + } + + metadata = + opts + |> build_event_store_subscribe_metadata() + |> maybe_put(:error, opts) + + event_name = [:commanded, :event_store, :subscribe, :stop] + + {event_name, measurements, metadata} + end + + def build_telemetry_event(:event_store_unsubscribe_start, opts) do + measurements = %{ + monotonic_time: Keyword.get(opts, :monotonic_time, -576_460_751_014_001_958), + system_time: Keyword.get(opts, :system_time, 1_769_134_231_431_071_083) + } + + metadata = build_event_store_unsubscribe_metadata(opts) + event_name = [:commanded, :event_store, :unsubscribe, :start] + + {event_name, measurements, metadata} + end + + def build_telemetry_event(:event_store_unsubscribe_stop, opts) do + measurements = %{ + monotonic_time: Keyword.get(opts, :monotonic_time, -576_460_750_998_181_375), + duration: Keyword.get(opts, :duration, 15_820_583) + } + + metadata = + opts + |> build_event_store_unsubscribe_metadata() + |> maybe_put(:error, opts) + + event_name = [:commanded, :event_store, :unsubscribe, :stop] + + {event_name, measurements, metadata} + end + + def build_telemetry_event(:event_store_delete_subscription_start, opts) do + measurements = %{ + monotonic_time: Keyword.get(opts, :monotonic_time, -576_460_751_154_259_667), + system_time: Keyword.get(opts, :system_time, 1_769_134_231_290_813_417) + } + + metadata = build_event_store_delete_subscription_metadata(opts) + event_name = [:commanded, :event_store, :delete_subscription, :start] + + {event_name, measurements, metadata} + end + + def build_telemetry_event(:event_store_delete_subscription_stop, opts) do + measurements = %{ + monotonic_time: Keyword.get(opts, :monotonic_time, -576_460_751_123_691_917), + duration: Keyword.get(opts, :duration, 30_567_750) + } + + metadata = + opts + |> build_event_store_delete_subscription_metadata() + |> maybe_put(:error, opts) + + event_name = [:commanded, :event_store, :delete_subscription, :stop] + + {event_name, measurements, metadata} + end + def build_account_scenario(opts \\ []) do account_id = Keyword.get(opts, :account_id, UUID.uuid4()) owner = Keyword.get(opts, :owner, "Test User") @@ -775,4 +1055,174 @@ defmodule Commanded.TestSupport.Factory do account_id: account_id } end + + def build_event_store_append_to_stream_metadata(opts \\ []) do + defaults = [ + application: Keyword.get(opts, :application, MockApp), + stream_uuid: Keyword.get(opts, :stream_uuid, UUID.uuid4()), + expected_version: Keyword.get(opts, :expected_version, 0) + ] + + opts = Keyword.merge(defaults, opts) + + %{ + application: Keyword.fetch!(opts, :application), + stream_uuid: Keyword.fetch!(opts, :stream_uuid), + expected_version: Keyword.fetch!(opts, :expected_version) + } + end + + def build_event_store_stream_forward_metadata(opts \\ []) do + defaults = [ + application: Keyword.get(opts, :application, MockApp), + stream_uuid: Keyword.get(opts, :stream_uuid, UUID.uuid4()), + start_version: Keyword.get(opts, :start_version, 0), + read_batch_size: Keyword.get(opts, :read_batch_size, 1000) + ] + + opts = Keyword.merge(defaults, opts) + + %{ + application: Keyword.fetch!(opts, :application), + stream_uuid: Keyword.fetch!(opts, :stream_uuid), + start_version: Keyword.fetch!(opts, :start_version), + read_batch_size: Keyword.fetch!(opts, :read_batch_size) + } + end + + def build_event_store_subscribe_to_metadata(opts \\ []) do + defaults = [ + application: Keyword.get(opts, :application, MockApp), + stream_uuid: Keyword.get(opts, :stream_uuid, :all), + subscription_name: Keyword.get(opts, :subscription_name, "test_subscription"), + subscriber: Keyword.get(opts, :subscriber, self()), + start_from: Keyword.get(opts, :start_from, :origin) + ] + + opts = Keyword.merge(defaults, opts) + + %{ + application: Keyword.fetch!(opts, :application), + stream_uuid: Keyword.fetch!(opts, :stream_uuid), + subscription_name: Keyword.fetch!(opts, :subscription_name), + subscriber: Keyword.fetch!(opts, :subscriber), + start_from: Keyword.fetch!(opts, :start_from) + } + end + + def build_event_store_ack_event_metadata(opts \\ []) do + defaults = [ + application: Keyword.get(opts, :application, MockApp), + subscription: Keyword.get(opts, :subscription, self()), + event: Keyword.get(opts, :event, %{}) + ] + + opts = Keyword.merge(defaults, opts) + + %{ + application: Keyword.fetch!(opts, :application), + subscription: Keyword.fetch!(opts, :subscription), + event: Keyword.fetch!(opts, :event) + } + end + + def build_event_store_record_snapshot_metadata(opts \\ []) do + source_uuid = Keyword.get(opts, :source_uuid, UUID.uuid4()) + + snapshot = + Keyword.get_lazy(opts, :snapshot, fn -> + %Commanded.EventStore.SnapshotData{ + source_uuid: source_uuid, + source_version: 5, + source_type: "TestAggregate", + data: %{state: "test"}, + metadata: %{}, + created_at: DateTime.utc_now() + } + end) + + defaults = [ + application: Keyword.get(opts, :application, MockApp), + snapshot: snapshot + ] + + opts = Keyword.merge(defaults, opts) + + %{ + application: Keyword.fetch!(opts, :application), + snapshot: Keyword.fetch!(opts, :snapshot) + } + end + + def build_event_store_read_snapshot_metadata(opts \\ []) do + defaults = [ + application: Keyword.get(opts, :application, MockApp), + source_uuid: Keyword.get(opts, :source_uuid, UUID.uuid4()) + ] + + opts = Keyword.merge(defaults, opts) + + %{ + application: Keyword.fetch!(opts, :application), + source_uuid: Keyword.fetch!(opts, :source_uuid) + } + end + + def build_event_store_delete_snapshot_metadata(opts \\ []) do + defaults = [ + application: Keyword.get(opts, :application, MockApp), + source_uuid: Keyword.get(opts, :source_uuid, UUID.uuid4()) + ] + + opts = Keyword.merge(defaults, opts) + + %{ + application: Keyword.fetch!(opts, :application), + source_uuid: Keyword.fetch!(opts, :source_uuid) + } + end + + def build_event_store_subscribe_metadata(opts \\ []) do + defaults = [ + application: Keyword.get(opts, :application, MockApp), + stream_uuid: Keyword.get(opts, :stream_uuid, UUID.uuid4()) + ] + + opts = Keyword.merge(defaults, opts) + + %{ + application: Keyword.fetch!(opts, :application), + stream_uuid: Keyword.fetch!(opts, :stream_uuid) + } + end + + def build_event_store_unsubscribe_metadata(opts \\ []) do + defaults = [ + application: Keyword.get(opts, :application, MockApp), + subscription: Keyword.get(opts, :subscription, self()) + ] + + opts = Keyword.merge(defaults, opts) + + %{ + application: Keyword.fetch!(opts, :application), + subscription: Keyword.fetch!(opts, :subscription) + } + end + + def build_event_store_delete_subscription_metadata(opts \\ []) do + defaults = [ + application: Keyword.get(opts, :application, MockApp), + subscribe_to: Keyword.get(opts, :subscribe_to, :all), + handler_name: Keyword.get(opts, :handler_name, "TestHandler") + ] + + opts = Keyword.merge(defaults, opts) + + %{ + application: Keyword.fetch!(opts, :application), + subscribe_to: Keyword.fetch!(opts, :subscribe_to), + handler_name: Keyword.fetch!(opts, :handler_name) + } + end end diff --git a/test/support/test_domain.ex b/test/support/test_domain.ex index 0ea21030..e34c2b92 100644 --- a/test/support/test_domain.ex +++ b/test/support/test_domain.ex @@ -5,36 +5,43 @@ defmodule Commanded.TestSupport.TestDomain do defmodule OpenAccount do @moduledoc false + @derive Jason.Encoder defstruct [:account_id, :owner, :initial_balance] end defmodule DepositMoney do @moduledoc false + @derive Jason.Encoder defstruct [:account_id, :amount] end defmodule WithdrawMoney do @moduledoc false + @derive Jason.Encoder defstruct [:account_id, :amount] end defmodule AccountOpened do @moduledoc false + @derive Jason.Encoder defstruct [:account_id, :owner, :initial_balance] end defmodule MoneyDeposited do @moduledoc false + @derive Jason.Encoder defstruct [:account_id, :amount, :new_balance] end defmodule MoneyWithdrawn do @moduledoc false + @derive Jason.Encoder defstruct [:account_id, :amount, :new_balance] end defmodule Account do @moduledoc false + @derive Jason.Encoder defstruct [:account_id, :owner, :balance, :status] end end