Merge pull request #49 from ccamel/refactor/store-apps

Refactor/store apps

Author: ccamel

.github/scripts/sync_readme.escript

@@ -6,7 +6,7 @@
 %% between <!-- DEMO-START --> and <!-- DEMO-END --> markers.
 
 main([]) ->
-    main(["README.md", "examples/demo_bank.script"]);
+    main(["README.md", "apps/es_xp/examples/demo_bank.script"]);
 main([ReadmePath, ScriptPath]) ->
     {ok, ReadmeBin} = file:read_file(ReadmePath),
     {ok, ScriptBin} = file:read_file(ScriptPath),

.gitignore

@@ -16,3 +16,6 @@ deps
 _build/
 _checkouts/
 doc/
+
+# application files
+data/

README.md

@@ -29,6 +29,17 @@ Function shall return The initial aggregate state.
 """.
 -callback init() -> aggregate_state().
 
+-doc """
+Return the event type identifier for a given domain event payload.
+
+This callback is responsible for mapping a payload to its canonical event type
+(e.g. `user_created`, `user_deleted`). It must return a valid
+`es_contract_event:type()` for the payload.
+""".
+-callback event_type(Event) -> Type when
+    Event :: es_contract_event:payload(),
+    Type :: es_contract_event:type().
+
 -doc """
 Handle a domain command.
 

apps/es_kernel/src/es_contract_aggregate.erl

