packages feed

eventium-core 0.5.2 → 0.6.0

raw patch · 12 files changed

+394/−16 lines, 12 files

Files

eventium-core.cabal view
@@ -5,7 +5,7 @@ -- see: https://github.com/sol/hpack  name:           eventium-core-version:        0.5.2+version:        0.6.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,@@ -53,7 +53,9 @@       Eventium.SchemaEvolution.Types       Eventium.Store.Class       Eventium.Store.Queries+      Eventium.Store.Telemetry       Eventium.Store.Types+      Eventium.Telemetry       Eventium.TH       Eventium.TH.Projection       Eventium.TH.SumType@@ -95,7 +97,11 @@       Eventium.JsonSpec       Eventium.ProjectionSpec       Eventium.SchemaEvolutionSpec+      Eventium.Store.MetadataSpec+      Eventium.Store.MetadataTagSpec+      Eventium.Store.TelemetrySpec       Eventium.SumTypeSpec+      Eventium.TelemetrySpec       Eventium       Eventium.Codec       Eventium.CommandDispatcher@@ -115,7 +121,9 @@       Eventium.SchemaEvolution.Types       Eventium.Store.Class       Eventium.Store.Queries+      Eventium.Store.Telemetry       Eventium.Store.Types+      Eventium.Telemetry       Eventium.TH       Eventium.TH.Projection       Eventium.TH.SumType
src/Eventium.hs view
@@ -15,5 +15,7 @@ import Eventium.ReadModel as X import Eventium.SchemaEvolution as X import Eventium.Store.Class as X+import Eventium.Store.Telemetry as X+import Eventium.Telemetry as X import Eventium.TypeEmbedding as X import Eventium.UUID as X
src/Eventium/CommandDispatcher.hs view
@@ -11,6 +11,7 @@     mkAggregateHandler,     mkAggregateHandlerWith,     commandHandlerDispatcher,+    commandHandlerDispatcherWithTag,   ) where @@ -20,8 +21,8 @@ import Eventium.Codec (Codec) import Eventium.CommandHandler (CommandHandler, CommandHandlerError (..), applyCommandHandler) import Eventium.ProcessManager (CommandDispatchResult (..), CommandDispatcher (..), RejectionReason (..))-import Eventium.Store.Class (EventStoreWriter, VersionedEventStoreReader, metadataEnrichingEventStoreWriterWithEnricher)-import Eventium.Store.Types (EventVersion, TaggedEvent)+import Eventium.Store.Class (EventStoreWriter, VersionedEventStoreReader, metadataEnrichingEventStoreWriterWithTag)+import Eventium.Store.Types (EventTypeName, EventVersion, TaggedEvent, eventTypeNameOf) import Eventium.UUID (UUID)  -- | An embedded command handler paired with an error formatter.@@ -61,6 +62,11 @@ --   * @Right []@ — command did not match this handler → try next -- -- If no handler matches (all return @Right []@), reports 'CommandSucceeded' (no-op).+--+-- Tags each emitted event's 'EventMetadata.eventType' via 'Typeable'. If the+-- event type is an application-wide sum (e.g. @AccountingEvent@) whose+-- Typeable name isn't the useful discriminator, use+-- 'commandHandlerDispatcherWithTag' instead. commandHandlerDispatcher ::   (MonadIO m, Typeable event) =>   Codec event encoded ->@@ -68,9 +74,26 @@   VersionedEventStoreReader m event ->   [AggregateHandler event command] ->   CommandDispatcher m command-commandHandlerDispatcher codec taggedWriter reader handlers =+commandHandlerDispatcher = commandHandlerDispatcherWithTag eventTypeNameOf++-- | Like 'commandHandlerDispatcher' but the caller supplies the+-- 'EventTypeName' per event (instead of deriving it from 'Typeable'). Use+-- when the event is a wrapper sum whose Typeable name isn't the useful+-- discriminator — e.g. an application-wide event sum type such as+-- @AccountingEvent@, where every value shares the same Typeable name+-- regardless of which case it wraps. This matters for saga/process-manager+-- emitted events, which are routed through this dispatcher.+commandHandlerDispatcherWithTag ::+  (MonadIO m) =>+  (event -> EventTypeName) ->+  Codec event encoded ->+  EventStoreWriter UUID EventVersion m (TaggedEvent encoded) ->+  VersionedEventStoreReader m event ->+  [AggregateHandler event command] ->+  CommandDispatcher m command+commandHandlerDispatcherWithTag tagOf codec taggedWriter reader handlers =   CommandDispatcher $ \uuid cmd enricher ->-    let writer = metadataEnrichingEventStoreWriterWithEnricher enricher codec taggedWriter+    let writer = metadataEnrichingEventStoreWriterWithTag tagOf enricher codec taggedWriter      in go handlers writer uuid cmd   where     go [] _ _ _ = pure CommandSucceeded
src/Eventium/Store/Class.hs view
@@ -20,6 +20,7 @@     codecEventStoreWriter,     metadataEnrichingEventStoreWriter,     metadataEnrichingEventStoreWriterWithEnricher,+    metadataEnrichingEventStoreWriterWithTag,     tagEvents,      -- * Type embedding@@ -192,14 +193,29 @@   Codec event encoded ->   EventStoreWriter key position m (TaggedEvent encoded) ->   EventStoreWriter key position m event-metadataEnrichingEventStoreWriterWithEnricher enricher codec (EventStoreWriter write) =+metadataEnrichingEventStoreWriterWithEnricher = metadataEnrichingEventStoreWriterWithTag eventTypeNameOf++-- | Like 'metadataEnrichingEventStoreWriterWithEnricher' but the caller+-- supplies the 'EventTypeName' per event (instead of deriving it from+-- 'Typeable'). Use when the event is a wrapper sum whose Typeable name+-- isn't the useful discriminator — e.g. an application-wide event sum+-- type such as @AccountingEvent@, where every value shares the same+-- Typeable name regardless of which case it wraps.+metadataEnrichingEventStoreWriterWithTag ::+  (MonadIO m) =>+  (event -> EventTypeName) ->+  MetadataEnricher ->+  Codec event encoded ->+  EventStoreWriter key position m (TaggedEvent encoded) ->+  EventStoreWriter key position m event+metadataEnrichingEventStoreWriterWithTag tagOf enricher codec (EventStoreWriter write) =   EventStoreWriter $ \key pos events -> do     now <- liftIO getCurrentTime     let tagged =           map             ( \e ->                 TaggedEvent-                  (enricher (EventMetadata (eventTypeNameOf e) Nothing Nothing (Just now)))+                  (enricher (EventMetadata (tagOf e) Nothing Nothing (Just now) mempty))                   (codec.encode e)             )             events@@ -216,7 +232,7 @@ tagEvents codec now =   map $ \e ->     TaggedEvent-      (EventMetadata (eventTypeNameOf e) Nothing Nothing (Just now))+      (EventMetadata (eventTypeNameOf e) Nothing Nothing (Just now) mempty)       (codec.encode e)  -- | Like 'codecEventStoreWriter' but uses a 'TypeEmbedding' instead of
+ src/Eventium/Store/Telemetry.hs view
@@ -0,0 +1,34 @@+-- | Write-path telemetry: a decorator that emits a 'Signal' on each versioned+-- write outcome. Specialized to the versioned (aggregate) path so the stream+-- key ('UUID') and positions ('EventVersion') are concrete.+module Eventium.Store.Telemetry+  ( telemetryEventStoreWriter,+  )+where++import Eventium.Store.Class (EventStoreWriter (..), VersionedEventStoreWriter)+import Eventium.Store.Types+  ( EventWriteError (..),+    TaggedEvent (..),+  )+import Eventium.Telemetry++-- | Wrap a versioned 'TaggedEvent' writer so it emits 'EventsPersisted' on a+-- successful write and 'WriteConflict' on an optimistic-concurrency failure.+-- An empty batch emits nothing. A store-level exception is not reported (the+-- decorator does not bracket). The stream 'UUID' is carried through as-is.+telemetryEventStoreWriter ::+  (Monad m) =>+  Telemetry m ->+  VersionedEventStoreWriter m (TaggedEvent encoded) ->+  VersionedEventStoreWriter m (TaggedEvent encoded)+telemetryEventStoreWriter telemetry (EventStoreWriter write) =+  EventStoreWriter $ \key expectedPos events -> do+    result <- write key expectedPos events+    case events of+      [] -> pure ()+      _ -> case result of+        Right wr -> telemetry.emit (EventsPersisted key (map (.metadata) events) wr)+        Left (EventStreamNotAtExpectedVersion actualPos) ->+          telemetry.emit (WriteConflict key (ConflictInfo expectedPos actualPos))+    pure result
src/Eventium/Store/Types.hs view
@@ -1,7 +1,7 @@ {-# LANGUAGE AllowAmbiguousTypes #-}-{-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE DeriveTraversable #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TypeApplications #-} @@ -21,6 +21,7 @@     eventTypeNameOf,     EventMetadata (..),     emptyMetadata,+    insertCustomMetadata,     MetadataEnricher,     TaggedEvent (..), @@ -44,10 +45,11 @@  import Data.Aeson import qualified Data.List.NonEmpty as NE+import Data.Map.Strict (Map)+import qualified Data.Map.Strict as Map import Data.Text (Text, pack) import Data.Time (UTCTime) import Eventium.UUID-import GHC.Generics (Generic) import Type.Reflection (Typeable, typeRep) import Web.HttpApiData import Web.PathPieces@@ -74,20 +76,37 @@   { eventType :: !EventTypeName,     correlationId :: !(Maybe UUID),     causationId :: !(Maybe UUID),-    createdAt :: !(Maybe UTCTime)+    createdAt :: !(Maybe UTCTime),+    custom :: !(Map Text Text)   }-  deriving (Show, Eq, Generic)+  deriving (Show, Eq)  instance ToJSON EventMetadata where-  toJSON = genericToJSON defaultOptions-  toEncoding = genericToEncoding defaultOptions+  toJSON md =+    object $+      ["eventType" .= md.eventType]+        ++ maybe [] (\v -> ["correlationId" .= v]) md.correlationId+        ++ maybe [] (\v -> ["causationId" .= v]) md.causationId+        ++ maybe [] (\v -> ["createdAt" .= v]) md.createdAt+        ++ ["custom" .= md.custom | not (Map.null md.custom)]  instance FromJSON EventMetadata where-  parseJSON = genericParseJSON defaultOptions+  parseJSON = withObject "EventMetadata" $ \o ->+    EventMetadata+      <$> o .: "eventType"+      <*> o .:? "correlationId"+      <*> o .:? "causationId"+      <*> o .:? "createdAt"+      <*> o .:? "custom" .!= mempty  -- | 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 mempty++-- | Insert one key/value into an event's 'custom' context map.+-- @insertCustomMetadata "userId" uid@.+insertCustomMetadata :: Text -> Text -> EventMetadata -> EventMetadata+insertCustomMetadata k v md = md {custom = Map.insert k v md.custom}  -- | Builder function for customizing event metadata. --
+ src/Eventium/Telemetry.hs view
@@ -0,0 +1,54 @@+-- | A generic, framework-free, structured telemetry sink. The host app+-- supplies one interpreter ('Telemetry'); eventium emits typed 'Signal's+-- through it. No logging-framework dependency — a contravariant-style sink over+-- a domain signal type.+module Eventium.Telemetry+  ( Telemetry (..),+    Signal (..),+    ConflictInfo (..),+    silentTelemetry,+  )+where++import Eventium.Store.Types+  ( EventMetadata,+    EventVersion,+    EventWriteResult,+    ExpectedPosition,+  )+import Eventium.UUID (UUID)++-- | A structured telemetry sink. @emit@ runs in the caller's monad @m@.+-- Interpreters MUST NOT throw: a write-path emit may run inside the write+-- transaction, so a throwing interpreter could roll back a committed write.+newtype Telemetry m = Telemetry {emit :: Signal -> m ()}++-- | Everything eventium can report, across all subsystems. One growing closed+-- sum type. This slice introduces only the write-path constructors.+--+-- Write-path constructors carry the versioned-stream key as a 'UUID' (the+-- aggregate id), eventium's own stream identifier — interpreters render it as+-- they see fit.+data Signal+  = -- | Events durably written on the versioned (aggregate) write path:+    -- stream 'UUID', per-event metadata (each carries @eventType@,+    -- @correlationId@, @custom@), and assigned per-stream versions + global+    -- positions.+    EventsPersisted !UUID ![EventMetadata] !EventWriteResult+  | -- | An expected-position (optimistic concurrency) check failed; nothing+    -- was written.+    WriteConflict !UUID !ConflictInfo+  deriving (Show, Eq)++-- | Optimistic-concurrency conflict detail. @expected@ is the caller's asserted+-- position; @actual@ is the stream's real end version.+data ConflictInfo = ConflictInfo+  { expected :: !(ExpectedPosition EventVersion),+    actual :: !EventVersion+  }+  deriving (Show, Eq)++-- | No-op sink — the default everywhere; guarantees silent, zero-cost behaviour+-- unless the app opts in.+silentTelemetry :: (Applicative m) => Telemetry m+silentTelemetry = Telemetry (const (pure ()))
tests/Eventium/CommandDispatcherSpec.hs view
@@ -60,6 +60,34 @@     filterByQuery (QueryRange uuid _ _) =       filter (\(StreamEvent k _ _ _) -> k == uuid) +-- | Like 'mkTestStore', but also captures every 'TaggedEvent' passed to the+-- writer (metadata included) so tests can assert on the tag actually used.+mkCapturingTestStore ::+  IO+    ( IORef [TaggedEvent CounterEvent],+      VersionedEventStoreWriter IO (TaggedEvent CounterEvent),+      VersionedEventStoreReader IO CounterEvent+    )+mkCapturingTestStore = do+  eventsRef <- newIORef ([] :: [VersionedStreamEvent CounterEvent])+  capturedRef <- newIORef ([] :: [TaggedEvent CounterEvent])+  let taggedWriter = EventStoreWriter $ \uuid _expected taggedEvents -> do+        modifyIORef capturedRef (++ taggedEvents)+        existing <- readIORef eventsRef+        let events = map (.payload) taggedEvents+            startVersion = fromIntegral (length existing)+            versioned = zipWith (\i e -> StreamEvent uuid i (emptyMetadata "") e) [startVersion ..] events+            poss = take (length events) [SequenceNumber (length existing + 1) ..]+        modifyIORef eventsRef (++ versioned)+        pure (Right (zip [startVersion ..] poss))+      reader = EventStoreReader $ \query -> do+        allEvts <- readIORef eventsRef+        pure $ filterByQuery query allEvts+  pure (capturedRef, taggedWriter, reader)+  where+    filterByQuery (QueryRange uuid _ _) =+      filter (\(StreamEvent k _ _ _) -> k == uuid)+ spec :: Spec spec = describe "CommandDispatcher" $ do   describe "commandHandlerDispatcher" $ do@@ -91,3 +119,27 @@       -- Unknown returns Right [], so no handler "matches" (produces events)       result <- dispatcher.dispatchCommand (uuidFromInteger 1) Unknown id       result `shouldBe` CommandSucceeded++    it "tags emitted events with the Typeable event type name" $ do+      (capturedRef, taggedWriter, reader) <- mkCapturingTestStore++      let handlers = [mkAggregateHandler counterHandler]+          dispatcher = commandHandlerDispatcher testCodec taggedWriter reader handlers++      _ <- dispatcher.dispatchCommand (uuidFromInteger 1) Increment id+      captured <- readIORef capturedRef+      map (.metadata.eventType) captured `shouldBe` ["CounterEvent"]++  describe "commandHandlerDispatcherWithTag" $ do+    it "tags emitted events with the caller-supplied event type name" $ do+      (capturedRef, taggedWriter, reader) <- mkCapturingTestStore++      let handlers = [mkAggregateHandler counterHandler]+          dispatcher =+            commandHandlerDispatcherWithTag (const "SpecificTag") testCodec taggedWriter reader handlers++      result <- dispatcher.dispatchCommand (uuidFromInteger 1) Increment id+      result `shouldBe` CommandSucceeded++      captured <- readIORef capturedRef+      map (.metadata.eventType) captured `shouldBe` ["SpecificTag"]
+ tests/Eventium/Store/MetadataSpec.hs view
@@ -0,0 +1,41 @@+{-# LANGUAGE OverloadedStrings #-}++module Eventium.Store.MetadataSpec (spec) where++import Data.Aeson (Value (Null, Object), decode, encode, object, (.=))+import qualified Data.Aeson.KeyMap as KM+import Eventium.Store.Types (EventMetadata, emptyMetadata, insertCustomMetadata)+import Test.Hspec++spec :: Spec+spec = describe "EventMetadata JSON" $ do+  it "round-trips a non-empty custom map and includes the key" $ do+    let md = insertCustomMetadata "userId" "u-1" (emptyMetadata "Foo")+    decode (encode md) `shouldBe` Just md+    case decode (encode md) :: Maybe Value of+      Just (Object o) -> KM.member "custom" o `shouldBe` True+      _ -> expectationFailure "expected object"++  it "omits custom when empty" $ do+    case decode (encode (emptyMetadata "Foo")) :: Maybe Value of+      Just (Object o) -> KM.member "custom" o `shouldBe` False+      _ -> expectationFailure "expected object"++  it "omits Nothing Maybe fields (no explicit null)" $ do+    case decode (encode (emptyMetadata "Foo")) :: Maybe Value of+      Just (Object o) -> KM.member "correlationId" o `shouldBe` False+      _ -> expectationFailure "expected object"++  it "decodes a legacy row with explicit nulls and no custom key" $ do+    let legacy =+          object+            [ "eventType" .= ("Foo" :: String),+              "correlationId" .= Null,+              "causationId" .= Null,+              "createdAt" .= Null+            ]+    (decode (encode legacy) :: Maybe EventMetadata) `shouldBe` Just (emptyMetadata "Foo")++  it "decodes a new row that omits the optional keys to the same value" $ do+    let new = object ["eventType" .= ("Foo" :: String)]+    (decode (encode new) :: Maybe EventMetadata) `shouldBe` Just (emptyMetadata "Foo")
+ tests/Eventium/Store/MetadataTagSpec.hs view
@@ -0,0 +1,46 @@+{-# LANGUAGE OverloadedStrings #-}++module Eventium.Store.MetadataTagSpec (spec) where++import Data.IORef+import Data.Text (Text)+import Eventium.Codec (idCodec)+import Eventium.Store.Class+  ( EventStoreWriter (..),+    metadataEnrichingEventStoreWriterWithEnricher,+    metadataEnrichingEventStoreWriterWithTag,+  )+import Eventium.Store.Types+  ( EventMetadata (..),+    ExpectedPosition (..),+    TaggedEvent (..),+  )+import qualified Eventium.UUID as UUID+import Test.Hspec++capturing :: IO (IORef [TaggedEvent Text], EventStoreWriter UUID.UUID Int IO (TaggedEvent Text))+capturing = do+  ref <- newIORef []+  let writer = EventStoreWriter $ \_ _ events -> do+        modifyIORef' ref (++ events)+        pure (Right [])+  pure (ref, writer)++key :: UUID.UUID+key = UUID.nil++spec :: Spec+spec = describe "metadata event-type tagging" $ do+  it "metadataEnrichingEventStoreWriterWithTag carries the caller-supplied tag" $ do+    (ref, inner) <- capturing+    let wr = metadataEnrichingEventStoreWriterWithTag (const "SpecificTag") id idCodec inner+    _ <- wr.storeEvents key AnyPosition ["payload" :: Text]+    [tagged] <- readIORef ref+    tagged.metadata.eventType `shouldBe` ("SpecificTag" :: Text)++  it "metadataEnrichingEventStoreWriterWithEnricher still uses the Typeable name (unchanged)" $ do+    (ref, inner) <- capturing+    let wr = metadataEnrichingEventStoreWriterWithEnricher id idCodec inner+    _ <- wr.storeEvents key AnyPosition ["payload" :: Text]+    [tagged] <- readIORef ref+    tagged.metadata.eventType `shouldBe` ("Text" :: Text)
+ tests/Eventium/Store/TelemetrySpec.hs view
@@ -0,0 +1,61 @@+{-# LANGUAGE OverloadedStrings #-}++module Eventium.Store.TelemetrySpec (spec) where++import Data.IORef+import Data.Text (Text)+import Eventium.Store.Class (EventStoreWriter (..))+import Eventium.Store.Telemetry (telemetryEventStoreWriter)+import Eventium.Store.Types+  ( EventVersion (..),+    EventWriteError (..),+    ExpectedPosition (..),+    SequenceNumber (..),+    TaggedEvent (..),+    emptyMetadata,+  )+import Eventium.Telemetry+import qualified Eventium.UUID as UUID+import Test.Hspec++capturing :: IO (IORef [Signal], Telemetry IO)+capturing = do+  ref <- newIORef []+  pure (ref, Telemetry (\s -> modifyIORef' ref (++ [s])))++ev :: TaggedEvent Text+ev = TaggedEvent (emptyMetadata "Foo") "payload"++key :: UUID.UUID+key = UUID.nil++spec :: Spec+spec = describe "telemetryEventStoreWriter" $ do+  it "emits EventsPersisted on a successful write" $ do+    (ref, t) <- capturing+    let wr = [(EventVersion 1, SequenceNumber 10)]+    let inner = EventStoreWriter (\_ _ _ -> pure (Right wr))+    _ <- (telemetryEventStoreWriter t inner).storeEvents key AnyPosition [ev]+    signals <- readIORef ref+    signals `shouldBe` [EventsPersisted key [emptyMetadata "Foo"] wr]++  it "emits WriteConflict on an expected-position failure" $ do+    (ref, t) <- capturing+    let inner = EventStoreWriter (\_ _ _ -> pure (Left (EventStreamNotAtExpectedVersion (EventVersion 7))))+    _ <- (telemetryEventStoreWriter t inner).storeEvents key (ExactPosition (EventVersion 3)) [ev]+    signals <- readIORef ref+    signals `shouldBe` [WriteConflict key (ConflictInfo (ExactPosition (EventVersion 3)) (EventVersion 7))]++  it "emits nothing for an empty batch (even on a Left)" $ do+    (ref, t) <- capturing+    let inner = EventStoreWriter (\_ _ _ -> pure (Left (EventStreamNotAtExpectedVersion (EventVersion 7))))+    _ <- (telemetryEventStoreWriter t inner).storeEvents key (ExactPosition (EventVersion 3)) ([] :: [TaggedEvent Text])+    readIORef ref `shouldReturn` []++  it "silentTelemetry emits nothing and passes the result through" $ do+    (ref, _) <- capturing+    let wr = [(EventVersion 1, SequenceNumber 10)]+    let inner = EventStoreWriter (\_ _ _ -> pure (Right wr))+    r <- (telemetryEventStoreWriter silentTelemetry inner).storeEvents key AnyPosition [ev]+    r `shouldBe` Right wr+    readIORef ref `shouldReturn` []
+ tests/Eventium/TelemetrySpec.hs view
@@ -0,0 +1,22 @@+{-# LANGUAGE OverloadedStrings #-}++module Eventium.TelemetrySpec (spec) where++import Data.IORef+import Eventium.Store.Types (emptyMetadata)+import Eventium.Telemetry+import qualified Eventium.UUID as UUID+import Test.Hspec++spec :: Spec+spec = describe "Telemetry" $ do+  it "silentTelemetry emits nothing (runs cleanly, returns unit)" $ do+    let t = silentTelemetry :: Telemetry IO+    t.emit (EventsPersisted UUID.nil [emptyMetadata "Foo"] []) `shouldReturn` ()++  it "a capturing sink records the signal it is given" $ do+    ref <- newIORef []+    let t = Telemetry (\sig -> modifyIORef' ref (sig :)) :: Telemetry IO+    let sig = EventsPersisted UUID.nil [emptyMetadata "Foo"] []+    t.emit sig+    readIORef ref `shouldReturn` [sig]