packages feed

eventium-core 0.2.1 → 0.3.0

raw patch · 7 files changed

+110/−61 lines, 7 files

Files

eventium-core.cabal view
@@ -5,7 +5,7 @@ -- see: https://github.com/sol/hpack  name:           eventium-core-version:        0.2.1+version:        0.3.0 synopsis:       Core module for eventium description:    Eventium-core provides the core abstractions and utilities for building event sourcing systems in Haskell.                 It includes event store interfaces, command handlers, projections, event handlers, event publishers,
src/Eventium/CommandDispatcher.hs view
@@ -14,10 +14,15 @@   ) where +import Control.Monad.IO.Class (MonadIO) import qualified Data.Text as T+import Data.Typeable (Typeable)+import Eventium.Codec (Codec) import Eventium.CommandHandler (CommandHandler, CommandHandlerError (..), applyCommandHandler)-import Eventium.ProcessManager (CommandDispatchResult (..), CommandDispatcher, RejectionReason (..), mkCommandDispatcher)-import Eventium.Store.Class (VersionedEventStoreReader, VersionedEventStoreWriter)+import Eventium.ProcessManager (CommandDispatchResult (..), CommandDispatcher (..), RejectionReason (..))+import Eventium.Store.Class (EventStoreWriter, VersionedEventStoreReader, metadataEnrichingEventStoreWriterWithEnricher)+import Eventium.Store.Types (EventVersion, TaggedEvent)+import Eventium.UUID (UUID)  -- | An embedded command handler paired with an error formatter. --@@ -45,6 +50,9 @@  -- | Build a 'CommandDispatcher' from a list of 'AggregateHandler's. --+-- Accepts a tagged writer and codec; the 'MetadataEnricher' supplied at+-- dispatch time is applied to create a per-dispatch enriched writer.+-- -- Tries each handler in order: -- --   * @Right (e:es)@ — command matched and produced events → 'CommandSucceeded'@@ -54,19 +62,22 @@ -- -- If no handler matches (all return @Right []@), reports 'CommandSucceeded' (no-op). commandHandlerDispatcher ::-  (Monad m) =>-  VersionedEventStoreWriter m event ->+  (MonadIO m, Typeable event) =>+  Codec event encoded ->+  EventStoreWriter UUID EventVersion m (TaggedEvent encoded) ->   VersionedEventStoreReader m event ->   [AggregateHandler event command] ->   CommandDispatcher m command-commandHandlerDispatcher writer reader handlers =-  mkCommandDispatcher $ \uuid cmd -> go handlers uuid cmd+commandHandlerDispatcher codec taggedWriter reader handlers =+  CommandDispatcher $ \uuid cmd enricher ->+    let writer = metadataEnrichingEventStoreWriterWithEnricher enricher codec taggedWriter+     in go handlers writer uuid cmd   where-    go [] _ _ = pure CommandSucceeded-    go (AggregateHandler handler formatErr : rest) uuid cmd = do+    go [] _ _ _ = pure CommandSucceeded+    go (AggregateHandler handler formatErr : rest) writer uuid cmd = do       result <- applyCommandHandler writer reader handler uuid cmd       case result of         Right (_ : _) -> pure CommandSucceeded         Left (CommandRejected err) -> pure (CommandFailed (formatErr err))         Left (ConcurrencyConflict _) -> pure (CommandFailed "Concurrency conflict")-        Right [] -> go rest uuid cmd+        Right [] -> go rest writer uuid cmd
src/Eventium/EventPublisher.hs view
@@ -12,10 +12,12 @@   ( EventPublisher (..),     synchronousPublisher,     publishingEventStoreWriter,-    publishingEventStoreWriterTagged,+    publishingTaggedCodecEventStoreWriter,   ) where +import Control.Exception (throw)+import Eventium.Codec (Codec (..), DecodeError (..)) import Eventium.EventHandler import Eventium.Store.Class import Eventium.UUID@@ -61,14 +63,19 @@         return $ Right endVersion  -- | Like 'publishingEventStoreWriter' but for writers that accept--- 'TaggedEvent's. The metadata from each tagged event is preserved--- in the 'StreamEvent' wrappers passed to the publisher.-publishingEventStoreWriterTagged ::+-- @TaggedEvent encoded@. Each tagged event's payload is decoded through the+-- supplied 'Codec' before publishing — handlers receive domain events while+-- the writer stores serialized payloads. Metadata from each 'TaggedEvent' is+-- preserved in the 'StreamEvent' wrappers passed to the publisher.+--+-- Throws 'DecodeError' if any event fails to decode.+publishingTaggedCodecEventStoreWriter ::   (Monad m) =>-  VersionedEventStoreWriter m (TaggedEvent event) ->+  Codec event encoded ->+  VersionedEventStoreWriter m (TaggedEvent encoded) ->   EventPublisher m event ->-  VersionedEventStoreWriter m (TaggedEvent event)-publishingEventStoreWriterTagged (EventStoreWriter write) (EventPublisher publish) =+  VersionedEventStoreWriter m (TaggedEvent encoded)+publishingTaggedCodecEventStoreWriter codec (EventStoreWriter write) (EventPublisher publish) =   EventStoreWriter $ \uuid expectedPos taggedEvents -> do     result <- write uuid expectedPos taggedEvents     case result of@@ -77,7 +84,12 @@         let startVersion = endVersion - fromIntegral (length taggedEvents) + 1             versionedEvents =               zipWith-                (\v (TaggedEvent meta e) -> StreamEvent uuid v meta e)+                ( \v (TaggedEvent meta enc) ->+                    let event = case codec.decode enc of+                          Just e -> e+                          Nothing -> throw $ DecodeError "publishingTaggedCodecEventStoreWriter" "Failed to decode tagged event payload"+                     in StreamEvent uuid v meta event+                )                 [startVersion ..]                 taggedEvents         publish uuid versionedEvents
src/Eventium/ProcessManager.hs view
@@ -30,8 +30,8 @@ import Data.Text (Text) import Eventium.EventHandler (EventHandler (..)) import Eventium.Projection-import Eventium.Store.Class (GlobalEventStoreReader)-import Eventium.Store.Types+import Eventium.Store.Class (GlobalEventStoreReader, VersionedStreamEvent)+import Eventium.Store.Types (MetadataEnricher) import Eventium.UUID  -- | A 'ProcessManager' manages interaction between event streams. It@@ -55,24 +55,29 @@  -- | A side effect that a 'ProcessManager' wants to perform. This is a pure -- data type — it describes /what/ should happen, not /how/.+--+-- Each constructor carries a 'MetadataEnricher' so the saga can inject+-- application-level metadata (e.g. @occurredAt@) into events produced by the+-- dispatched command. Use 'id' when no enrichment is needed. data ProcessManagerEffect command   = -- | Issue a command to a specific aggregate (identified by 'UUID').-    IssueCommand UUID command+    IssueCommand UUID command MetadataEnricher   | -- | Issue a command with compensation: if the command fails, execute     -- the compensation effects produced by the failure handler.     --     -- The 'RejectionReason' argument to the compensation function is the failure reason     -- from 'CommandFailed'.-    IssueCommandWithCompensation UUID command (RejectionReason -> [ProcessManagerEffect command])+    IssueCommandWithCompensation UUID command MetadataEnricher (RejectionReason -> [ProcessManagerEffect command])  instance (Show command) => Show (ProcessManagerEffect command) where-  show (IssueCommand uuid cmd) = "IssueCommand " ++ show uuid ++ " " ++ show cmd-  show (IssueCommandWithCompensation uuid cmd _) =+  show (IssueCommand uuid cmd _) = "IssueCommand " ++ show uuid ++ " " ++ show cmd+  show (IssueCommandWithCompensation uuid cmd _ _) =     "IssueCommandWithCompensation " ++ show uuid ++ " " ++ show cmd ++ " <compensation>" +-- | Enricher is opaque (a function) — ignored in equality comparisons. instance (Eq command) => Eq (ProcessManagerEffect command) where-  IssueCommand u1 c1 == IssueCommand u2 c2 = u1 == u2 && c1 == c2-  IssueCommandWithCompensation u1 c1 _ == IssueCommandWithCompensation u2 c2 _ = u1 == u2 && c1 == c2+  IssueCommand u1 c1 _ == IssueCommand u2 c2 _ = u1 == u2 && c1 == c2+  IssueCommandWithCompensation u1 c1 _ _ == IssueCommandWithCompensation u2 c2 _ _ = u1 == u2 && c1 == c2   _ == _ = False  -- | Result of dispatching a command to an aggregate.@@ -89,22 +94,22 @@ -- Use 'fireAndForgetDispatcher' to adapt a legacy @UUID -> command -> m ()@ -- callback that does not report failures. newtype CommandDispatcher m command = CommandDispatcher-  { dispatchCommand :: UUID -> command -> m CommandDispatchResult+  { dispatchCommand :: UUID -> command -> MetadataEnricher -> m CommandDispatchResult   }  -- | Construct a 'CommandDispatcher' from a dispatch function. mkCommandDispatcher ::-  (UUID -> command -> m CommandDispatchResult) ->+  (UUID -> command -> MetadataEnricher -> m CommandDispatchResult) ->   CommandDispatcher m command mkCommandDispatcher = CommandDispatcher  -- | Adapt a legacy fire-and-forget dispatcher into a 'CommandDispatcher'--- that always reports 'CommandSucceeded'.+-- that always reports 'CommandSucceeded'. The enricher is ignored. fireAndForgetDispatcher ::   (Monad m) =>   (UUID -> command -> m ()) ->   CommandDispatcher m command-fireAndForgetDispatcher f = CommandDispatcher $ \uuid cmd ->+fireAndForgetDispatcher f = CommandDispatcher $ \uuid cmd _enricher ->   f uuid cmd >> pure CommandSucceeded  -- | Execute a list of 'ProcessManagerEffect's using the provided@@ -116,10 +121,10 @@   m () runProcessManagerEffects dispatcher = mapM_ go   where-    go (IssueCommand uuid cmd) =-      void $ dispatcher.dispatchCommand uuid cmd-    go (IssueCommandWithCompensation uuid cmd onFailure) = do-      result <- dispatcher.dispatchCommand uuid cmd+    go (IssueCommand uuid cmd enricher) =+      void $ dispatcher.dispatchCommand uuid cmd enricher+    go (IssueCommandWithCompensation uuid cmd enricher onFailure) = do+      result <- dispatcher.dispatchCommand uuid cmd enricher       case result of         CommandSucceeded -> pure ()         CommandFailed reason -> mapM_ go (onFailure reason)
src/Eventium/Store/Class.hs view
@@ -19,6 +19,7 @@     codecGlobalEventStoreReader,     codecEventStoreWriter,     metadataEnrichingEventStoreWriter,+    metadataEnrichingEventStoreWriterWithEnricher,     tagEvents,      -- * Type embedding@@ -171,30 +172,35 @@   EventStoreWriter key position m event codecEventStoreWriter codec = contramap codec.encode +-- | Like 'metadataEnrichingEventStoreWriterWithEnricher' with 'id' as the+-- enricher — no custom metadata modifications are applied.+metadataEnrichingEventStoreWriter ::+  (MonadIO m, Typeable event) =>+  Codec event encoded ->+  EventStoreWriter key position m (TaggedEvent encoded) ->+  EventStoreWriter key position m event+metadataEnrichingEventStoreWriter = metadataEnrichingEventStoreWriterWithEnricher id+ -- | Wraps an 'EventStoreWriter' that accepts 'TaggedEvent's, producing a -- writer that accepts domain events. Each event is encoded and tagged -- with metadata (event type name derived from 'Typeable', current--- UTC timestamp).------ Use this instead of 'codecEventStoreWriter' when you want metadata--- to be populated. The underlying writer must accept 'TaggedEvent's.------ @--- writer = metadataEnrichingEventStoreWriter myCodec taggedStore--- @-metadataEnrichingEventStoreWriter ::+-- UTC timestamp). The supplied 'MetadataEnricher' is applied to the+-- generated metadata before writing — use this to inject application-level+-- fields such as @occurredAt@.+metadataEnrichingEventStoreWriterWithEnricher ::   (MonadIO m, Typeable event) =>+  MetadataEnricher ->   Codec event encoded ->   EventStoreWriter key position m (TaggedEvent encoded) ->   EventStoreWriter key position m event-metadataEnrichingEventStoreWriter codec (EventStoreWriter write) =+metadataEnrichingEventStoreWriterWithEnricher enricher codec (EventStoreWriter write) =   EventStoreWriter $ \key pos events -> do     now <- liftIO getCurrentTime     let tagged =           map             ( \e ->                 TaggedEvent-                  (EventMetadata (T.pack . show $ typeOf e) Nothing Nothing (Just now))+                  (enricher (EventMetadata (T.pack . show $ typeOf e) Nothing Nothing (Just now) Nothing))                   (codec.encode e)             )             events@@ -211,7 +217,7 @@ tagEvents codec now =   map $ \e ->     TaggedEvent-      (EventMetadata (T.pack . show $ typeOf e) Nothing Nothing (Just now))+      (EventMetadata (T.pack . show $ typeOf e) Nothing Nothing (Just now) Nothing)       (codec.encode e)  -- | Like 'codecEventStoreWriter' but uses a 'TypeEmbedding' instead of
src/Eventium/Store/Types.hs view
@@ -15,6 +15,7 @@     -- * Event metadata     EventMetadata (..),     emptyMetadata,+    MetadataEnricher,     TaggedEvent (..),      -- * Expected position@@ -40,7 +41,8 @@   { eventType :: !Text,     correlationId :: !(Maybe UUID),     causationId :: !(Maybe UUID),-    createdAt :: !(Maybe UTCTime)+    createdAt :: !(Maybe UTCTime),+    occurredAt :: !(Maybe UTCTime)   }   deriving (Show, Eq, Generic) @@ -53,7 +55,14 @@  -- | Construct 'EventMetadata' with only an event type name. emptyMetadata :: Text -> EventMetadata-emptyMetadata et = EventMetadata et Nothing Nothing Nothing+emptyMetadata et = EventMetadata et Nothing Nothing Nothing Nothing++-- | Builder function for customizing event metadata.+--+-- Used to inject application-level metadata (e.g. 'occurredAt') into events+-- at write time. The enricher is applied after the base metadata is generated.+-- Use 'id' when no enrichment is needed.+type MetadataEnricher = EventMetadata -> EventMetadata  -- | An event paired with pre-computed metadata. Used to thread metadata -- through the 'EventStoreWriter' interface without changing its type signature.
tests/Eventium/CommandDispatcherSpec.hs view
@@ -3,6 +3,7 @@ module Eventium.CommandDispatcherSpec (spec) where  import Data.IORef+import Eventium.Codec import Eventium.CommandDispatcher import Eventium.CommandHandler import Eventium.ProcessManager (CommandDispatchResult (..), CommandDispatcher (..), RejectionReason (..))@@ -35,20 +36,25 @@       | otherwise = Right [Decremented]     decide' _ Unknown = Right [] +testCodec :: Codec CounterEvent CounterEvent+testCodec = Codec id Just+ -- | Simple IORef-based event store for testing.-mkTestStore :: IO (VersionedEventStoreWriter IO CounterEvent, VersionedEventStoreReader IO CounterEvent)+-- Returns a tagged writer, a reader that reads domain events, and raw reader.+mkTestStore :: IO (VersionedEventStoreWriter IO (TaggedEvent CounterEvent), VersionedEventStoreReader IO CounterEvent) mkTestStore = do   eventsRef <- newIORef ([] :: [VersionedStreamEvent CounterEvent])-  let writer = EventStoreWriter $ \uuid _expected events -> do+  let taggedWriter = EventStoreWriter $ \uuid _expected taggedEvents -> do         existing <- readIORef eventsRef-        let startVersion = fromIntegral (length existing)+        let events = map (.payload) taggedEvents+            startVersion = fromIntegral (length existing)             versioned = zipWith (\i e -> StreamEvent uuid i (emptyMetadata "") e) [startVersion ..] events         modifyIORef eventsRef (++ versioned)         pure (Right (startVersion + fromIntegral (length events) - 1))       reader = EventStoreReader $ \query -> do         allEvts <- readIORef eventsRef         pure $ filterByQuery query allEvts-  pure (writer, reader)+  pure (taggedWriter, reader)   where     filterByQuery (QueryRange uuid _ _) =       filter (\(StreamEvent k _ _ _) -> k == uuid)@@ -57,30 +63,30 @@ spec = describe "CommandDispatcher" $ do   describe "commandHandlerDispatcher" $ do     it "routes command to matching handler and reports success" $ do-      (writer, reader) <- mkTestStore+      (taggedWriter, reader) <- mkTestStore        let handlers = [mkAggregateHandler counterHandler]-          dispatcher = commandHandlerDispatcher writer reader handlers+          dispatcher = commandHandlerDispatcher testCodec taggedWriter reader handlers -      result <- dispatcher.dispatchCommand (uuidFromInteger 1) Increment+      result <- dispatcher.dispatchCommand (uuidFromInteger 1) Increment id       result `shouldBe` CommandSucceeded      it "reports failure when command is rejected" $ do-      (writer, reader) <- mkTestStore+      (taggedWriter, reader) <- mkTestStore        let handlers = [mkAggregateHandler counterHandler]-          dispatcher = commandHandlerDispatcher writer reader handlers+          dispatcher = commandHandlerDispatcher testCodec taggedWriter reader handlers        -- Counter starts at 0, Decrement should fail-      result <- dispatcher.dispatchCommand (uuidFromInteger 1) Decrement+      result <- dispatcher.dispatchCommand (uuidFromInteger 1) Decrement id       result `shouldBe` CommandFailed (RejectionReason "AlreadyZero")      it "returns CommandSucceeded when no handler matches" $ do-      (writer, reader) <- mkTestStore+      (taggedWriter, reader) <- mkTestStore        let handlers = [mkAggregateHandler counterHandler]-          dispatcher = commandHandlerDispatcher writer reader handlers+          dispatcher = commandHandlerDispatcher testCodec taggedWriter reader handlers        -- Unknown returns Right [], so no handler "matches" (produces events)-      result <- dispatcher.dispatchCommand (uuidFromInteger 1) Unknown+      result <- dispatcher.dispatchCommand (uuidFromInteger 1) Unknown id       result `shouldBe` CommandSucceeded