@@ -2,7 +2,7 @@
 
 -export([
     new/6,
-    key/1,
+    target/1,
     with_metadata/2,
     put_metadata/3,
     with_tags/2,
@@ -12,7 +12,7 @@
 
 -export_type([
     domain/0,
-    stream_id/0,
+    aggregate_id/0,
     sequence/0,
     type/0,
     metadata_key/0,
@@ -22,7 +22,7 @@
     tags/0,
     payload/0,
     t/0,
-    key/0
+    target/0
 ]).
 
 %%--------------------------------------------------------------------
@@ -32,8 +32,12 @@
 -doc "Domain identifier, representing a bounded context.".
 -type domain() :: atom().
 
--doc "Stream identifier, uniquely identifying the target aggregate stream.".
--type stream_id() :: binary().
+-doc """
+Aggregate identifier, uniquely identifying an aggregate instance.
+
+Can be any term (UUID, binary, integer, etc.).
+""".
+-type aggregate_id() :: term().
 
 -doc "Sequence of the command when batching operations (optional semantic).".
 -type sequence() :: non_neg_integer().
@@ -67,43 +71,49 @@ A command represents an intent to change the state of an aggregate in a specific
 It consists of:
 - `domain`: The domain this command belongs to
 - `type`: The type of command to execute
-- `stream_id`: Identifier of the target aggregate stream
+- `aggregate_id`: Identifier of the target aggregate instance
 - `sequence`: Optional sequencing information for idempotency/correlation
 - `metadata`: Additional contextual information (user, timestamp, correlation ID, etc.)
 - `tags`: Labels for categorization or routing
-- `payload`: The actual command data (self-sufficient, includes aggregate ID(s))
+- `payload`: The actual command data
 """.
 -type t() :: #{
     domain := domain(),
     type := type(),
-    stream_id := stream_id(),
+    aggregate_id := aggregate_id(),
     sequence := sequence(),
     metadata := metadata(),
     tags := tags(),
     payload := payload()
 }.
 
--type key() :: {domain(), stream_id(), sequence()}.
+-doc """
+Command target identifier.
+
+The target is a tuple of the Domain and the AggregateId that identifies
+which aggregate this command is addressed to.
+""".
+-type target() :: {domain(), aggregate_id()}.
 
 %%--------------------------------------------------------------------
 %% Functions
 %%--------------------------------------------------------------------
 
--spec new(domain(), type(), stream_id(), sequence(), metadata(), payload()) -> t().
-new(Domain, Type, StreamId, Sequence, Metadata, Payload) ->
+-spec new(domain(), type(), aggregate_id(), sequence(), metadata(), payload()) -> t().
+new(Domain, Type, AggregateId, Sequence, Metadata, Payload) ->
     #{
         domain => Domain,
         type => Type,
-        stream_id => StreamId,
+        aggregate_id => AggregateId,
         sequence => Sequence,
         metadata => Metadata,
         tags => [],
         payload => Payload
     }.
 
--spec key(t()) -> key().
-key(#{domain := D, stream_id := S, sequence := Seq}) ->
-    {D, S, Seq}.
+-spec target(t()) -> target().
+target(#{domain := D, aggregate_id := AggId}) ->
+    {D, AggId}.
 
 -spec with_metadata(metadata(), t()) -> t().
 with_metadata(Meta, Command) when is_map(Command) ->

apps/es_kernel/src/es_contract_command.erl

@@ -32,8 +32,20 @@
 -doc "Domain identifier, representing a bounded context.".
 -type domain() :: atom().
 
--doc "Stream identifier, uniquely identifying an event stream within a domain.".
--type stream_id() :: binary().
+-doc """
+Aggregate identifier, uniquely identifying an aggregate instance.
+
+Can be any term (UUID, binary, integer, etc.).
+""".
+-type aggregate_id() :: term().
+
+-doc """
+Stream identifier, uniquely identifying an event stream.
+
+A stream is identified by a tuple of {Domain, AggregateId}, ensuring no collisions
+across different domains and aggregate instances.
+""".
+-type stream_id() :: {domain(), aggregate_id()}.
 
 -doc "Sequence number of the event within its stream, starting from 0.".
 -type sequence() :: non_neg_integer().
@@ -84,10 +96,10 @@ An event represents a fact that something has happened in the system. It consist
 -doc """
 Composite key uniquely identifying an event.
 
-The key is a tuple of the Domain, the StreamId and the Sequence that uniquely identifies
+The key is a tuple of the StreamId and the Sequence that uniquely identifies
 an event within the entire event store.
 """.
--type key() :: {domain(), stream_id(), sequence()}.
+-type key() :: {stream_id(), sequence()}.
 
 %%--------------------------------------------------------------------
 %% Functions
@@ -106,8 +118,8 @@ new(Domain, Type, StreamId, Sequence, Metadata, Payload) ->
     }.
 
 -spec key(t()) -> key().
-key(#{domain := D, stream_id := S, sequence := Seq}) ->
-    {D, S, Seq}.
+key(#{stream_id := S, sequence := Seq}) ->
+    {S, Seq}.
 
 -spec with_metadata(metadata(), t()) -> t().
 with_metadata(Meta, Event) when is_map(Event) ->

apps/es_kernel/src/es_contract_event.erl

@@ -27,8 +27,18 @@
 -doc "Domain identifier, representing a bounded context.".
 -type domain() :: atom().
 
--doc "Stream identifier, uniquely identifying an event stream within a domain.".
--type stream_id() :: binary().
+-doc """
+Aggregate identifier, uniquely identifying an aggregate instance (can be UUID, binary, etc.).
+""".
+-type aggregate_id() :: term().
+
+-doc """
+Stream identifier, uniquely identifying an event stream.
+
+A stream is identified by a tuple of {Domain, AggregateId}, ensuring no collisions
+across different domains and aggregate instances.
+""".
+-type stream_id() :: {domain(), aggregate_id()}.
 
 -doc "Sequence number of the event within its stream, starting from 0.".
 -type sequence() :: non_neg_integer().
@@ -60,10 +70,11 @@ A snapshot represents a point-in-time capture of an aggregate's state. It consis
 
 -doc """
 Composite key uniquely identifying a snapshot.
-The key is a tuple of the Domain, the StreamId and the Sequence that uniquely identifies
+
+The key is a tuple of the StreamId and the Sequence that uniquely identifies
 a snapshot within the entire snapshot store.
 """.
--type key() :: {domain(), stream_id(), sequence()}.
+-type key() :: {stream_id(), sequence()}.
 
 %%--------------------------------------------------------------------
 %% Functions
@@ -80,8 +91,8 @@ new(Domain, StreamId, Sequence, Metadata, State) ->
     }.
 
 -spec key(t()) -> key().
-key(#{domain := D, stream_id := S, sequence := Seq}) ->
-    {D, S, Seq}.
+key(#{stream_id := S, sequence := Seq}) ->
+    {S, Seq}.
 
 -spec with_metadata(metadata(), t()) -> t().
 with_metadata(Meta, Snap) when is_map(Snap) ->

apps/es_kernel/src/es_contract_snapshot.erl

@@ -6,7 +6,8 @@
     {applications, [kernel, stdlib]},
     {env, [
         {event_store, es_store_ets},
-        {snapshot_store, es_store_ets}
+        {snapshot_store, es_store_ets},
+        {snapshot_interval, 0}
     ]},
     {modules, [
         es_contract_aggregate,