eventium-core 0.1.0 → 0.2.1

raw patch · 37 files changed

+2683/−878 lines, 37 filesdep +timedep −x-sum-type-boilerplatedep ~template-haskell

Dependencies added: time

Dependencies removed: x-sum-type-boilerplate

Dependency ranges changed: template-haskell

Files

• CHANGELOG.md +55/−0
• README.md +154/−53
• eventium-core.cabal +66/−26
• src/Eventium.hs +8/−4
• src/Eventium/Codec.hs +211/−0
• src/Eventium/CommandDispatcher.hs +72/−0
• src/Eventium/CommandHandler.hs +111/−48
• src/Eventium/EventBus.hs +0/−45
• src/Eventium/EventHandler.hs +89/−0
• src/Eventium/EventPublisher.hs +84/−0
• src/Eventium/EventSubscription.hs +237/−0
• src/Eventium/Json.hs +42/−0
• src/Eventium/Json/TH.hs +20/−0
• src/Eventium/ProcessManager.hs +129/−44
• src/Eventium/Projection.hs +62/−33
• src/Eventium/ProjectionCache/Cache.hs +140/−0
• src/Eventium/ProjectionCache/Types.hs +15/−115
• src/Eventium/ReadModel.hs +111/−0
• src/Eventium/ReadModel/Class.hs +0/−48
• src/Eventium/Serializer.hs +0/−204
• src/Eventium/Store/Class.hs +114/−109
• src/Eventium/Store/Queries.hs +3/−3
• src/Eventium/Store/Types.hs +127/−0
• src/Eventium/TH.hs +2/−1
• src/Eventium/TH/Projection.hs +1/−1
• src/Eventium/TH/SumType.hs +165/−0
• src/Eventium/TH/SumTypeCodec.hs +122/−0
• src/Eventium/TH/SumTypeSerializer.hs +0/−86
• src/Eventium/TypeEmbedding.hs +49/−0
• src/Eventium/UUID.hs +2/−0
• tests/Eventium/CodecSpec.hs +81/−0
• tests/Eventium/CommandDispatcherSpec.hs +86/−0
• tests/Eventium/EventHandlerSpec.hs +100/−0
• tests/Eventium/JsonSpec.hs +69/−0
• tests/Eventium/ProjectionSpec.hs +72/−0
• tests/Eventium/SerializerSpec.hs +0/−58
• tests/Eventium/SumTypeSpec.hs +84/−0

@@ -0,0 +1,55 @@+# eventium-core Changelog++## 0.2.1 (Unreleased)++### Additions++- `snapshotEventHandler` -- `EventHandler` that auto-updates a `VersionedProjectionCache` on each event.+- `snapshotGlobalEventHandler` -- same for `GlobalProjectionCache`.+- `applyCommandHandlerWithCache` -- like `applyCommandHandler` but loads from cache and updates after write.+- `Eventium.ReadModel` module:+  - `ReadModel` record type for queryable persistent views driven by the global event stream.+  - `runReadModel`, `rebuildReadModel`, `combineReadModels` combinators.++## 0.2.0 (Unreleased)++See the [root changelog](../CHANGELOG.md) for full details.++### Breaking changes++- Record field prefixes removed from all core types. Fields accessed via+  `OverloadedRecordDot` (e.g. `projection.seed`, `event.metadata`).+  `NoFieldSelectors` and `DuplicateRecordFields` enabled as default extensions.+- `StreamEvent` now has 4 fields (added `EventMetadata`).+- `CommandHandler` gained an `err` type parameter; `commandHandlerHandler` renamed to `decide`.+- `ProcessManager` is now pure: `react` returns `[ProcessManagerEffect]`. `ProcessManagerEffect` only supports `IssueCommand` (removed `EmitEvent`).+- `EventPublisher`: removed `synchronousEventBusWrapper`; added `publishingEventStoreWriter` and `synchronousPublisher`.+- `PollingPeriodSeconds` (Double) replaced by `PollingIntervalMillis` (Int).+- Codec wrapper argument order changed (codec comes first).+- Removed `Eventium.ReadModel.Memory`.++### Internal++- Absorbed `x-sum-type-boilerplate` into `Eventium.TH.SumType`.+- Dropped `persistent-template` dependency.++### Additions++- `Eventium.TH.SumType` module: `constructSumType`, `sumTypeConverter`,+  `partialSumTypeConverter` (previously from `x-sum-type-boilerplate`).+- `NoFieldSelectors`, `DuplicateRecordFields`, `OverloadedRecordDot` default extensions.+- `EventMetadata`, `emptyMetadata`.+- `lenientCodecEventStoreReader`, `lenientCodecProjection`.+- `runProjectionSubscription`, `eventHandlerMapMaybe`.+- `CommandDispatcher` newtype with `mkCommandDispatcher` and `fireAndForgetDispatcher`.+- `CommandDispatchResult` (`CommandSucceeded` | `CommandFailed Text`).+- `IssueCommandWithCompensation` effect for saga compensation workflows.+- `processManagerEventHandler` for wiring a `ProcessManager` to an `EventHandler`.+- `Eventium.CommandDispatcher` module: `AggregateHandler`, `mkAggregateHandler`,+  `commandHandlerDispatcher` for list-based multi-aggregate command routing.+- `embeddedCommandHandler` returns `Right []` for non-matching commands+  (was `DecodeError` exception).++## 0.1.0++Initial release.

@@ -1,90 +1,191 @@ # Eventium Core -Core abstractions and utilities for building event sourcing systems in Haskell.+Core abstractions for building event-sourced applications in Haskell.  ## Overview -`eventium-core` is the foundational package of the Eventium event sourcing framework. It provides all the essential abstractions, interfaces, and utilities needed to build event-sourced applications with CQRS patterns.+`eventium-core` is the foundational package of the Eventium framework. It defines storage-agnostic interfaces and pure types -- no database drivers, no I/O beyond what the caller provides via monad parameters. -## Key Components+## Key Types -### Event Store (`Eventium.Store.Class`)-- **`EventStoreReader`** - Read events from streams (versioned and global)-- **`EventStoreWriter`** - Append events with optimistic concurrency control-- **`StreamEvent`** - Event metadata wrapper with stream keys and positions+### StreamEvent -### Projection (`Eventium.Projection`)-Pure event fold for rebuilding state from events. Used for both domain aggregates (write model) and read models (query side).+```haskell+data StreamEvent key position event = StreamEvent+  { key      :: key+  , position :: position+  , metadata :: EventMetadata+  , payload  :: event+  }+``` +Every stored event carries a stream key, a position, metadata (event type, correlation/causation IDs, timestamp), and the domain payload.++### EventStoreReader / EventStoreWriter+ ```haskell+newtype EventStoreReader key position m event = EventStoreReader+  { getEvents :: QueryRange key position -> m [event] }++newtype EventStoreWriter key position m event = EventStoreWriter+  { storeEvents :: key -> ExpectedPosition position -> [event]+                -> m (Either (EventWriteError position) EventVersion) }+```++Polymorphic over key, position, monad, and event types. `runEventStoreReaderUsing` / `runEventStoreWriterUsing` lift between monads. `codecEventStoreReader` / `codecEventStoreWriter` wrap with a `Codec`.++### Projection++```haskell data Projection state event = Projection-  { projectionSeed :: state-  , projectionEventHandler :: state -> event -> state+  { seed         :: state+  , eventHandler :: state -> event -> state   } ``` -### Command Handler (`Eventium.CommandHandler`)-Implements the aggregate pattern from DDD/Event Sourcing. Processes commands, validates against current state, and emits events.+Pure fold for rebuilding state from events. Used for both aggregates (write side) and read models (query side). `getLatestStreamProjection` loads events from a reader and applies them. +### CommandHandler+ ```haskell-data CommandHandler state event command = CommandHandler-  { commandHandlerHandler :: state -> command -> [event]-  , commandHandlerProjection :: Projection state event+data CommandHandler state event command err = CommandHandler+  { decide     :: state -> command -> Either err [event]+  , projection :: Projection state event   } ``` -### Process Manager (`Eventium.ProcessManager`)-Coordinates long-running business processes across multiple aggregates. Implements the Saga pattern for complex workflows.+Validates a command against current state, returning either a domain error or new events. `applyCommandHandler` orchestrates the full load-decide-write cycle. -### Read Model (`Eventium.ReadModel.Class`)-Builds denormalized views optimized for queries. Tracks processed events for eventual consistency with the write side.+### ProcessManager -### Serializer (`Eventium.Serializer`)-Type-safe event serialization/deserialization with JSON support and Template Haskell utilities for automatic boilerplate generation.+```haskell+data ProcessManager state event command = ProcessManager+  { projection :: Projection state (VersionedStreamEvent event)+  , react      :: state -> VersionedStreamEvent event+               -> [ProcessManagerEffect command]+  } -### Template Haskell Utilities (`Eventium.TH`)-- **`deriveJSON`** - Generate JSON instances-- **`deriveSumTypeSerializer`** - Generate serializers for sum types (event polymorphism)-- **`makeProjection`** - Generate projection boilerplate+data ProcessManagerEffect command+  = IssueCommand UUID command+  | IssueCommandWithCompensation UUID command (Text -> [ProcessManagerEffect command])+``` -## Features+Coordinates cross-aggregate workflows. `react` is pure; `runProcessManagerEffects` dispatches the resulting commands via a `CommandDispatcher`. `IssueCommandWithCompensation` triggers compensation effects when a command fails. -- ✅ Type-safe event sourcing abstractions-- ✅ Optimistic concurrency control with `ExpectedPosition`-- ✅ CQRS pattern support (command/query separation)-- ✅ Process Manager (Saga) pattern-- ✅ Projection caching for performance-- ✅ Template Haskell for reducing boilerplate-- ✅ Storage backend agnostic (memory, SQL, NoSQL)+### CommandDispatcher -## Usage+```haskell+newtype CommandDispatcher m command = CommandDispatcher+  { dispatchCommand :: UUID -> command -> m CommandDispatchResult } -Add `eventium-core` to your package dependencies:+data CommandDispatchResult = CommandSucceeded | CommandFailed Text+``` -```yaml-dependencies:-  - eventium-core+Routes commands to aggregates and reports outcomes. Construct with `mkCommandDispatcher`, or use `fireAndForgetDispatcher` for legacy callbacks. `commandHandlerDispatcher` (from `Eventium.CommandDispatcher`) builds a dispatcher from a list of `AggregateHandler`s for multi-aggregate routing.++`processManagerEventHandler` wires a `ProcessManager` to a global reader and dispatcher, producing a ready-to-use `EventHandler`.++### ProjectionCache++```haskell+data ProjectionCache key position encoded m = ProjectionCache+  { storeSnapshot :: key -> position -> encoded -> m ()+  , loadSnapshot  :: key -> m (Maybe (position, encoded))+  } ``` -Then choose a storage backend:-- `eventium-memory` - In-memory (development/testing)-- `eventium-sqlite` - SQLite (single-process apps)-- `eventium-postgresql` - PostgreSQL (production systems)+Snapshot store for aggregate state, avoiding full event replay on every load. Two flavors: -## Design Principles+- **VersionedProjectionCache** -- keyed by `UUID` + `EventVersion`, one snapshot per aggregate instance.+- **GlobalProjectionCache** -- keyed by `()` + `SequenceNumber`, singleton snapshot for global projections. -1. **Type Safety** - Phantom types prevent mixing concerns-2. **Functional Purity** - Projections are pure folds-3. **Abstraction** - Backend-agnostic core types-4. **CQRS** - Clear command/query separation-5. **Standard Patterns** - Aggregates, Sagas, Read Models+Wiring helpers: +- `snapshotEventHandler` -- `EventHandler` that auto-updates a `VersionedProjectionCache` on each event. Compose with `publishingEventStoreWriter` for transparent snapshotting.+- `snapshotGlobalEventHandler` -- same for `GlobalProjectionCache`.+- `applyCommandHandlerWithCache` -- like `applyCommandHandler` but loads from cache and updates after write.++### ReadModel++```haskell+data ReadModel m event = ReadModel+  { initialize     :: m ()+  , eventHandler   :: EventHandler m (GlobalStreamEvent event)+  , checkpointStore :: CheckpointStore m SequenceNumber+  , reset          :: m ()+  }+```++Abstraction for queryable persistent views driven by the global event stream. Users define their own schema and event handler; the library manages the event pipeline and checkpointing. ReadModels always consume the global stream (cross-aggregate views need total ordering).++- `runReadModel` -- polling subscription that keeps the view updated (runs forever).+- `rebuildReadModel` -- reset + replay all events (one-shot rebuild).+- `combineReadModels` -- fan-out events to multiple read models.++### EventHandler / EventPublisher / EventSubscription++- **EventHandler** -- composable event consumer (`Contravariant`, `Semigroup`, `Monoid`).+- **EventPublisher** -- decouples post-write dispatch. `publishingEventStoreWriter` wraps a writer for auto-publish.+- **EventSubscription** -- push-based delivery. `pollingSubscription` polls the global stream with a `CheckpointStore`.++### Codec++```haskell+data Codec a b = Codec+  { encode :: a -> b+  , decode :: b -> Maybe a+  }+```++Value-level bidirectional conversion. Composable via `composeCodecs`. `Eventium.TH` generates sum-type codecs and JSON instances.++### TypeEmbedding++```haskell+data TypeEmbedding a b = TypeEmbedding+  { embed   :: a -> b+  , extract :: b -> Maybe a+  }+```++Embeds one sum type into another (e.g. aggregate events into an application-wide event type). Separate from `Codec` to distinguish type-level subset relationships from wire-format encoding.++## Modules++| Module | Purpose |+|--------|---------|+| `Eventium.Store.Class` | Reader/Writer interfaces, monad lifting, codec wrappers |+| `Eventium.Store.Types` | `StreamEvent`, `EventMetadata`, `EventVersion`, `SequenceNumber`, `ExpectedPosition` |+| `Eventium.Store.Queries` | `QueryRange` builders (`allEvents`, `eventsStartingAt`, etc.) |+| `Eventium.Projection` | `Projection`, `StreamProjection`, `getLatestStreamProjection` |+| `Eventium.CommandHandler` | `CommandHandler`, `applyCommandHandler`, `applyCommandHandlerWithCache` |+| `Eventium.ProcessManager` | `ProcessManager`, `ProcessManagerEffect`, `CommandDispatcher`, `CommandDispatchResult`, `runProcessManagerEffects`, `processManagerEventHandler` |+| `Eventium.CommandDispatcher` | `AggregateHandler`, `mkAggregateHandler`, `commandHandlerDispatcher` |+| `Eventium.EventHandler` | `EventHandler`, `handleEvents` |+| `Eventium.EventPublisher` | `EventPublisher`, `publishingEventStoreWriter`, `synchronousPublisher` |+| `Eventium.EventSubscription` | `EventSubscription`, `pollingSubscription`, `CheckpointStore` |+| `Eventium.ReadModel` | `ReadModel`, `runReadModel`, `rebuildReadModel`, `combineReadModels` |+| `Eventium.ProjectionCache.Cache` | `ProjectionCache` helpers: `snapshotEventHandler`, `snapshotGlobalEventHandler`, `getLatestVersionedProjectionWithCache`, `updateVersionedProjectionCache` |+| `Eventium.ProjectionCache.Types` | `ProjectionCache`, `VersionedProjectionCache`, `GlobalProjectionCache` |+| `Eventium.Codec` | `Codec`, `jsonCodec`, `jsonTextCodec`, `composeCodecs` |+| `Eventium.UUID` | UUID utilities (`uuidNextRandom`, `uuidFromText`, `uuidFromInteger`) |+| `Eventium.TH` | Template Haskell: `deriveJSON`, `mkSumTypeCodec`, `mkSumTypeEmbedding`, `makeProjection` |++## Usage++```yaml+dependencies:+  - eventium-core >= 0.1.0+```++Then pick a storage backend: `eventium-memory`, `eventium-sqlite`, or `eventium-postgresql`.+ ## Documentation -- [Main README](../README.md) - Project overview-- [Design Documentation](../DESIGN.md) - Detailed architecture-- [Examples](../examples/) - Working applications+- [Main README](../README.md)+- [Design](../DESIGN.md)+- [Examples](../examples/)  ## License -MIT - see [LICENSE.md](LICENSE.md)+MIT -- see [LICENSE.md](LICENSE.md)

@@ -1,16 +1,16 @@ cabal-version: 1.12 --- This file has been generated from package.yaml by hpack version 0.37.0.+-- This file has been generated from package.yaml by hpack version 0.38.3. -- -- see: https://github.com/sol/hpack  name:           eventium-core-version:        0.1.0+version:        0.2.1 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, read models, process managers, and-                Template Haskell utilities for generating boilerplate code. This is the foundational package that all-                other Eventium modules build upon.+                It includes event store interfaces, command handlers, projections, event handlers, event publishers,+                event subscriptions, process managers, and Template Haskell utilities for generating boilerplate code.+                This is the foundational package that all other Eventium modules build upon. category:       Database,Eventsourcing stability:      experimental homepage:       https://github.com/aleks-sidorenko/eventium#readme@@ -27,65 +27,103 @@   type: git   location: https://github.com/aleks-sidorenko/eventium +flag ci+  description: Enable -Werror for CI builds+  manual: True+  default: False+ library   exposed-modules:       Eventium+      Eventium.Codec+      Eventium.CommandDispatcher       Eventium.CommandHandler-      Eventium.EventBus+      Eventium.EventHandler+      Eventium.EventPublisher+      Eventium.EventSubscription+      Eventium.Json+      Eventium.Json.TH       Eventium.ProcessManager       Eventium.Projection+      Eventium.ProjectionCache.Cache       Eventium.ProjectionCache.Types-      Eventium.ReadModel.Class-      Eventium.Serializer+      Eventium.ReadModel       Eventium.Store.Class       Eventium.Store.Queries+      Eventium.Store.Types       Eventium.TH       Eventium.TH.Projection-      Eventium.TH.SumTypeSerializer+      Eventium.TH.SumType+      Eventium.TH.SumTypeCodec+      Eventium.TypeEmbedding       Eventium.UUID   other-modules:       Paths_eventium_core   hs-source-dirs:       src-  ghc-options: -Wall+  default-extensions:+      NoFieldSelectors+      DuplicateRecordFields+      OverloadedRecordDot+  ghc-options: -Wall -Wcompat -Widentities -Wincomplete-record-updates -Wredundant-constraints -Wno-type-defaults -Wno-incomplete-uni-patterns   build-depends:       aeson >=1.5 && <2.3     , base >=4.9 && <5     , containers >=0.6 && <0.8-    , contravariant >=1.5 && <1.6+    , contravariant ==1.5.*     , http-api-data >=0.4 && <0.7-    , path-pieces >=0.2 && <0.3-    , template-haskell >=2.16 && <2.22+    , path-pieces ==0.2.*+    , template-haskell >=2.20 && <2.23     , text >=1.2 && <2.2+    , time >=1.9 && <1.14     , transformers >=0.5 && <0.7-    , uuid >=1.3 && <1.4-    , x-sum-type-boilerplate >=0.1 && <0.2+    , uuid ==1.3.*   default-language: Haskell2010+  if flag(ci)+    ghc-options: -Werror  test-suite spec   type: exitcode-stdio-1.0   main-is: Spec.hs   other-modules:-      Eventium.SerializerSpec+      Eventium.CodecSpec+      Eventium.CommandDispatcherSpec+      Eventium.EventHandlerSpec+      Eventium.JsonSpec+      Eventium.ProjectionSpec+      Eventium.SumTypeSpec       Eventium+      Eventium.Codec+      Eventium.CommandDispatcher       Eventium.CommandHandler-      Eventium.EventBus+      Eventium.EventHandler+      Eventium.EventPublisher+      Eventium.EventSubscription+      Eventium.Json+      Eventium.Json.TH       Eventium.ProcessManager       Eventium.Projection+      Eventium.ProjectionCache.Cache       Eventium.ProjectionCache.Types-      Eventium.ReadModel.Class-      Eventium.Serializer+      Eventium.ReadModel       Eventium.Store.Class       Eventium.Store.Queries+      Eventium.Store.Types       Eventium.TH       Eventium.TH.Projection-      Eventium.TH.SumTypeSerializer+      Eventium.TH.SumType+      Eventium.TH.SumTypeCodec+      Eventium.TypeEmbedding       Eventium.UUID       Paths_eventium_core   hs-source-dirs:       tests       src-  ghc-options: -Wall+  default-extensions:+      NoFieldSelectors+      DuplicateRecordFields+      OverloadedRecordDot+  ghc-options: -Wall -Wcompat -Widentities -Wincomplete-record-updates -Wredundant-constraints -Wno-type-defaults -Wno-incomplete-uni-patterns   build-tool-depends:       hspec-discover:hspec-discover   build-depends:@@ -93,13 +131,15 @@     , aeson >=1.5 && <2.3     , base >=4.9 && <5     , containers >=0.6 && <0.8-    , contravariant >=1.5 && <1.6+    , contravariant ==1.5.*     , hspec     , http-api-data >=0.4 && <0.7-    , path-pieces >=0.2 && <0.3-    , template-haskell >=2.16 && <2.22+    , path-pieces ==0.2.*+    , template-haskell >=2.20 && <2.23     , text >=1.2 && <2.2+    , time >=1.9 && <1.14     , transformers >=0.5 && <0.7-    , uuid >=1.3 && <1.4-    , x-sum-type-boilerplate >=0.1 && <0.2+    , uuid ==1.3.*   default-language: Haskell2010+  if flag(ci)+    ghc-options: -Werror

@@ -3,12 +3,16 @@   ) where +import Eventium.Codec as X+import Eventium.CommandDispatcher as X import Eventium.CommandHandler as X-import Eventium.EventBus as X+import Eventium.EventHandler as X+import Eventium.EventPublisher as X+import Eventium.EventSubscription as X import Eventium.ProcessManager as X import Eventium.Projection as X-import Eventium.ProjectionCache.Types as X-import Eventium.ReadModel.Class as X-import Eventium.Serializer as X+import Eventium.ProjectionCache.Cache as X+import Eventium.ReadModel as X import Eventium.Store.Class as X+import Eventium.TypeEmbedding as X import Eventium.UUID as X

@@ -0,0 +1,211 @@+{-# LANGUAGE DefaultSignatures #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE TypeOperators #-}++module Eventium.Codec+  ( -- * Class+    Codec (..),+    composeCodecs,++    -- * Common codecs+    idCodec,+    traverseCodec,+    jsonCodec,+    jsonTextCodec,+    dynamicCodec,++    -- * Sum types+    EventSumType (..),+    eventSumTypeCodec,++    -- * Exceptions+    DecodeError (..),+    EncodeError (..),+  )+where++import Control.Applicative ((<|>))+import Control.Exception (Exception, throw)+import Data.Aeson (FromJSON, Result (..), ToJSON, Value, fromJSON, toJSON)+import qualified Data.Aeson as Aeson+import Data.Dynamic+import Data.Maybe (fromMaybe)+import qualified Data.Text.Lazy as TL+import qualified Data.Text.Lazy.Encoding as TLE+import Data.Typeable (typeOf)+import GHC.Generics++-- | A 'Codec' describes the injective conversion between types @a@ and+-- @b@. In plain English, this means that you can go from @a@ to @b@, and you+-- can 'Maybe' go from @b@ back to @a@. This is often used to encode events+-- to an event store, and then decode them back.+data Codec a b = Codec+  { encode :: a -> b,+    decode :: b -> Maybe a+  }++-- | Exception thrown when a strict decode function encounters+-- data it cannot decode. Can be caught specifically using+-- @Control.Exception.catch@ or @Control.Exception.try@.+--+-- In production, prefer the lenient variants (e.g. 'lenientCodecProjection',+-- 'lenientCodecEventHandler', 'lenientCodecEventStoreReader') which+-- silently skip unrecognized events instead of throwing.+data DecodeError = DecodeError+  { context :: !String,+    message :: !String+  }+  deriving (Show, Eq)++instance Exception DecodeError++-- | Exception thrown when 'eventSumTypeCodec' cannot encode a value.+-- This indicates a programming error: the source sum type contains a+-- constructor not present in the target sum type.+data EncodeError = EncodeError+  { context :: !String,+    message :: !String+  }+  deriving (Show, Eq)++instance Exception EncodeError++-- | Apply an intermediate 'Codec' to a codec to go from type @a@ to+-- @c@ with @b@ in the middle. Note that with decoding, if the conversion+-- from @c@ to @b@ or from @b@ to @a@ fails, the whole decoding fails.+composeCodecs :: Codec a b -> Codec b c -> Codec a c+composeCodecs codec1 codec2 = Codec encode' decode'+  where+    encode' = codec2.encode . codec1.encode+    decode' x = codec2.decode x >>= codec1.decode++-- | Simple "codec" using 'id'. Useful for when an API requires a+-- codec but you don't need to actually change types.+idCodec :: Codec a a+idCodec = Codec id Just++-- | Uses 'Traversable' to wrap a 'Codec'.+traverseCodec ::+  (Traversable t) =>+  Codec a b ->+  Codec (t a) (t b)+traverseCodec codec =+  Codec encode' decode'+  where+    encode' = fmap codec.encode+    decode' = traverse codec.decode++-- | A 'Codec' for aeson 'Value's.+jsonCodec :: (ToJSON a, FromJSON a) => Codec a Value+jsonCodec =+  Codec+    { encode = toJSON,+      decode = \x ->+        case fromJSON x of+          Success a -> Just a+          Error _ -> Nothing+    }++-- | A 'Codec' to convert JSON to/from lazy text. Useful for Sql event+-- stores that store JSON values as text.+jsonTextCodec :: (ToJSON a, FromJSON a) => Codec a TL.Text+jsonTextCodec =+  Codec+    { encode = TLE.decodeUtf8 . Aeson.encode,+      decode = Aeson.decode . TLE.encodeUtf8+    }++-- | A 'Codec' for 'Dynamic' values using 'toDyn' and 'fromDynamic'.+dynamicCodec :: (Typeable a) => Codec a Dynamic+dynamicCodec = Codec toDyn fromDynamic++-- | A 'Codec' from one 'EventSumType' instance to another. WARNING: If+-- not all events in the source 'EventSumType' are in the @encoded@+-- 'EventSumType', then this function will be partial!+eventSumTypeCodec :: (Typeable a, EventSumType a, EventSumType b) => Codec a b+eventSumTypeCodec = Codec encode' decode'+  where+    encode' event =+      fromMaybe+        (throw $ EncodeError "eventSumTypeCodec" ("Can't encode event of type " ++ show (typeOf event)))+        (eventFromDyn $ eventToDyn event)+    decode' = eventFromDyn . eventToDyn++-- | This is a type class for encoding sum types of events to 'Dynamic'+-- without the associated constructor. This is useful when transforming between+-- two sum types of events. A common pattern is to put all the events in an+-- application in one big event sum type, and then have a smaller sum type for+-- each 'Projection'. Then, you can use 'eventSumTypeCodec' to transform+-- between the two.+--+-- It is meant to be derived with 'Generic'. For example:+--+-- @+--    data EventA = EventA deriving (Show)+--    data EventB = EventB deriving (Show)+--    data EventC = EventC deriving (Show)+--+--    data AllEvents+--      = AllEventsEventA EventA+--      | AllEventsEventB EventB+--      | AllEventsEventC EventC+--      deriving (Show, Generic)+--+--    instance EventSumType AllEvents+--+--    data MyEvents+--      = MyEventsEventA EventA+--      | MyEventsEventB EventB+--      deriving (Show, Generic)+--+--    instance EventSumType MyEvents+-- @+--+-- Now we can encode to 'Dynamic' without a constructor tag:+--+-- >>> eventToDyn (MyEventsEventA EventA)+-- <<EventA>>+--+-- We can also go from a 'MyEvents' value to an 'AllEvents' value:+--+-- >>> eventFromDyn (eventToDyn (MyEventsEventA EventA)) :: Maybe AllEvents+-- Just (AllEventsEventA EventA)+class EventSumType a where+  -- | Convert an event to a 'Dynamic' without the constructor tag+  eventToDyn :: a -> Dynamic++  -- | Go from a 'Dynamic' to an event with the constructor tag. Note, this+  -- function is @O(n)@ to the number of constructors.+  eventFromDyn :: Dynamic -> Maybe a++  default eventToDyn :: (Generic a, EventSumType' (Rep a)) => a -> Dynamic+  eventToDyn x = eventToDyn' (from x)++  default eventFromDyn :: (Generic a, EventSumType' (Rep a)) => Dynamic -> Maybe a+  eventFromDyn = fmap to . eventFromDyn'++-- Auxiliary type class for 'EventSumType' Generic fun+class EventSumType' f where+  eventToDyn' :: f p -> Dynamic+  eventFromDyn' :: Dynamic -> Maybe (f p)++-- M1 is the top-level metadata. We don't need the metadata so we just pass on+-- through.+instance (EventSumType' f) => EventSumType' (M1 i t f) where+  eventToDyn' (M1 x) = eventToDyn' x+  eventFromDyn' = fmap M1 . eventFromDyn'++-- The :+: operator is for when a type has multiple constructors. When+-- encoding, we just pass on through. When decoding, we try the first+-- constructor, and if that fails then the second.+instance (EventSumType' f, EventSumType' g) => EventSumType' (f :+: g) where+  eventToDyn' (L1 x) = eventToDyn' x+  eventToDyn' (R1 x) = eventToDyn' x+  eventFromDyn' dyn = (L1 <$> eventFromDyn' dyn) <|> (R1 <$> eventFromDyn' dyn)++-- K1 R represents an actual constructor. This is where we do the actual+-- conversion to/from 'Dynamic'.+instance (Typeable c) => EventSumType' (K1 R c) where+  eventToDyn' (K1 x) = toDyn x+  eventFromDyn' dyn = K1 <$> fromDynamic dyn

@@ -0,0 +1,72 @@+{-# LANGUAGE ExistentialQuantification #-}+{-# LANGUAGE OverloadedStrings #-}++-- | Provides list-based command routing for multi-aggregate systems.+--+-- Instead of manually trying each command handler in a nested cascade,+-- collect handlers into a list and let 'commandHandlerDispatcher' route+-- commands automatically.+module Eventium.CommandDispatcher+  ( AggregateHandler,+    mkAggregateHandler,+    mkAggregateHandlerWith,+    commandHandlerDispatcher,+  )+where++import qualified Data.Text as T+import Eventium.CommandHandler (CommandHandler, CommandHandlerError (..), applyCommandHandler)+import Eventium.ProcessManager (CommandDispatchResult (..), CommandDispatcher, RejectionReason (..), mkCommandDispatcher)+import Eventium.Store.Class (VersionedEventStoreReader, VersionedEventStoreWriter)++-- | An embedded command handler paired with an error formatter.+--+-- Erases the aggregate state type and error type so that multiple+-- aggregate handlers can be collected in a homogeneous list for routing.+data AggregateHandler event command+  = forall state err.+    AggregateHandler+      (CommandHandler state event command err)+      (err -> RejectionReason)++-- | Construct an 'AggregateHandler' using 'Show' to format errors.+mkAggregateHandler ::+  (Show err) =>+  CommandHandler state event command err ->+  AggregateHandler event command+mkAggregateHandler h = AggregateHandler h (RejectionReason . T.pack . show)++-- | Construct an 'AggregateHandler' with an explicit error formatter.+mkAggregateHandlerWith ::+  (err -> RejectionReason) ->+  CommandHandler state event command err ->+  AggregateHandler event command+mkAggregateHandlerWith fmt h = AggregateHandler h fmt++-- | Build a 'CommandDispatcher' from a list of 'AggregateHandler's.+--+-- Tries each handler in order:+--+--   * @Right (e:es)@ — command matched and produced events → 'CommandSucceeded'+--   * @Left (CommandRejected err)@ — command matched but was rejected → 'CommandFailed'+--   * @Left (ConcurrencyConflict _)@ — optimistic locking failure → 'CommandFailed'+--   * @Right []@ — command did not match this handler → try next+--+-- If no handler matches (all return @Right []@), reports 'CommandSucceeded' (no-op).+commandHandlerDispatcher ::+  (Monad m) =>+  VersionedEventStoreWriter m event ->+  VersionedEventStoreReader m event ->+  [AggregateHandler event command] ->+  CommandDispatcher m command+commandHandlerDispatcher writer reader handlers =+  mkCommandDispatcher $ \uuid cmd -> go handlers uuid cmd+  where+    go [] _ _ = pure CommandSucceeded+    go (AggregateHandler handler formatErr : rest) 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

@@ -1,75 +1,138 @@-{-# LANGUAGE RecordWildCards #-}- -- | Defines a Command Handler type.+--+-- A 'CommandHandler' is the combination of a 'Projection' (to reconstruct+-- aggregate state from events) and a pure decision function that validates+-- a command against the current state. The decision function returns+-- @'Either' err [event]@: 'Right' with new events on success, or 'Left'+-- with a domain error on rejection. module Eventium.CommandHandler   ( CommandHandler (..),-    allCommandHandlerStates,+    CommandHandlerError (..),     applyCommandHandler,-    serializedCommandHandler,+    applyCommandHandlerWithCache,+    codecCommandHandler,+    embeddedCommandHandler,   ) where -import Data.Foldable (foldl')-import Data.List (scanl')+import Control.Exception (throw)+import Eventium.Codec import Eventium.Projection-import Eventium.Serializer+import Eventium.ProjectionCache.Cache import Eventium.Store.Class+import Eventium.TypeEmbedding import Eventium.UUID --- | An 'CommandHandler' is a combination of a 'Projection' and a function to+-- | A 'CommandHandler' is a combination of a 'Projection' and a function to -- validate commands against that 'Projection'. When using a command handler in -- some service, it is common to simply load the latest projection state from -- the event store and handle the command. If the command is valid then the new -- events are applied to the projection in the event store.-data CommandHandler state event command+--+-- The @err@ type parameter represents the domain error type returned when a+-- command is rejected.+data CommandHandler state event command err   = CommandHandler-  { commandHandlerHandler :: state -> command -> [event],-    commandHandlerProjection :: Projection state event+  { decide :: state -> command -> Either err [event],+    projection :: Projection state event   } --- | Given a list commands, produce all of the states the command handler's--- projection sees. This is useful for unit testing a 'CommandHandler'.-allCommandHandlerStates ::-  CommandHandler state event command ->-  [command] ->-  [state]-allCommandHandlerStates (CommandHandler commandHandler (Projection seed eventHandler)) =-  scanl' go seed-  where-    go state command = foldl' eventHandler state $ commandHandler state command+-- | Errors that can occur when applying a command handler.+data CommandHandlerError err+  = -- | The command was rejected by the domain logic.+    CommandRejected err+  | -- | An optimistic concurrency conflict occurred when writing events.+    ConcurrencyConflict (EventWriteError EventVersion)+  deriving (Show, Eq) --- | Loads the latest version of a 'Projection' from the event store and tries to--- apply the 'CommandHandler' command to it. If the command succeeds, then this--- saves the events back to the store as well.+-- | Loads the latest version of a 'Projection' from the event store and tries+-- to apply the 'CommandHandler' command to it. If the command is accepted,+-- the events are saved back to the store.+--+-- Returns @'Left' ('CommandRejected' err)@ if the domain logic rejects the+-- command, or @'Left' ('ConcurrencyConflict' ...)@ if the event store+-- position has changed since reading. Returns @'Right' events@ on success. applyCommandHandler ::   (Monad m) =>   VersionedEventStoreWriter m event ->   VersionedEventStoreReader m event ->-  CommandHandler state event command ->+  CommandHandler state event command err ->   UUID ->   command ->-  m [event]-applyCommandHandler writer reader (CommandHandler handler proj) uuid command = do-  StreamProjection {..} <- getLatestStreamProjection reader (versionedStreamProjection uuid proj)-  let events = handler streamProjectionState command-  mError <- storeEvents writer uuid (ExactPosition streamProjectionPosition) events-  case mError of-    Left err -> error $ "TODO: Create CommandHandler restart logic. " ++ show err-    Right _ -> return events+  m (Either (CommandHandlerError err) [event])+applyCommandHandler writer reader cmdHandler uuid command = do+  sp <- getLatestStreamProjection reader (versionedStreamProjection uuid cmdHandler.projection)+  case cmdHandler.decide sp.state command of+    Left err -> return $ Left (CommandRejected err)+    Right events -> do+      result <- writer.storeEvents uuid (ExactPosition sp.position) events+      case result of+        Left writeErr -> return $ Left (ConcurrencyConflict writeErr)+        Right _ -> return $ Right events --- | Use a pair of 'Serializer's to wrap a 'CommandHandler' with event type @event@--- and command type @command@ so it uses the @serializedEvent@ and--- @serializedCommand@ types.-serializedCommandHandler ::-  CommandHandler state event command ->-  Serializer event serializedEvent ->-  Serializer command serializedCommand ->-  CommandHandler state serializedEvent serializedCommand-serializedCommandHandler (CommandHandler commandHandler projection) eventSerializer commandSerializer =-  CommandHandler serializedHandler serializedProjection'+-- | Like 'applyCommandHandler', but uses a 'VersionedProjectionCache' to load+-- the latest projection state (falling back to the event store when the cache+-- is stale or empty). After a successful write, the cache is updated with the+-- new projection state so subsequent calls can skip replaying events.+applyCommandHandlerWithCache ::+  (Monad m) =>+  VersionedEventStoreWriter m event ->+  VersionedEventStoreReader m event ->+  VersionedProjectionCache state m ->+  CommandHandler state event command err ->+  UUID ->+  command ->+  m (Either (CommandHandlerError err) [event])+applyCommandHandlerWithCache writer reader cache cmdHandler uuid command = do+  sp <- getLatestVersionedProjectionWithCache reader cache (versionedStreamProjection uuid cmdHandler.projection)+  case cmdHandler.decide sp.state command of+    Left err -> return $ Left (CommandRejected err)+    Right events -> do+      result <- writer.storeEvents uuid (ExactPosition sp.position) events+      case result of+        Left writeErr -> return $ Left (ConcurrencyConflict writeErr)+        Right endVersion -> do+          let newState = foldl' cmdHandler.projection.eventHandler sp.state events+          cache.storeSnapshot uuid endVersion newState+          return $ Right events++-- | Use a pair of 'Codec's to wrap a 'CommandHandler' with event type+-- @event@ and command type @command@ so it uses the @encodedEvent@ and+-- @encodedCommand@ types.+--+-- Throws 'DecodeError' if the command cannot be decoded.+codecCommandHandler ::+  Codec event encodedEvent ->+  Codec command encodedCommand ->+  CommandHandler state event command err ->+  CommandHandler state encodedEvent encodedCommand err+codecCommandHandler eventCodec commandCodec cmdHandler =+  CommandHandler codecHandler codecProjection'   where-    serializedProjection' = serializedProjection projection eventSerializer-    -- Try to deserialize the command and apply the handler. If we can't-    -- deserialize, then just return no events. We also need to serialize the-    -- events after of course.-    serializedHandler state = map (serialize eventSerializer) . maybe [] (commandHandler state) . deserialize commandSerializer+    codecProjection' = codecProjection eventCodec cmdHandler.projection+    codecHandler st encodedCmd =+      case commandCodec.decode encodedCmd of+        Nothing -> throw $ DecodeError "codecCommandHandler" "Failed to decode command"+        Just cmd -> fmap (map eventCodec.encode) (cmdHandler.decide st cmd)++-- | Like 'codecCommandHandler' but uses 'TypeEmbedding's instead of+-- 'Codec's. Intended for embedding aggregate-specific event\/command+-- types into application-wide sum types.+--+-- The projection uses lenient extraction (non-matching events are skipped).+-- Non-matching commands return @'Right' []@ (no events produced), which+-- enables safe multi-aggregate command dispatching — callers can try+-- multiple embedded handlers in sequence without catching exceptions.+embeddedCommandHandler ::+  TypeEmbedding event adaptedEvent ->+  TypeEmbedding command adaptedCommand ->+  CommandHandler state event command err ->+  CommandHandler state adaptedEvent adaptedCommand err+embeddedCommandHandler eventEmb commandEmb cmdHandler =+  CommandHandler embeddedHandler embeddedProjection'+  where+    embeddedProjection' = embeddedProjection eventEmb cmdHandler.projection+    embeddedHandler st adaptedCmd =+      case commandEmb.extract adaptedCmd of+        Nothing -> Right []+        Just cmd -> fmap (map eventEmb.embed) (cmdHandler.decide st cmd)

@@ -1,45 +0,0 @@-module Eventium.EventBus-  ( synchronousEventBusWrapper,-    storeAndPublishEvents,-  )-where--import Eventium.Store.Class-import Eventium.UUID---- | This function wraps an event store by sending events to event handlers--- after running 'storeEvents'. This is useful to quickly wire up event--- handlers in your application (like read models or process managers), and it--- is also useful for integration testing along with an in-memory event store.-synchronousEventBusWrapper ::-  (Monad m) =>-  VersionedEventStoreWriter m event ->-  [VersionedEventStoreWriter m event -> UUID -> event -> m ()] ->-  VersionedEventStoreWriter m event-synchronousEventBusWrapper writer handlers = wrappedStore-  where-    -- NB: We need to use recursive let bindings so we can put wrappedStore-    -- inside the event handlers-    handlers' = map ($ wrappedStore) handlers-    wrappedStore = EventStoreWriter $ storeAndPublishEvents writer handlers'---- | Stores events in the store and then publishes them to the event handlers.--- This is used in the 'synchronousEventBusWrapper'.-storeAndPublishEvents ::-  (Monad m) =>-  VersionedEventStoreWriter m event ->-  [UUID -> event -> m ()] ->-  UUID ->-  ExpectedPosition EventVersion ->-  [event] ->-  m (Either (EventWriteError EventVersion) EventVersion)-storeAndPublishEvents store handlers uuid expectedVersion events = do-  result <- storeEvents store uuid expectedVersion events-  case result of-    Left err -> return $ Left err-    Right vers -> do-      -- NB: If a handler stores events, then its events will be published-      -- before the events of the next handler. That is, we will be storing-      -- events generated by handlers in depth-first order.-      mapM_ (\handler -> mapM_ (handler uuid) events) handlers-      return $ Right vers

@@ -0,0 +1,89 @@+-- | Defines a universal event handler abstraction.+--+-- An 'EventHandler' is the simplest building block for consuming events.+-- It can be composed via 'Semigroup' (fan-out to multiple handlers),+-- adapted via 'Contravariant' (change the event type), and filtered via+-- 'eventHandlerMapMaybe'.+--+-- This abstraction is used by 'Eventium.EventPublisher.EventPublisher' for+-- synchronous dispatch and by 'Eventium.EventSubscription.EventSubscription'+-- for push-based delivery.+module Eventium.EventHandler+  ( EventHandler (..),+    handleEvents,+    eventHandlerMapMaybe,+    codecEventHandler,+    lenientCodecEventHandler,+    embeddedEventHandler,+  )+where++import Control.Exception (throw)+import Data.Foldable+import Data.Functor.Contravariant+import Eventium.Codec+import Eventium.TypeEmbedding++-- | An 'EventHandler' consumes events of type @event@ in some monad @m@.+--+-- Instances:+--+-- * 'Contravariant' — adapt the event type via @contramap@+-- * 'Semigroup' — fan-out: @h1 <> h2@ runs both handlers for each event+-- * 'Monoid' — @mempty@ is a no-op handler+newtype EventHandler m event = EventHandler {handleEvent :: event -> m ()}++instance Contravariant (EventHandler m) where+  contramap f (EventHandler h) = EventHandler (h . f)++instance (Applicative m) => Semigroup (EventHandler m event) where+  EventHandler h1 <> EventHandler h2 = EventHandler $ \e -> h1 e *> h2 e++instance (Applicative m) => Monoid (EventHandler m event) where+  mempty = EventHandler $ \_ -> pure ()++-- | Apply an 'EventHandler' to a list of events in order.+handleEvents :: (Monad m) => EventHandler m event -> [event] -> m ()+handleEvents (EventHandler h) = mapM_ h++-- | Filter events before they reach the handler. Events for which the+-- function returns 'Nothing' are silently dropped.+eventHandlerMapMaybe ::+  (Applicative m) =>+  (eventB -> Maybe eventA) ->+  EventHandler m eventA ->+  EventHandler m eventB+eventHandlerMapMaybe f (EventHandler h) = EventHandler $ \e -> for_ (f e) h++-- | Wrap an 'EventHandler' with a 'Codec' so it can consume events+-- of the encoded type. Throws 'DecodeError' if decoding fails.+-- Use 'lenientCodecEventHandler' to silently skip failures.+codecEventHandler ::+  Codec event encoded ->+  EventHandler m event ->+  EventHandler m encoded+codecEventHandler codec (EventHandler h) = EventHandler $ \e ->+  case codec.decode e of+    Just a -> h a+    Nothing -> throw $ DecodeError "codecEventHandler" "Failed to decode event"++-- | Like 'codecEventHandler' but silently drops events that fail to+-- decode.+--+-- Recommended for production use when using sum-type codecs, as it+-- allows event handlers to gracefully handle events they don't recognize.+lenientCodecEventHandler ::+  (Applicative m) =>+  Codec event encoded ->+  EventHandler m event ->+  EventHandler m encoded+lenientCodecEventHandler codec = eventHandlerMapMaybe codec.decode++-- | Adapt an 'EventHandler' using a 'TypeEmbedding'. Events that do not+-- belong to the embedded subset are silently dropped.+embeddedEventHandler ::+  (Applicative m) =>+  TypeEmbedding event adapted ->+  EventHandler m event ->+  EventHandler m adapted+embeddedEventHandler emb = eventHandlerMapMaybe emb.extract

@@ -0,0 +1,84 @@+{-# LANGUAGE OverloadedStrings #-}++-- | Defines an event publisher abstraction that decouples event notification+-- from event storage.+--+-- An 'EventPublisher' is responsible for distributing newly stored events to+-- interested consumers using breadth-first dispatch.+--+-- Use 'publishingEventStoreWriter' to wrap an 'EventStoreWriter' so that+-- events are automatically published after a successful write.+module Eventium.EventPublisher+  ( EventPublisher (..),+    synchronousPublisher,+    publishingEventStoreWriter,+    publishingEventStoreWriterTagged,+  )+where++import Eventium.EventHandler+import Eventium.Store.Class+import Eventium.UUID++-- | An 'EventPublisher' pushes versioned stream events to consumers after+-- they have been successfully written to the event store.+newtype EventPublisher m event = EventPublisher+  { publishEvents :: UUID -> [VersionedStreamEvent event] -> m ()+  }++-- | Create an 'EventPublisher' that synchronously delivers events to an+-- 'EventHandler'. Each event in the batch is delivered to the handler in+-- order.+synchronousPublisher ::+  (Monad m) =>+  EventHandler m (VersionedStreamEvent event) ->+  EventPublisher m event+synchronousPublisher handler = EventPublisher $ \_ events ->+  handleEvents handler events++-- | Wrap a 'VersionedEventStoreWriter' so that after a successful write,+-- events are published via the given 'EventPublisher'.+--+-- If the write fails (e.g. due to an optimistic concurrency conflict),+-- no events are published.+--+-- The events are tagged with their stream key and version positions starting+-- from the version returned by the write.+publishingEventStoreWriter ::+  (Monad m) =>+  VersionedEventStoreWriter m event ->+  EventPublisher m event ->+  VersionedEventStoreWriter m event+publishingEventStoreWriter (EventStoreWriter write) (EventPublisher publish) =+  EventStoreWriter $ \uuid expectedPos events -> do+    result <- write uuid expectedPos events+    case result of+      Left err -> return $ Left err+      Right endVersion -> do+        let startVersion = endVersion - fromIntegral (length events) + 1+            versionedEvents = zipWith (\v e -> StreamEvent uuid v (emptyMetadata "") e) [startVersion ..] events+        publish uuid versionedEvents+        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 ::+  (Monad m) =>+  VersionedEventStoreWriter m (TaggedEvent event) ->+  EventPublisher m event ->+  VersionedEventStoreWriter m (TaggedEvent event)+publishingEventStoreWriterTagged (EventStoreWriter write) (EventPublisher publish) =+  EventStoreWriter $ \uuid expectedPos taggedEvents -> do+    result <- write uuid expectedPos taggedEvents+    case result of+      Left err -> return $ Left err+      Right endVersion -> do+        let startVersion = endVersion - fromIntegral (length taggedEvents) + 1+            versionedEvents =+              zipWith+                (\v (TaggedEvent meta e) -> StreamEvent uuid v meta e)+                [startVersion ..]+                taggedEvents+        publish uuid versionedEvents+        return $ Right endVersion

@@ -0,0 +1,237 @@+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}++-- | Defines an event subscription abstraction for push-based event delivery.+--+-- An 'EventSubscription' cleanly separates the delivery mechanism from the+-- consumer (which is just an 'EventHandler').+--+-- This design makes it easy to swap delivery mechanisms (polling, RabbitMQ,+-- Kafka, etc.) without changing the consumer code.+module Eventium.EventSubscription+  ( EventSubscription (..),+    CheckpointStore (..),+    PollingIntervalMillis,+    pollingSubscription,+    runProjectionSubscription,+    RetryConfig (..),+    defaultRetryConfig,+    resilientPollingSubscription,+    resilientRunProjectionSubscription,+  )+where++import Control.Concurrent (threadDelay)+import Control.Exception (SomeException, throwIO, try)+import Control.Monad (forever)+import Control.Monad.IO.Class (MonadIO, liftIO)+import qualified Data.List.NonEmpty as NE+import Eventium.EventHandler+import Eventium.Projection+import Eventium.Store.Class++-- | Polling interval in milliseconds.+type PollingIntervalMillis = Int++-- | A pair of actions for reading and writing checkpoint state.+--+-- Groups the get\/set operations for the state a subscription needs to+-- persist between polls (e.g. a sequence number, or a sequence number+-- paired with projection state).+data CheckpointStore m a = CheckpointStore+  { getCheckpoint :: m a,+    saveCheckpoint :: a -> m ()+  }++-- | An 'EventSubscription' delivers events to an 'EventHandler' using+-- some delivery mechanism (polling, message queue, etc.).+--+-- Call 'runSubscription' with an 'EventHandler' to start consuming events.+-- The subscription runs indefinitely (typically in a separate thread).+newtype EventSubscription m event = EventSubscription+  { runSubscription :: EventHandler m event -> m ()+  }++-- | A single poll cycle: read checkpoint, fetch new events, handle them,+-- update checkpoint, delay.+pollOnce ::+  (MonadIO m) =>+  GlobalEventStoreReader m event ->+  CheckpointStore m SequenceNumber ->+  PollingIntervalMillis ->+  EventHandler m (GlobalStreamEvent event) ->+  m ()+pollOnce globalReader checkpoint pollIntervalMs handler = do+  latestSeq <- checkpoint.getCheckpoint+  newEvents <- globalReader.getEvents (eventsStartingAt () $ latestSeq + 1)+  handleEvents handler newEvents+  case NE.nonEmpty newEvents of+    Nothing -> return ()+    Just ne -> checkpoint.saveCheckpoint (NE.last ne).position+  delayMillis pollIntervalMs++-- | Create a polling-based 'EventSubscription' that reads from the global+-- event stream at a regular interval.+--+-- The caller is responsible for lifting the store reader into the target+-- monad (e.g. via 'runEventStoreReaderUsing') before passing it in.+--+-- Parameters:+--+-- * @globalReader@ — the global event store reader (already in monad @m@)+-- * @checkpoint@ — checkpoint store for tracking the consumed position+-- * @pollIntervalMs@ — milliseconds to wait between polls+pollingSubscription ::+  (MonadIO m) =>+  GlobalEventStoreReader m event ->+  CheckpointStore m SequenceNumber ->+  PollingIntervalMillis ->+  EventSubscription m (GlobalStreamEvent event)+pollingSubscription globalReader checkpoint pollIntervalMs =+  EventSubscription $ \handler ->+    forever $+      pollOnce globalReader checkpoint pollIntervalMs handler++projectionPollOnce ::+  (MonadIO m) =>+  GlobalEventStoreReader m event ->+  Projection state (VersionedStreamEvent event) ->+  CheckpointStore m (SequenceNumber, state) ->+  PollingIntervalMillis ->+  m ()+projectionPollOnce globalReader proj checkpoint pollIntervalMs = do+  (latestSeq, currentState) <- checkpoint.getCheckpoint+  newEvents <- globalReader.getEvents (eventsStartingAt () $ latestSeq + 1)+  let (finalSeq, finalState) = foldl' applyGlobalEvent (latestSeq, currentState) newEvents+  checkpoint.saveCheckpoint (finalSeq, finalState)+  delayMillis pollIntervalMs+  where+    applyGlobalEvent (_, st) (StreamEvent _ newSeq _ innerEvent) =+      let newState = proj.eventHandler st innerEvent+       in (newSeq, newState)++-- | Convenience function to run a polling loop that maintains a+-- 'GlobalStreamProjection'. The projection state is updated with each+-- batch of events and the latest sequence number is tracked.+--+-- The 'CheckpointStore' allows the caller to store the+-- @(SequenceNumber, state)@ pair wherever they choose (e.g. TVar, IORef,+-- database).+runProjectionSubscription ::+  (MonadIO m) =>+  GlobalEventStoreReader m event ->+  Projection state (VersionedStreamEvent event) ->+  CheckpointStore m (SequenceNumber, state) ->+  PollingIntervalMillis ->+  m ()+runProjectionSubscription globalReader proj checkpoint pollIntervalMs =+  forever $ projectionPollOnce globalReader proj checkpoint pollIntervalMs++-- | Configuration for retry behavior when a polling subscription+-- encounters an exception.+data RetryConfig = RetryConfig+  { -- | Initial delay before the first retry, in milliseconds.+    initialDelayMs :: !Int,+    -- | Maximum backoff delay in milliseconds (caps exponential growth).+    maxDelayMs :: !Int,+    -- | Multiplier for exponential backoff on consecutive failures.+    backoffMultiplier :: !Double,+    -- | Predicate called on each exception. Return 'True' to retry+    -- (after backoff), 'False' to re-throw and kill the subscription.+    onError :: SomeException -> IO Bool,+    -- | Callback invoked on each caught exception. Receives the exception+    -- and the consecutive error count (1 on first failure). Use for logging.+    onErrorCallback :: SomeException -> Int -> IO ()+  }++-- | Sensible defaults: 1 second initial delay, 30 second cap, 2x backoff,+-- always retry, no-op callback.+defaultRetryConfig :: RetryConfig+defaultRetryConfig =+  RetryConfig+    { initialDelayMs = 1000,+      maxDelayMs = 30000,+      backoffMultiplier = 2.0,+      onError = const (return True),+      onErrorCallback = \_ _ -> return ()+    }++-- | Like 'pollingSubscription' but with automatic error recovery.+--+-- When an exception occurs during a poll cycle, the exception is caught+-- and the subscription retries after an exponentially increasing backoff+-- delay. On a successful poll, the backoff resets to zero.+--+-- The @unlift@ parameter converts @m@ actions to @IO@, needed to catch+-- exceptions. This follows the same pattern as 'runEventStoreReaderUsing'.+--+-- @+-- -- With m ~ IO:+-- resilientPollingSubscription id globalReader checkpoint 500 defaultRetryConfig+--+-- -- With m ~ SqlPersistT IO:+-- resilientPollingSubscription (flip runSqlPool pool) globalReader checkpoint 500 defaultRetryConfig+-- @+resilientPollingSubscription ::+  (MonadIO m) =>+  (forall a. m a -> IO a) ->+  GlobalEventStoreReader m event ->+  CheckpointStore m SequenceNumber ->+  PollingIntervalMillis ->+  RetryConfig ->+  EventSubscription m (GlobalStreamEvent event)+resilientPollingSubscription unlift globalReader checkpoint pollIntervalMs retryCfg =+  EventSubscription $ \handler -> loop handler 0+  where+    loop handler consecutiveErrors = do+      result <- liftIO $ try $ unlift $ pollOnce globalReader checkpoint pollIntervalMs handler+      case result of+        Right () -> loop handler 0+        Left (ex :: SomeException) -> do+          shouldRetry <- liftIO $ retryCfg.onError ex+          if shouldRetry+            then do+              let newCount = consecutiveErrors + 1+              liftIO $ retryCfg.onErrorCallback ex newCount+              liftIO $ threadDelay (backoffMicros retryCfg newCount)+              loop handler newCount+            else liftIO $ throwIO ex++-- | Like 'runProjectionSubscription' but with automatic error recovery.+-- See 'resilientPollingSubscription' for details on the retry behavior.+resilientRunProjectionSubscription ::+  (MonadIO m) =>+  (forall a. m a -> IO a) ->+  GlobalEventStoreReader m event ->+  Projection state (VersionedStreamEvent event) ->+  CheckpointStore m (SequenceNumber, state) ->+  PollingIntervalMillis ->+  RetryConfig ->+  m ()+resilientRunProjectionSubscription unlift globalReader proj checkpoint pollIntervalMs retryCfg =+  loop 0+  where+    loop consecutiveErrors = do+      result <- liftIO $ try $ unlift $ projectionPollOnce globalReader proj checkpoint pollIntervalMs+      case result of+        Right () -> loop 0+        Left (ex :: SomeException) -> do+          shouldRetry <- liftIO $ retryCfg.onError ex+          if shouldRetry+            then do+              let newCount = consecutiveErrors + 1+              liftIO $ retryCfg.onErrorCallback ex newCount+              liftIO $ threadDelay (backoffMicros retryCfg newCount)+              loop newCount+            else liftIO $ throwIO ex++backoffMicros :: RetryConfig -> Int -> Int+backoffMicros retryCfg count =+  let delayMs =+        min retryCfg.maxDelayMs $+          round (fromIntegral retryCfg.initialDelayMs * retryCfg.backoffMultiplier ^^ max 0 (count - 1))+   in delayMs * 1000++-- | Sleep for the given number of milliseconds.+delayMillis :: (MonadIO m) => PollingIntervalMillis -> m ()+delayMillis ms = liftIO $ threadDelay (ms * 1000)

@@ -0,0 +1,42 @@+-- | Shared JSON serialization utilities. Provides Aeson 'Options' and string+-- helpers that strip record-field prefixes, matching the convention used+-- across all eventium packages and example applications.+module Eventium.Json+  ( unPrefixLower,+    dropPrefix,+    dropSuffix,+  )+where++import Data.Aeson+import Data.Char (toLower)++-- | Aeson 'Options' that strip a given prefix from record field names+-- and lowercase the first character of the remainder.+unPrefixLower :: String -> Options+unPrefixLower prefix =+  defaultOptions+    { fieldLabelModifier = unCapitalize . dropPrefix prefix+    }++unCapitalize :: String -> String+unCapitalize [] = []+unCapitalize (c : cs) = toLower c : cs++-- | Strip an exact prefix from a string. Errors if the prefix is not present.+dropPrefix :: String -> String -> String+dropPrefix = dropPrefix' "dropPrefix" id++-- | Strip an exact suffix from a string. Errors if the suffix is not present.+dropSuffix :: String -> String -> String+dropSuffix prefix input = reverse $ dropPrefix' "dropSuffix" reverse (reverse prefix) (reverse input)++dropPrefix' :: String -> (String -> String) -> String -> String -> String+dropPrefix' fnName strTrans prefix input = go prefix input+  where+    go pre [] = error $ contextual $ "prefix leftover: " ++ strTrans pre+    go [] (c : cs) = c : cs+    go (p : preRest) (c : cRest)+      | p == c = go preRest cRest+      | otherwise = error $ contextual $ "not equal: " ++ strTrans (p : preRest) ++ " " ++ strTrans (c : cRest)+    contextual msg = fnName ++ ": " ++ msg ++ ". " ++ strTrans prefix ++ " " ++ strTrans input

@@ -0,0 +1,20 @@+-- | Template Haskell helpers for JSON serialization. Separated from+-- "Eventium.Json" to avoid TH linker issues in downstream packages.+module Eventium.Json.TH+  ( deriveJSONUnPrefixLower,+  )+where++import Data.Aeson.TH+import Data.Char (toLower)+import Eventium.Json (unPrefixLower)+import Language.Haskell.TH++-- | Derive 'ToJSON' and 'FromJSON' using 'unPrefixLower' with the type name+-- (first character lowercased) as prefix.+deriveJSONUnPrefixLower :: Name -> Q [Dec]+deriveJSONUnPrefixLower name = deriveJSON (unPrefixLower $ firstCharToLower $ nameBase name) name++firstCharToLower :: String -> String+firstCharToLower [] = []+firstCharToLower (x : xs) = toLower x : xs

@@ -1,60 +1,145 @@-{-# LANGUAGE ExistentialQuantification #-}-{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE OverloadedStrings #-} +-- | Defines a Process Manager (saga) abstraction for orchestrating+-- interactions across multiple event streams.+--+-- A 'ProcessManager' is the combination of a 'Projection' (to track state+-- across streams) and a pure @react@ function that decides what commands to+-- issue in response to each event. Commands are represented as+-- 'ProcessManagerEffect' values — a pure data type — which are then+-- executed by 'runProcessManagerEffects'.+--+-- This design cleanly separates the pure decision logic from effectful+-- execution, making process managers easy to unit test. module Eventium.ProcessManager   ( ProcessManager (..),-    ProcessManagerCommand (..),-    applyProcessManagerCommandsAndEvents,+    ProcessManagerEffect (..),+    CommandDispatchResult (..),+    RejectionReason (..),+    CommandDispatcher (..),+    mkCommandDispatcher,+    fireAndForgetDispatcher,+    runProcessManagerEffects,+    processManagerEventHandler,   ) where -import Control.Monad (forM_, void)-import Eventium.CommandHandler+import Control.Monad (void)+import Data.String (IsString)+import Data.Text (Text)+import Eventium.EventHandler (EventHandler (..)) import Eventium.Projection-import Eventium.Store.Class+import Eventium.Store.Class (GlobalEventStoreReader)+import Eventium.Store.Types import Eventium.UUID --- | A 'ProcessManager' manages interaction between event streams. It works by--- listening to events on an event bus and applying events to its internal--- 'Projection' (see 'applyProcessManagerCommandsAndEvents'). Then, pending--- commands and events are plucked off of that Projection and applied to the--- appropriate 'CommandHandler' or Projections in other streams.-data ProcessManager state event command-  = ProcessManager-  { processManagerProjection :: Projection state (VersionedStreamEvent event),-    processManagerPendingCommands :: state -> [ProcessManagerCommand event command],-    processManagerPendingEvents :: state -> [StreamEvent UUID () event]+-- | A 'ProcessManager' manages interaction between event streams. It+-- listens to events and decides what commands to issue to other aggregates.+--+-- * 'projection' — a pure fold over versioned stream events+--   that tracks the process manager's state.+-- * 'react' — a pure function that, given the current state+--   and a new event, returns a list of effects to execute.+data ProcessManager state event command = ProcessManager+  { projection :: Projection state (VersionedStreamEvent event),+    react :: state -> VersionedStreamEvent event -> [ProcessManagerEffect command]   } --- | This is a @command@ along with the UUID of the target 'CommandHandler', as--- well as the 'CommandHandler' type. Note that this uses an existential type--- to hide the @state@ type parameter on the CommandHandler.-data ProcessManagerCommand event command-  = forall state. ProcessManagerCommand-  { processManagerCommandTargetId :: UUID,-    processManagerCommandCommandHandler :: CommandHandler state event command,-    processManagerCommandCommand :: command+-- | A typed wrapper for the reason a command was rejected.+--+-- Use 'RejectionReason' instead of raw 'Text' to avoid accidentally+-- mixing rejection reasons with other textual values at the dispatch boundary.+newtype RejectionReason = RejectionReason {unRejectionReason :: Text}+  deriving (Show, Eq, Ord, IsString)++-- | A side effect that a 'ProcessManager' wants to perform. This is a pure+-- data type — it describes /what/ should happen, not /how/.+data ProcessManagerEffect command+  = -- | Issue a command to a specific aggregate (identified by 'UUID').+    IssueCommand UUID command+  | -- | 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])++instance (Show command) => Show (ProcessManagerEffect command) where+  show (IssueCommand uuid cmd) = "IssueCommand " ++ show uuid ++ " " ++ show cmd+  show (IssueCommandWithCompensation uuid cmd _) =+    "IssueCommandWithCompensation " ++ show uuid ++ " " ++ show cmd ++ " <compensation>"++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+  _ == _ = False++-- | Result of dispatching a command to an aggregate.+data CommandDispatchResult+  = -- | The command was accepted and events were stored.+    CommandSucceeded+  | -- | The command was rejected by the aggregate with a reason.+    CommandFailed RejectionReason+  deriving (Show, Eq)++-- | A command dispatcher routes commands to aggregates and reports the outcome.+--+-- Use 'mkCommandDispatcher' to construct one from a dispatch function.+-- 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   } -instance (Show command, Show event) => Show (ProcessManagerCommand event command) where-  show (ProcessManagerCommand uuid _ command) =-    "ProcessManagerCommand{processManagerCommandCommandHandlerId = "-      ++ show uuid-      ++ ", processManagerCommandCommand = "-      ++ show command-      ++ "}"+-- | Construct a 'CommandDispatcher' from a dispatch function.+mkCommandDispatcher ::+  (UUID -> command -> m CommandDispatchResult) ->+  CommandDispatcher m command+mkCommandDispatcher = CommandDispatcher --- | Plucks the pending commands and events off of the process manager's state--- and applies them to the appropriate locations in the event store.-applyProcessManagerCommandsAndEvents ::+-- | Adapt a legacy fire-and-forget dispatcher into a 'CommandDispatcher'+-- that always reports 'CommandSucceeded'.+fireAndForgetDispatcher ::   (Monad m) =>-  ProcessManager state event command ->-  VersionedEventStoreWriter m event ->-  VersionedEventStoreReader m event ->-  state ->+  (UUID -> command -> m ()) ->+  CommandDispatcher m command+fireAndForgetDispatcher f = CommandDispatcher $ \uuid cmd ->+  f uuid cmd >> pure CommandSucceeded++-- | Execute a list of 'ProcessManagerEffect's using the provided+-- 'CommandDispatcher'.+runProcessManagerEffects ::+  (Monad m) =>+  CommandDispatcher m command ->+  [ProcessManagerEffect command] ->   m ()-applyProcessManagerCommandsAndEvents ProcessManager {..} writer reader state = do-  forM_ (processManagerPendingCommands state) $ \(ProcessManagerCommand targetId commandHandler command) ->-    void $ applyCommandHandler writer reader commandHandler targetId command-  forM_ (processManagerPendingEvents state) $ \(StreamEvent projectionId () event) ->-    storeEvents writer projectionId AnyPosition [event]+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+      case result of+        CommandSucceeded -> pure ()+        CommandFailed reason -> mapM_ go (onFailure reason)++-- | Create an 'EventHandler' that wires a 'ProcessManager' to a global+-- event store reader and a command dispatcher.+--+-- For each incoming event:+--+--   1. Rebuilds the process manager state from the global event stream+--   2. Calls 'react' with the current state and the new event+--   3. Executes the resulting effects via the dispatcher+processManagerEventHandler ::+  (Monad m) =>+  ProcessManager state event command ->+  GlobalEventStoreReader m event ->+  CommandDispatcher m command ->+  EventHandler m (VersionedStreamEvent event)+processManagerEventHandler pm globalReader dispatcher = EventHandler $ \event -> do+  let globalProj = globalStreamProjection pm.projection+  sp <- getLatestStreamProjection globalReader globalProj+  let effects = pm.react sp.state event+  runProcessManagerEffects dispatcher effects

@@ -1,5 +1,3 @@-{-# LANGUAGE RecordWildCards #-}- module Eventium.Projection   ( Projection (..),     latestProjection,@@ -12,16 +10,19 @@     globalStreamProjection,     streamProjectionEventHandler,     getLatestStreamProjection,-    serializedProjection,+    codecProjection,+    lenientCodecProjection,+    embeddedProjection,     projectionMapMaybe,   ) where -import Data.Foldable (foldl')+import Control.Exception (throw) import Data.Functor.Contravariant import Data.List (scanl')-import Eventium.Serializer+import Eventium.Codec import Eventium.Store.Class+import Eventium.TypeEmbedding import Eventium.UUID  -- | A 'Projection' is a piece of @state@ that is constructed only from@@ -33,27 +34,27 @@ data Projection state event   = Projection   { -- | Initial state of a projection-    projectionSeed :: state,+    seed :: state,     -- | The function that applies and event to the current state, producing a     -- new state.-    projectionEventHandler :: state -> event -> state+    eventHandler :: state -> event -> state   }  instance Contravariant (Projection state) where-  contramap f (Projection seed handler) = Projection seed handler'+  contramap f (Projection s handler) = Projection s handler'     where       handler' state event = handler state (f event)  -- | Computes the latest state of a 'Projection' from some events. latestProjection :: (Foldable t) => Projection state event -> t event -> state-latestProjection (Projection seed handler) = foldl' handler seed+latestProjection proj = foldl' proj.eventHandler proj.seed  -- | Given a list of events, produce all the Projections that were ever--- produced. Just a 'scanl' using 'projectionEventHandler'. This function is+-- produced. Just a 'scanl' using 'eventHandler'. This function is -- useful for testing 'Projection's; you can easily assert that all the states -- of a Projection are valid given a list of events. allProjections :: Projection state event -> [event] -> [state]-allProjections (Projection seed handler) = scanl' handler seed+allProjections proj = scanl' proj.eventHandler proj.seed  -- | A 'StreamProjection' is a 'Projection' that has been constructed from -- events from a particular event stream. This is useful when we want to cache@@ -61,10 +62,10 @@ -- is caught up to. data StreamProjection key position state event   = StreamProjection-  { streamProjectionKey :: !key,-    streamProjectionPosition :: !position,-    streamProjectionProjection :: !(Projection state event),-    streamProjectionState :: !state+  { key :: !key,+    position :: !position,+    projection :: !(Projection state event),+    state :: !state   }  type VersionedStreamProjection = StreamProjection UUID EventVersion@@ -77,8 +78,8 @@   position ->   Projection state event ->   StreamProjection key position state event-streamProjection key position projection@Projection {..} =-  StreamProjection key position projection projectionSeed+streamProjection k pos proj =+  StreamProjection k pos proj proj.seed  -- | Initialize a 'VersionedStreamProjection'. versionedStreamProjection ::@@ -101,11 +102,10 @@   StreamProjection key position state event ->   StreamEvent eventKey position event ->   StreamProjection key position state event-streamProjectionEventHandler StreamProjection {..} event =-  let Projection {..} = streamProjectionProjection-      position' = streamEventPosition event-      state' = projectionEventHandler streamProjectionState (streamEventEvent event)-   in StreamProjection streamProjectionKey position' streamProjectionProjection state'+streamProjectionEventHandler sp streamEvent =+  let position' = streamEvent.position+      state' = sp.projection.eventHandler sp.state streamEvent.payload+   in StreamProjection sp.key position' sp.projection state'  -- | Gets the latest projection from a store by querying events from the latest -- order key and then applying the events using the Projection's event handler.@@ -114,24 +114,53 @@   EventStoreReader key position m (StreamEvent key position event) ->   StreamProjection key position state event ->   m (StreamProjection key position state event)-getLatestStreamProjection (EventStoreReader getEvents') projection@StreamProjection {..} = do-  events <- getEvents' (eventsStartingAt streamProjectionKey $ streamProjectionPosition + 1)-  return $ foldl' streamProjectionEventHandler projection events+getLatestStreamProjection (EventStoreReader getEvents') sp = do+  events <- getEvents' (eventsStartingAt sp.key $ sp.position + 1)+  return $ foldl' streamProjectionEventHandler sp events --- | Use a 'Serializer' to wrap a 'Projection' with event type @event@ so it--- uses the @serialized@ type.-serializedProjection ::+-- | Use a 'Codec' to wrap a 'Projection' with event type @event@ so it+-- uses the @encoded@ type. Throws 'DecodeError' if decoding+-- fails. Use 'lenientCodecProjection' to silently skip unrecognized events.+codecProjection ::+  Codec event encoded ->   Projection state event ->-  Serializer event serialized ->-  Projection state serialized-serializedProjection proj Serializer {..} = projectionMapMaybe deserialize proj+  Projection state encoded+codecProjection codec (Projection s handler) =+  Projection s handler'+  where+    handler' st encoded =+      case codec.decode encoded of+        Just event -> handler st event+        Nothing -> throw $ DecodeError "codecProjection" "Failed to decode event" +-- | Like 'codecProjection' but silently drops events that fail to+-- decode. Useful for projections that only care about a subset of event+-- types (e.g. when using a sum-type codec).+--+-- Recommended for production use when using sum-type codecs, as it+-- allows projections to gracefully handle events they don't recognize.+lenientCodecProjection ::+  Codec event encoded ->+  Projection state event ->+  Projection state encoded+lenientCodecProjection codec = projectionMapMaybe codec.decode++-- | Adapt a 'Projection' using a 'TypeEmbedding'. Events that do not belong+-- to the embedded subset are silently skipped. This is the correct semantics+-- for type embeddings, where it is expected that many events in the superset+-- will not match the subset.+embeddedProjection ::+  TypeEmbedding event adapted ->+  Projection state event ->+  Projection state adapted+embeddedProjection emb = projectionMapMaybe emb.extract+ -- | Transform a 'Projection' when you only have a partial relationship between -- the source event type and the target event type. projectionMapMaybe ::   (eventB -> Maybe eventA) ->   Projection state eventA ->   Projection state eventB-projectionMapMaybe f (Projection seed handler) = Projection seed handler'+projectionMapMaybe f (Projection s handler) = Projection s handler'   where-    handler' state = maybe state (handler state) . f+    handler' st = maybe st (handler st) . f

@@ -0,0 +1,140 @@+{-# LANGUAGE RankNTypes #-}++module Eventium.ProjectionCache.Cache+  ( runProjectionCacheUsing,+    codecProjectionCache,+    getLatestVersionedProjectionWithCache,+    getLatestGlobalProjectionWithCache,+    updateVersionedProjectionCache,+    updateGlobalProjectionCache,+    snapshotEventHandler,+    snapshotGlobalEventHandler,+    module Eventium.ProjectionCache.Types,+  )+where++import Eventium.Codec+import Eventium.EventHandler+import Eventium.Projection+import Eventium.ProjectionCache.Types+import Eventium.Store.Class++-- | Changes the monad a 'ProjectionCache' runs in. This is useful to run the+-- cache in another 'Monad' while forgetting the original 'Monad'.+runProjectionCacheUsing ::+  (forall a. mstore a -> m a) ->+  ProjectionCache key position encoded mstore ->+  ProjectionCache key position encoded m+runProjectionCacheUsing runCache pc =+  ProjectionCache+    { storeSnapshot = \uuid version st -> runCache $ pc.storeSnapshot uuid version st,+      loadSnapshot = runCache . pc.loadSnapshot+    }++-- | Wraps a 'ProjectionCache' and transparently encodes/decodes events for+-- you. Note that in this implementation decoding errors when using+-- 'getEvents' are simply ignored (the event is not returned).+codecProjectionCache ::+  (Monad m) =>+  Codec state encoded ->+  ProjectionCache key position encoded m ->+  ProjectionCache key position state m+codecProjectionCache codec pc =+  ProjectionCache storeSnapshot' loadSnapshot'+  where+    storeSnapshot' uuid version = pc.storeSnapshot uuid version . codec.encode+    loadSnapshot' uuid = do+      mState <- pc.loadSnapshot uuid+      return $ mState >>= traverse codec.decode++-- | Like 'getLatestVersionedProjection', but uses a 'ProjectionCache' if it contains+-- more recent state.+getLatestVersionedProjectionWithCache ::+  (Monad m) =>+  VersionedEventStoreReader m event ->+  VersionedProjectionCache state m ->+  VersionedStreamProjection state event ->+  m (VersionedStreamProjection state event)+getLatestVersionedProjectionWithCache store cache proj =+  getLatestProjectionWithCache' cache proj proj.key >>= getLatestStreamProjection store++-- | Like 'getLatestGlobalProjection', but uses a 'ProjectionCache' if it+-- contains more recent state.+getLatestGlobalProjectionWithCache ::+  (Monad m) =>+  GlobalEventStoreReader m event ->+  GlobalProjectionCache state m ->+  GlobalStreamProjection state event ->+  m (GlobalStreamProjection state event)+getLatestGlobalProjectionWithCache store cache proj =+  getLatestProjectionWithCache' cache proj () >>= getLatestStreamProjection store++getLatestProjectionWithCache' ::+  (Monad m, Ord position) =>+  ProjectionCache key position state m ->+  StreamProjection projKey position state event ->+  key ->+  m (StreamProjection projKey position state event)+getLatestProjectionWithCache' cache proj k = do+  mLatestState <- cache.loadSnapshot k+  let mkProjection' (pos, st) =+        if pos > proj.position+          then+            proj+              { position = pos,+                state = st+              }+          else proj+  return $ maybe proj mkProjection' mLatestState++-- | Loads the latest projection state from the cache/store and stores this+-- value back into the projection cache.+updateVersionedProjectionCache ::+  (Monad m) =>+  VersionedEventStoreReader m event ->+  VersionedProjectionCache state m ->+  VersionedStreamProjection state event ->+  m ()+updateVersionedProjectionCache reader cache proj = do+  sp <- getLatestVersionedProjectionWithCache reader cache proj+  cache.storeSnapshot sp.key sp.position sp.state++-- | Analog of 'updateVersionedProjectionCache' for a 'GlobalProjectionCache'.+updateGlobalProjectionCache ::+  (Monad m) =>+  GlobalEventStoreReader m event ->+  GlobalProjectionCache state m ->+  GlobalStreamProjection state event ->+  m ()+updateGlobalProjectionCache reader cache proj = do+  sp <- getLatestGlobalProjectionWithCache reader cache proj+  cache.storeSnapshot () sp.position sp.state++-- | Creates an 'EventHandler' that updates a 'VersionedProjectionCache'+-- whenever a versioned stream event is received. Compose with+-- @synchronousPublisher@ or @publishingEventStoreWriter@ to keep the cache+-- up to date as events are written.+snapshotEventHandler ::+  (Monad m) =>+  VersionedEventStoreReader m event ->+  VersionedProjectionCache state m ->+  Projection state event ->+  EventHandler m (VersionedStreamEvent event)+snapshotEventHandler reader cache proj =+  EventHandler $ \streamEvent -> do+    let uuid = streamEvent.key+    updateVersionedProjectionCache reader cache (versionedStreamProjection uuid proj)++-- | Creates an 'EventHandler' that updates a 'GlobalProjectionCache'+-- whenever a global stream event is received. Compose with+-- @synchronousPublisher@ or @publishingEventStoreWriter@ to keep the cache+-- up to date as events are written.+snapshotGlobalEventHandler ::+  (Monad m) =>+  GlobalEventStoreReader m event ->+  GlobalProjectionCache state m ->+  Projection state (VersionedStreamEvent event) ->+  EventHandler m (GlobalStreamEvent event)+snapshotGlobalEventHandler reader cache proj =+  EventHandler $ \_ ->+    updateGlobalProjectionCache reader cache (globalStreamProjection proj)

@@ -1,27 +1,16 @@-{-# LANGUAGE RankNTypes #-}-{-# LANGUAGE RecordWildCards #-}- module Eventium.ProjectionCache.Types   ( ProjectionCache (..),     VersionedProjectionCache,-    GlobalStreamProjectionCache,-    runProjectionCacheUsing,-    serializedProjectionCache,-    getLatestVersionedProjectionWithCache,-    getLatestGlobalProjectionWithCache,-    updateProjectionCache,-    updateGlobalProjectionCache,+    GlobalProjectionCache,   ) where -import Eventium.Projection-import Eventium.Serializer-import Eventium.Store.Class+import Eventium.Store.Types import Eventium.UUID  -- | A 'ProjectionCache' caches snapshots of 'Projection's in event streams. -- This is useful if your event streams are very large. This cache operates on--- some 'Monad' @m@ and stores the 'Projection' state of type @serialized@.+-- some 'Monad' @m@ and stores the 'Projection' state of type @encoded@. -- -- At its core, this is essentially just a key-value store with knowledge of -- the stream 'UUID' and 'EventVersion'. It is recommended to use the other@@ -31,114 +20,25 @@ -- The @key@ and @position@ type parameters are polymorphic so we can abstract -- over a cache for individual event streams, and a cache for globally ordered -- streams.-data ProjectionCache key position serialized m+data ProjectionCache key position encoded m   = ProjectionCache   { -- | Stores the state for a projection at a given @key@ and @position@.     -- This is pretty unsafe, because there is no guarantee what is stored is     -- actually derived from the events in the stream. Consider using-    -- 'updateProjectionCache'.-    storeProjectionSnapshot :: key -> position -> serialized -> m (),+    -- 'updateVersionedProjectionCache'.+    storeSnapshot :: key -> position -> encoded -> m (),     -- | Loads the latest projection state from the cache.-    loadProjectionSnapshot :: key -> m (Maybe (position, serialized))+    loadSnapshot :: key -> m (Maybe (position, encoded))   }  -- | Type synonym for a 'ProjectionCache' used on individual event streams.-type VersionedProjectionCache serialized m = ProjectionCache UUID EventVersion serialized m+type VersionedProjectionCache encoded m = ProjectionCache UUID EventVersion encoded m  -- | Type synonym for a 'ProjectionCache' that is used in conjunction with a--- 'GlobalStreamEventStore'.-type GlobalStreamProjectionCache key serialized m = ProjectionCache key SequenceNumber serialized m---- | Changes the monad a 'ProjectionCache' runs in. This is useful to run the--- cache in another 'Monad' while forgetting the original 'Monad'.-runProjectionCacheUsing ::-  (Monad m, Monad mstore) =>-  (forall a. mstore a -> m a) ->-  ProjectionCache key position serialized mstore ->-  ProjectionCache key position serialized m-runProjectionCacheUsing runCache ProjectionCache {..} =-  ProjectionCache-    { storeProjectionSnapshot = \uuid version state -> runCache $ storeProjectionSnapshot uuid version state,-      loadProjectionSnapshot = runCache . loadProjectionSnapshot-    }---- | Wraps a 'ProjectionCache' and transparently serializes/deserializes events for--- you. Note that in this implementation deserialization errors when using--- 'getEvents' are simply ignored (the event is not returned).-serializedProjectionCache ::-  (Monad m) =>-  Serializer state serialized ->-  ProjectionCache key position serialized m ->-  ProjectionCache key position state m-serializedProjectionCache Serializer {..} ProjectionCache {..} =-  ProjectionCache storeProjectionSnapshot' loadProjectionSnapshot'-  where-    storeProjectionSnapshot' uuid version = storeProjectionSnapshot uuid version . serialize-    loadProjectionSnapshot' uuid = do-      mState <- loadProjectionSnapshot uuid-      return $ mState >>= traverse deserialize---- | Like 'getLatestVersionedProjection', but uses a 'ProjectionCache' if it contains--- more recent state.-getLatestVersionedProjectionWithCache ::-  (Monad m) =>-  VersionedEventStoreReader m event ->-  VersionedProjectionCache state m ->-  VersionedStreamProjection state event ->-  m (VersionedStreamProjection state event)-getLatestVersionedProjectionWithCache store cache projection =-  getLatestProjectionWithCache' cache projection (streamProjectionKey projection) >>= getLatestStreamProjection store---- | Like 'getLatestGlobalProjection', but uses a 'ProjectionCache' if it--- contains more recent state.-getLatestGlobalProjectionWithCache ::-  (Monad m) =>-  GlobalEventStoreReader m event ->-  GlobalStreamProjectionCache key state m ->-  GlobalStreamProjection state event ->-  key ->-  m (GlobalStreamProjection state event)-getLatestGlobalProjectionWithCache store cache projection key =-  getLatestProjectionWithCache' cache projection key >>= getLatestStreamProjection store--getLatestProjectionWithCache' ::-  (Monad m, Ord position) =>-  ProjectionCache key position state m ->-  StreamProjection projKey position state event ->-  key ->-  m (StreamProjection projKey position state event)-getLatestProjectionWithCache' cache projection key = do-  mLatestState <- loadProjectionSnapshot cache key-  let mkProjection' (position, state) =-        if position > streamProjectionPosition projection-          then-            projection-              { streamProjectionPosition = position,-                streamProjectionState = state-              }-          else projection-  return $ maybe projection mkProjection' mLatestState---- | Loads the latest projection state from the cache/store and stores this--- value back into the projection cache.-updateProjectionCache ::-  (Monad m) =>-  VersionedEventStoreReader m event ->-  VersionedProjectionCache state m ->-  VersionedStreamProjection state event ->-  m ()-updateProjectionCache reader cache projection = do-  StreamProjection {..} <- getLatestVersionedProjectionWithCache reader cache projection-  storeProjectionSnapshot cache streamProjectionKey streamProjectionPosition streamProjectionState---- | Analog of 'updateProjectionCache' for a 'GlobalStreamProjectionCache'.-updateGlobalProjectionCache ::-  (Monad m) =>-  GlobalEventStoreReader m event ->-  GlobalStreamProjectionCache key state m ->-  GlobalStreamProjection state event ->-  key ->-  m ()-updateGlobalProjectionCache reader cache projection key = do-  StreamProjection {..} <- getLatestGlobalProjectionWithCache reader cache projection key-  storeProjectionSnapshot cache key streamProjectionPosition streamProjectionState+-- 'GlobalStreamEventStore'. The key is fixed to @()@ (singleton) because+-- scoping is handled at construction time via 'ProjectionName' in the SQL+-- backends, rather than as a runtime parameter on each store\/load call.+-- The original @eventful@ library kept the key polymorphic+-- (@GloballyOrderedProjectionCache key serialized m@), but that is redundant+-- when each cache instance is already scoped by name.+type GlobalProjectionCache encoded m = ProjectionCache () SequenceNumber encoded m

@@ -0,0 +1,111 @@+-- | Defines the ReadModel abstraction for queryable persistent views.+--+-- A 'ReadModel' bundles everything needed to maintain an event-driven+-- persistent view: initialization, event handling, checkpointing, and reset.+--+-- The library manages the event-to-handler pipeline and checkpointing.+-- Users define their own schema, write the 'EventHandler' that populates+-- tables, and query via their own code (SQL, persistent, etc.).+module Eventium.ReadModel+  ( ReadModel (..),+    runReadModel,+    rebuildReadModel,+    combineReadModels,+  )+where++import Control.Concurrent (threadDelay)+import Control.Monad (forever)+import Control.Monad.IO.Class (MonadIO, liftIO)+import Data.Foldable (traverse_)+import qualified Data.List.NonEmpty as NE+import Eventium.EventHandler+import Eventium.EventSubscription (CheckpointStore (..), PollingIntervalMillis)+import Eventium.Store.Class++-- | A read model that maintains a queryable persistent view from events.+--+-- Users provide:+--+-- * 'initialize' — idempotent setup (run migrations, create tables)+-- * 'eventHandler' — processes global stream events, writes to user-defined storage+-- * 'checkpointStore' — tracks the last processed 'SequenceNumber'+-- * 'reset' — drop view data and reset checkpoint (for full rebuilds)+data ReadModel m event = ReadModel+  { initialize :: m (),+    eventHandler :: EventHandler m (GlobalStreamEvent event),+    checkpointStore :: CheckpointStore m SequenceNumber,+    reset :: m ()+  }++-- | Subscribe to the global event stream and keep the read model updated.+-- Runs forever, polling at the given interval.+runReadModel ::+  (MonadIO m) =>+  GlobalEventStoreReader m event ->+  PollingIntervalMillis ->+  ReadModel m event ->+  m ()+runReadModel globalReader pollIntervalMs rm = do+  rm.initialize+  forever $ pollReadModelOnce globalReader pollIntervalMs rm++-- | Reset the read model and replay all events from the beginning.+-- Returns after processing all currently available events.+rebuildReadModel ::+  (Monad m) =>+  GlobalEventStoreReader m event ->+  ReadModel m event ->+  m ()+rebuildReadModel globalReader rm = do+  rm.reset+  rm.initialize+  replayAll+  where+    replayAll = do+      latestSeq <- rm.checkpointStore.getCheckpoint+      newEvents <- globalReader.getEvents (eventsStartingAt () $ latestSeq + 1)+      case NE.nonEmpty newEvents of+        Nothing -> return ()+        Just ne -> do+          handleEvents rm.eventHandler newEvents+          rm.checkpointStore.saveCheckpoint (NE.last ne).position+          replayAll++-- | Combine multiple read models into one. Events are fanned out to all+-- handlers. Initialize and reset run all sub-models.+--+-- For independently checkpointed read models, prefer running them+-- separately with 'runReadModel'.+combineReadModels ::+  (Applicative m) =>+  [ReadModel m event] ->+  ReadModel m event+combineReadModels rms =+  ReadModel+    { initialize = traverse_ (.initialize) rms,+      eventHandler = foldMap (.eventHandler) rms,+      checkpointStore =+        CheckpointStore+          { getCheckpoint = case rms of+              [] -> pure 0+              (rm : _) -> rm.checkpointStore.getCheckpoint,+            saveCheckpoint = \sn -> traverse_ (\rm -> rm.checkpointStore.saveCheckpoint sn) rms+          },+      reset = traverse_ (.reset) rms+    }++pollReadModelOnce ::+  (MonadIO m) =>+  GlobalEventStoreReader m event ->+  PollingIntervalMillis ->+  ReadModel m event ->+  m ()+pollReadModelOnce globalReader pollIntervalMs rm = do+  latestSeq <- rm.checkpointStore.getCheckpoint+  newEvents <- globalReader.getEvents (eventsStartingAt () $ latestSeq + 1)+  handleEvents rm.eventHandler newEvents+  case NE.nonEmpty newEvents of+    Nothing -> return ()+    Just ne -> rm.checkpointStore.saveCheckpoint (NE.last ne).position+  liftIO $ threadDelay (pollIntervalMs * 1000)

@@ -1,48 +0,0 @@-{-# LANGUAGE RankNTypes #-}-{-# LANGUAGE RecordWildCards #-}--module Eventium.ReadModel.Class-  ( ReadModel (..),-    runPollingReadModel,-  )-where--import Control.Concurrent (threadDelay)-import Control.Monad (forever)-import Control.Monad.IO.Class (MonadIO, liftIO)-import Eventium.Store.Class--data ReadModel model serialized m-  = ReadModel-  { readModelModel :: model,-    readModelLatestAppliedSequence :: model -> m SequenceNumber,-    readModelHandleEvents :: model -> [GlobalStreamEvent serialized] -> m ()-  }--type PollingPeriodSeconds = Double--runPollingReadModel ::-  (MonadIO m, Monad mstore) =>-  ReadModel model serialized m ->-  GlobalEventStoreReader mstore serialized ->-  (forall a. mstore a -> m a) ->-  PollingPeriodSeconds ->-  m ()-runPollingReadModel ReadModel {..} globalReader runStore waitSeconds = forever $ do-  -- Get new events starting from latest applied sequence number-  latestSeq <- readModelLatestAppliedSequence readModelModel-  newEvents <- runStore $ getEvents globalReader (eventsStartingAt () $ latestSeq + 1)--  -- Handle the new events-  readModelHandleEvents readModelModel newEvents--  -- Wait before running again-  liftIO $ threadDelay $ ceiling (waitSeconds * 1000000) -- threadDelay accepts microseconds---- data EventHandler m serialized = forall event. (Serializable event serialized, Monad m) => EventHandler (event -> m ())---- combineHandlers :: (Monad m) => [EventHandler m serialized] -> (serialized -> m ())--- combineHandlers handlers event = mapM_ ($ event) (mkHandler <$> handlers)---- mkHandler :: EventHandler m serialized -> (serialized -> m ())--- mkHandler (EventHandler handler) event = maybe (return ()) handler (deserialize event)

@@ -1,204 +0,0 @@-{-# LANGUAGE DefaultSignatures #-}-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE FlexibleInstances #-}-{-# LANGUAGE RecordWildCards #-}-{-# LANGUAGE TypeOperators #-}--module Eventium.Serializer-  ( -- * Class-    Serializer (..),-    simpleSerializer,-    composeSerializers,--    -- * Common serializers-    idSerializer,-    traverseSerializer,-    jsonSerializer,-    jsonTextSerializer,-    dynamicSerializer,--    -- * Sum types-    EventSumType (..),-    eventSumTypeSerializer,-  )-where--import Control.Applicative ((<|>))-import Data.Aeson-import Data.Dynamic-import Data.Maybe (fromMaybe)-import qualified Data.Text.Lazy as TL-import qualified Data.Text.Lazy.Encoding as TLE-import Data.Typeable (typeOf)-import GHC.Generics---- | A 'Serializer' describes the injective conversion between types @a@ and--- @b@. In plain English, this means that you can go from @a@ to @b@, and you--- can 'Maybe' go from @b@ back to @a@. This is often used to serialize events--- to an event store, and then deserialize them back.-data Serializer a b-  = Serializer-  { serialize :: a -> b,-    deserialize :: b -> Maybe a,-    -- | Deserialize with additional information on failure-    deserializeEither :: b -> Either String a-  }---- | Simple constructor to just use 'deserialize' to construct--- 'deserializeEither'.-simpleSerializer ::-  (a -> b) ->-  (b -> Maybe a) ->-  Serializer a b-simpleSerializer serialize' deserialize' =-  Serializer-    { serialize = serialize',-      deserialize = deserialize',-      deserializeEither = maybe (Left "Serializable: Failed to deserialize") Right . deserialize'-    }---- | Apply an intermediate 'Serializer' to a serializer to go from type @a@ to--- @c@ with @b@ in the middle. Note that with deserializing, if the conversion--- from @c@ to @b@ or from @b@ to @a@ fails, the whole deserialization fails.-composeSerializers :: Serializer a b -> Serializer b c -> Serializer a c-composeSerializers serializer1 serializer2 = Serializer serialize' deserialize' deserializeEither'-  where-    serialize' = serialize serializer2 . serialize serializer1-    deserialize' x = deserialize serializer2 x >>= deserialize serializer1-    deserializeEither' x = deserializeEither serializer2 x >>= deserializeEither serializer1---- | Simple "serializer" using 'id'. Useful for when an API requires a--- serializer but you don't need to actually change types.-idSerializer :: Serializer a a-idSerializer = simpleSerializer id Just---- | Uses 'Traversable' to wrap a 'Serializer'.-traverseSerializer ::-  (Traversable t) =>-  Serializer a b ->-  Serializer (t a) (t b)-traverseSerializer Serializer {..} =-  Serializer serialize' deserialize' deserializeEither'-  where-    serialize' = fmap serialize-    deserialize' = traverse deserialize-    deserializeEither' = traverse deserializeEither---- | A 'Serializer' for aeson 'Value's.-jsonSerializer :: (ToJSON a, FromJSON a) => Serializer a Value-jsonSerializer =-  Serializer-    { serialize = toJSON,-      deserialize = \x ->-        case fromJSON x of-          Success a -> Just a-          Error _ -> Nothing,-      deserializeEither = \x ->-        case fromJSON x of-          Success a -> Right a-          Error e -> Left e-    }---- | A 'Serializer' to convert JSON to/from lazy text. Useful for Sql event--- stores that store JSON values as text.-jsonTextSerializer :: (ToJSON a, FromJSON a) => Serializer a TL.Text-jsonTextSerializer =-  Serializer-    { serialize = TLE.decodeUtf8 . encode,-      deserialize = decode . TLE.encodeUtf8,-      deserializeEither = eitherDecode . TLE.encodeUtf8-    }---- | A 'Serializer' for 'Dynamic' values using 'toDyn' and 'fromDynamic'.-dynamicSerializer :: (Typeable a) => Serializer a Dynamic-dynamicSerializer = simpleSerializer toDyn fromDynamic---- | A 'Serializer' from one 'EventSumType' instance to another. WARNING: If--- not all events in the source 'EventSumType' are in the @serialized@--- 'EventSumType', then this function will be partial!-eventSumTypeSerializer :: (Typeable a, EventSumType a, EventSumType b) => Serializer a b-eventSumTypeSerializer = simpleSerializer serialize' deserialize'-  where-    serialize' event =-      fromMaybe-        (error $ "Failure in eventSumTypeSerializer. Can't serialize " ++ show (typeOf event))-        (eventFromDyn $ eventToDyn event)-    deserialize' = eventFromDyn . eventToDyn---- | This is a type class for serializing sum types of events to 'Dynamic'--- without the associated constructor. This is useful when transforming between--- two sum types of events. A common pattern is to put all the events in an--- application in one big event sum type, and then have a smaller sum type for--- each 'Projection'. Then, you can use 'eventSumTypeSerializer' to transform--- between the two.------ It is meant to be derived with 'Generic'. For example:------ @---    data EventA = EventA deriving (Show)---    data EventB = EventB deriving (Show)---    data EventC = EventC deriving (Show)------    data AllEvents---      = AllEventsEventA EventA---      | AllEventsEventB EventB---      | AllEventsEventC EventC---      deriving (Show, Generic)------    instance EventSumType AllEvents------    data MyEvents---      = MyEventsEventA EventA---      | MyEventsEventB EventB---      deriving (Show, Generic)------    instance EventSumType MyEvents--- @------ Now we can serialize to 'Dynamic' without a constructor tag:------ >>> eventToDyn (MyEventsEventA EventA)--- <<EventA>>------ We can also go from a 'MyEvents' value to an 'AllEvents' value:------ >>> eventFromDyn (eventToDyn (MyEventsEventA EventA)) :: Maybe AllEvents--- Just (AllEventsEventA EventA)-class EventSumType a where-  -- | Convert an event to a 'Dynamic' without the constructor tag-  eventToDyn :: a -> Dynamic--  -- | Go from a 'Dynamic' to an event with the constructor tag. Note, this-  -- function is @O(n)@ to the number of constructors.-  eventFromDyn :: Dynamic -> Maybe a--  default eventToDyn :: (Generic a, EventSumType' (Rep a)) => a -> Dynamic-  eventToDyn x = eventToDyn' (from x)--  default eventFromDyn :: (Generic a, EventSumType' (Rep a)) => Dynamic -> Maybe a-  eventFromDyn = fmap to . eventFromDyn'---- Auxiliary type class for 'EventSumType' Generic fun-class EventSumType' f where-  eventToDyn' :: f p -> Dynamic-  eventFromDyn' :: Dynamic -> Maybe (f p)---- M1 is the top-level metadata. We don't need the metadata so we just pass on--- through.-instance (EventSumType' f) => EventSumType' (M1 i t f) where-  eventToDyn' (M1 x) = eventToDyn' x-  eventFromDyn' = fmap M1 . eventFromDyn'---- The :+: operator is for when a type has multiple constructors. When--- serializing, we just pass on through. When deserializing, we try the first--- constructor, and if that fails then the second.-instance (EventSumType' f, EventSumType' g) => EventSumType' (f :+: g) where-  eventToDyn' (L1 x) = eventToDyn' x-  eventToDyn' (R1 x) = eventToDyn' x-  eventFromDyn' dyn = (L1 <$> eventFromDyn' dyn) <|> (R1 <$> eventFromDyn' dyn)---- K1 R represents an actual constructor. This is where we do the actual--- conversion to/from 'Dynamic'.-instance (Typeable c) => EventSumType' (K1 R c) where-  eventToDyn' (K1 x) = toDyn x-  eventFromDyn' dyn = K1 <$> fromDynamic dyn

@@ -1,7 +1,4 @@-{-# LANGUAGE DeriveTraversable #-}-{-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE RankNTypes #-}-{-# LANGUAGE RecordWildCards #-}  module Eventium.Store.Class   ( -- * EventStore@@ -10,39 +7,41 @@     VersionedEventStoreReader,     GlobalEventStoreReader,     VersionedEventStoreWriter,-    StreamEvent (..),-    VersionedStreamEvent,-    GlobalStreamEvent,-    ExpectedPosition (..),-    EventWriteError (..),     runEventStoreReaderUsing,     runEventStoreWriterUsing,     module Eventium.Store.Queries,+    module Eventium.Store.Types, -    -- * Serialization-    serializedEventStoreReader,-    serializedVersionedEventStoreReader,-    serializedGlobalEventStoreReader,-    serializedEventStoreWriter,+    -- * Codec+    codecEventStoreReader,+    lenientCodecEventStoreReader,+    codecVersionedEventStoreReader,+    codecGlobalEventStoreReader,+    codecEventStoreWriter,+    metadataEnrichingEventStoreWriter,+    tagEvents, -    -- * Utility types-    EventVersion (..),-    SequenceNumber (..),+    -- * Type embedding+    embeddedEventStoreWriter,      -- * Utility functions     transactionalExpectedWriteHelper,   ) where -import Data.Aeson+import Control.Exception (throw)+import Control.Monad.IO.Class (MonadIO, liftIO) import Data.Functor ((<&>)) import Data.Functor.Contravariant import Data.Maybe (mapMaybe)-import Eventium.Serializer+import qualified Data.Text as T+import Data.Time (UTCTime, getCurrentTime)+import Data.Typeable (Typeable, typeOf)+import Eventium.Codec import Eventium.Store.Queries+import Eventium.Store.Types+import Eventium.TypeEmbedding import Eventium.UUID-import Web.HttpApiData-import Web.PathPieces  -- | An 'EventStoreReader' is a function to query a stream from an event store. -- It operates in some monad @m@ and returns events of type @event@ from a@@ -62,42 +61,10 @@   = EventStoreWriter {storeEvents :: key -> ExpectedPosition position -> [event] -> m (Either (EventWriteError position) EventVersion)}  instance Contravariant (EventStoreWriter key position m) where-  contramap f (EventStoreWriter writer) = EventStoreWriter $ \vers uuid -> writer vers uuid . fmap f+  contramap f (EventStoreWriter writer) = EventStoreWriter $ \key expectedPos -> writer key expectedPos . fmap f  type VersionedEventStoreWriter = EventStoreWriter UUID EventVersion --- | An event along with the @key@ for the event stream it is from and its--- @position@ in that event stream.-data StreamEvent key position event-  = StreamEvent-  { streamEventKey :: !key,-    streamEventPosition :: !position,-    streamEventEvent :: !event-  }-  deriving (Show, Eq, Functor, Foldable, Traversable)--type VersionedStreamEvent event = StreamEvent UUID EventVersion event--type GlobalStreamEvent event = StreamEvent () SequenceNumber (VersionedStreamEvent event)---- | ExpectedPosition is used to assert the event stream is at a certain--- position. This is used when multiple writers are concurrently writing to the--- event store. If the expected position is incorrect, then storing fails.-data ExpectedPosition position-  = -- | Used when the writer doesn't care what position the stream is at.-    AnyPosition-  | -- | The stream shouldn't exist yet.-    NoStream-  | -- | The stream should already exist.-    StreamExists-  | -- | Used to assert the stream is at a particular position.-    ExactPosition position-  deriving (Show, Eq)--newtype EventWriteError position-  = EventStreamNotAtExpectedVersion position-  deriving (Show, Eq)- -- | Helper to create 'storeEventsRaw' given a function to get the latest -- stream version and a function to write to the event store. **NOTE**: This -- only works if the monad @m@ is transactional.@@ -136,7 +103,6 @@ -- | Changes the monad an 'EventStoreReader' runs in. This is useful to run -- event stores in another 'Monad' while forgetting the original 'Monad'. runEventStoreReaderUsing ::-  (Monad m, Monad mstore) =>   (forall a. mstore a -> m a) ->   EventStoreReader key position mstore event ->   EventStoreReader key position m event@@ -144,76 +110,115 @@  -- | Analog of 'runEventStoreReaderUsing' for a 'EventStoreWriter'. runEventStoreWriterUsing ::-  (Monad m, Monad mstore) =>   (forall a. mstore a -> m a) ->-  EventStoreWriter key posirion mstore event ->-  EventStoreWriter key posirion m event+  EventStoreWriter key position mstore event ->+  EventStoreWriter key position m event runEventStoreWriterUsing runStore (EventStoreWriter f) =-  EventStoreWriter $ \vers uuid events -> runStore $ f vers uuid events+  EventStoreWriter $ \key expectedPos events -> runStore $ f key expectedPos events --- | Wraps an 'EventStoreReader' and transparently serializes/deserializes--- events for you. Note that in this implementation deserialization errors are--- simply ignored (the event is not returned).-serializedEventStoreReader ::+-- | Wraps an 'EventStoreReader' and transparently decodes events.+-- Throws 'DecodeError' if any event fails to decode.+-- Use 'lenientCodecEventStoreReader' to silently skip failures.+codecEventStoreReader ::   (Monad m) =>-  Serializer event serialized ->-  EventStoreReader key position m serialized ->+  Codec event encoded ->+  EventStoreReader key position m encoded ->   EventStoreReader key position m event-serializedEventStoreReader Serializer {..} (EventStoreReader reader) =-  EventStoreReader $ fmap (mapMaybe deserialize) . reader+codecEventStoreReader codec (EventStoreReader reader) =+  EventStoreReader $ fmap (map strictDecode) . reader+  where+    strictDecode s = case codec.decode s of+      Just a -> a+      Nothing -> throw $ DecodeError "codecEventStoreReader" "Failed to decode event" --- | Convenience wrapper around 'serializedEventStoreReader' for+-- | Like 'codecEventStoreReader' but silently drops events that fail+-- to decode.+--+-- Recommended for production use when using sum-type codecs, as it+-- allows readers to gracefully handle events they don't recognize.+lenientCodecEventStoreReader ::+  (Monad m) =>+  Codec event encoded ->+  EventStoreReader key position m encoded ->+  EventStoreReader key position m event+lenientCodecEventStoreReader codec (EventStoreReader reader) =+  EventStoreReader $ fmap (mapMaybe codec.decode) . reader++-- | Convenience wrapper around 'codecEventStoreReader' for -- 'VersionedEventStoreReader'.-serializedVersionedEventStoreReader ::+codecVersionedEventStoreReader ::   (Monad m) =>-  Serializer event serialized ->-  VersionedEventStoreReader m serialized ->+  Codec event encoded ->+  VersionedEventStoreReader m encoded ->   VersionedEventStoreReader m event-serializedVersionedEventStoreReader serializer = serializedEventStoreReader (traverseSerializer serializer)+codecVersionedEventStoreReader codec = codecEventStoreReader (traverseCodec codec) --- | Convenience wrapper around 'serializedEventStoreReader' for+-- | Convenience wrapper around 'codecEventStoreReader' for -- 'GlobalEventStoreReader'.-serializedGlobalEventStoreReader ::+codecGlobalEventStoreReader ::   (Monad m) =>-  Serializer event serialized ->-  GlobalEventStoreReader m serialized ->+  Codec event encoded ->+  GlobalEventStoreReader m encoded ->   GlobalEventStoreReader m event-serializedGlobalEventStoreReader serializer = serializedEventStoreReader (traverseSerializer (traverseSerializer serializer))+codecGlobalEventStoreReader codec = codecEventStoreReader (traverseCodec (traverseCodec codec)) --- | Like 'serializedEventStoreReader' but for an 'EventStoreWriter'. Note that+-- | Like 'codecEventStoreReader' but for an 'EventStoreWriter'. Note that -- 'EventStoreWriter' is an instance of 'Contravariant', so you can just use--- @contramap serialize@ instead of this function.-serializedEventStoreWriter ::-  (Monad m) =>-  Serializer event serialized ->-  EventStoreWriter key position m serialized ->+-- @contramap (encode codec)@ instead of this function.+codecEventStoreWriter ::+  Codec event encoded ->+  EventStoreWriter key position m encoded ->   EventStoreWriter key position m event-serializedEventStoreWriter Serializer {..} = contramap serialize+codecEventStoreWriter codec = contramap codec.encode --- | Event versions are a strictly increasing series of integers for each--- projection. They allow us to order the events when they are replayed, and--- they also help as a concurrency check in a multi-threaded environment so--- services modifying the projection can be sure the projection didn't change--- during their execution.-newtype EventVersion = EventVersion {unEventVersion :: Int}-  deriving (Show, Read, Ord, Eq, Enum, Num, FromJSON, ToJSON)+-- | 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 ::+  (MonadIO m, Typeable event) =>+  Codec event encoded ->+  EventStoreWriter key position m (TaggedEvent encoded) ->+  EventStoreWriter key position m event+metadataEnrichingEventStoreWriter 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))+                  (codec.encode e)+            )+            events+    write key pos tagged --- | The sequence number gives us a global ordering of events in a particular--- event store. Using sequence numbers is not strictly necessary for an event--- sourcing and CQRS system, but it makes it way easier to replay events--- consistently without having to use distributed transactions in an event bus.--- In SQL-based event stores, they are also very cheap to create.-newtype SequenceNumber = SequenceNumber {unSequenceNumber :: Int}-  deriving-    ( Show,-      Read,-      Ord,-      Eq,-      Enum,-      Num,-      FromJSON,-      ToJSON,-      PathPiece,-      ToHttpApiData,-      FromHttpApiData-    )+-- | Tag events with metadata in a pure context. The caller supplies+-- the current time. Useful when 'MonadIO' is not available.+tagEvents ::+  (Typeable event) =>+  Codec event encoded ->+  UTCTime ->+  [event] ->+  [TaggedEvent encoded]+tagEvents codec now =+  map $ \e ->+    TaggedEvent+      (EventMetadata (T.pack . show $ typeOf e) Nothing Nothing (Just now))+      (codec.encode e)++-- | Like 'codecEventStoreWriter' but uses a 'TypeEmbedding' instead of+-- a 'Codec'. Intended for embedding aggregate-specific event types into+-- an application-wide event sum type before writing to the store.+embeddedEventStoreWriter ::+  TypeEmbedding event adapted ->+  EventStoreWriter key position m adapted ->+  EventStoreWriter key position m event+embeddedEventStoreWriter emb = contramap emb.embed

@@ -17,9 +17,9 @@ -- and the start/stop points for the query. data QueryRange key position   = QueryRange-  { queryRangeKey :: key,-    queryRangeStart :: QueryStart position,-    queryRangeLimit :: QueryLimit position+  { key :: key,+    start :: QueryStart position,+    limit :: QueryLimit position   }   deriving (Show, Eq) 

@@ -0,0 +1,127 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DeriveTraversable #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}++-- | Pure data types for the event store. These types have no dependency on+-- any monad or effect system. They are extracted from "Eventium.Store.Class"+-- so that pure modules (like "Eventium.Projection") can reference them without+-- pulling in effectful store interfaces.+module Eventium.Store.Types+  ( -- * Stream events+    StreamEvent (..),+    VersionedStreamEvent,+    GlobalStreamEvent,++    -- * Event metadata+    EventMetadata (..),+    emptyMetadata,+    TaggedEvent (..),++    -- * Expected position+    ExpectedPosition (..),+    EventWriteError (..),++    -- * Utility types+    EventVersion (..),+    SequenceNumber (..),+  )+where++import Data.Aeson+import Data.Text (Text)+import Data.Time (UTCTime)+import Eventium.UUID+import GHC.Generics (Generic)+import Web.HttpApiData+import Web.PathPieces++-- | Metadata carried alongside every stored event.+data EventMetadata = EventMetadata+  { eventType :: !Text,+    correlationId :: !(Maybe UUID),+    causationId :: !(Maybe UUID),+    createdAt :: !(Maybe UTCTime)+  }+  deriving (Show, Eq, Generic)++instance ToJSON EventMetadata where+  toJSON = genericToJSON defaultOptions+  toEncoding = genericToEncoding defaultOptions++instance FromJSON EventMetadata where+  parseJSON = genericParseJSON defaultOptions++-- | Construct 'EventMetadata' with only an event type name.+emptyMetadata :: Text -> EventMetadata+emptyMetadata et = EventMetadata et Nothing Nothing Nothing++-- | An event paired with pre-computed metadata. Used to thread metadata+-- through the 'EventStoreWriter' interface without changing its type signature.+-- Store backends that accept @TaggedEvent@ will use the attached metadata+-- instead of generating empty metadata.+data TaggedEvent event = TaggedEvent+  { metadata :: !EventMetadata,+    payload :: !event+  }+  deriving (Show, Eq, Functor)++-- | An event along with the @key@ for the event stream it is from, its+-- @position@ in that event stream, and 'EventMetadata'.+data StreamEvent key position event+  = StreamEvent+  { key :: !key,+    position :: !position,+    metadata :: !EventMetadata,+    payload :: !event+  }+  deriving (Show, Eq, Functor, Foldable, Traversable)++type VersionedStreamEvent event = StreamEvent UUID EventVersion event++type GlobalStreamEvent event = StreamEvent () SequenceNumber (VersionedStreamEvent event)++-- | ExpectedPosition is used to assert the event stream is at a certain+-- position. This is used when multiple writers are concurrently writing to the+-- event store. If the expected position is incorrect, then storing fails.+data ExpectedPosition position+  = -- | Used when the writer doesn't care what position the stream is at.+    AnyPosition+  | -- | The stream shouldn't exist yet.+    NoStream+  | -- | The stream should already exist.+    StreamExists+  | -- | Used to assert the stream is at a particular position.+    ExactPosition position+  deriving (Show, Eq)++newtype EventWriteError position+  = EventStreamNotAtExpectedVersion position+  deriving (Show, Eq)++-- | Event versions are a strictly increasing series of integers for each+-- projection. They allow us to order the events when they are replayed, and+-- they also help as a concurrency check in a multi-threaded environment so+-- services modifying the projection can be sure the projection didn't change+-- during their execution.+newtype EventVersion = EventVersion {unEventVersion :: Int}+  deriving (Show, Read, Ord, Eq, Enum, Num, FromJSON, ToJSON)++-- | The sequence number gives us a global ordering of events in a particular+-- event store. Using sequence numbers is not strictly necessary for an event+-- sourcing and CQRS system, but it makes it way easier to replay events+-- consistently without having to use distributed transactions in an event bus.+-- In SQL-based event stores, they are also very cheap to create.+newtype SequenceNumber = SequenceNumber {unSequenceNumber :: Int}+  deriving+    ( Show,+      Read,+      Ord,+      Eq,+      Enum,+      Num,+      FromJSON,+      ToJSON,+      PathPiece,+      ToHttpApiData,+      FromHttpApiData+    )

@@ -4,4 +4,5 @@ where  import Eventium.TH.Projection as X-import Eventium.TH.SumTypeSerializer as X+import Eventium.TH.SumType as X+import Eventium.TH.SumTypeCodec as X

@@ -8,8 +8,8 @@  import Data.Char (toLower) import Eventium.Projection+import Eventium.TH.SumType import Language.Haskell.TH-import SumTypesX.TH  -- | Creates a 'Projection' for a given type and a list of events. The user of -- this function also needs to provide event handlers for each event. For

@@ -0,0 +1,165 @@+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE TemplateHaskell #-}++module Eventium.TH.SumType+  ( -- * Constructing sum types+    constructSumType,+    SumTypeOptions,+    defaultSumTypeOptions,+    sumTypeOptions,+    withTagOptions,+    withConstructorStrictness,+    SumTypeTagOptions (..),+    SumTypeConstructorStrictness (..),++    -- * Converting between sum types+    sumTypeConverter,+    partialSumTypeConverter,+  )+where++import Language.Haskell.TH+import Prelude++-- | Options for constructing sum types.+data SumTypeOptions = SumTypeOptions+  { sumTypeOptionsTagOptions :: SumTypeTagOptions,+    sumTypeOptionsConstructorStrictness :: SumTypeConstructorStrictness+  }++-- | Default options: prefix tags with type name, lazy constructors.+defaultSumTypeOptions :: SumTypeOptions+defaultSumTypeOptions =+  SumTypeOptions+    { sumTypeOptionsTagOptions = PrefixTagsWithTypeName,+      sumTypeOptionsConstructorStrictness = LazySumTypeConstructors+    }++-- | Construct 'SumTypeOptions' from tag options and constructor strictness.+sumTypeOptions :: SumTypeTagOptions -> SumTypeConstructorStrictness -> SumTypeOptions+sumTypeOptions = SumTypeOptions++-- | Set the tag options on a 'SumTypeOptions' value.+withTagOptions :: SumTypeTagOptions -> SumTypeOptions -> SumTypeOptions+withTagOptions opts (SumTypeOptions _ s) = SumTypeOptions opts s++-- | Set the constructor strictness on a 'SumTypeOptions' value.+withConstructorStrictness :: SumTypeConstructorStrictness -> SumTypeOptions -> SumTypeOptions+withConstructorStrictness s (SumTypeOptions opts _) = SumTypeOptions opts s++-- | How to name constructors in the generated sum type.+data SumTypeTagOptions+  = PrefixTagsWithTypeName+  | AppendTypeNameToTags+  | ConstructTagName (String -> String)++-- | Whether generated constructors have strict or lazy fields.+data SumTypeConstructorStrictness+  = LazySumTypeConstructors+  | StrictSumTypeConstructors+  deriving (Show, Eq)++-- | Construct a sum type from a list of existing types.+constructSumType :: String -> SumTypeOptions -> [Name] -> Q [Dec]+constructSumType typeName SumTypeOptions {..} types = do+  let strictness = toSourceStrictness sumTypeOptionsConstructorStrictness+      mkConstructor name =+        NormalC+          (constructorName sumTypeOptionsTagOptions typeName name)+          [(Bang NoSourceUnpackedness strictness, ConT name)]+      constructors = map mkConstructor types+  return [DataD [] (mkName typeName) [] Nothing constructors []]++-- | Generate a total conversion function between two sum types.+sumTypeConverter :: String -> Name -> Name -> Q [Dec]+sumTypeConverter functionName sourceType targetType = do+  bothConstructors <- matchTypeConstructors sourceType targetType+  let funcName = mkName functionName+  funcClauses <- mapM mkSerializeFunc bothConstructors+  typeDecl <- [t|$(conT sourceType) -> $(conT targetType)|]+  return+    [ SigD funcName typeDecl,+      FunD funcName funcClauses+    ]++-- | Generate a partial conversion function between two sum types.+partialSumTypeConverter :: String -> Name -> Name -> Q [Dec]+partialSumTypeConverter functionName sourceType targetType = do+  bothConstructors <- matchTypeConstructors targetType sourceType+  let funcName = mkName functionName+      wildcardClause = Clause [WildP] (NormalB (ConE 'Nothing)) []+  funcClauses <- mapM mkDeserializeFunc bothConstructors+  typeDecl <- [t|$(conT sourceType) -> Maybe $(conT targetType)|]+  return+    [ SigD funcName typeDecl,+      FunD funcName (funcClauses ++ [wildcardClause])+    ]++-- Internal helpers++constructorName :: SumTypeTagOptions -> String -> Name -> Name+constructorName PrefixTagsWithTypeName typeName =+  mkName . (typeName ++) . nameBase+constructorName AppendTypeNameToTags typeName =+  mkName . (++ typeName) . nameBase+constructorName (ConstructTagName mkCtor) _ =+  mkName . mkCtor . nameBase++toSourceStrictness :: SumTypeConstructorStrictness -> SourceStrictness+toSourceStrictness LazySumTypeConstructors = NoSourceStrictness+toSourceStrictness StrictSumTypeConstructors = SourceStrict++data BothConstructors = BothConstructors+  { innerType :: Type,+    sourceConstructor :: Name,+    targetConstructor :: Name+  }++matchTypeConstructors :: Name -> Name -> Q [BothConstructors]+matchTypeConstructors sourceType targetType = do+  sourceConstructors <- typeConstructors sourceType+  targetConstructors <- typeConstructors targetType+  mapM (matchConstructor targetConstructors) sourceConstructors++typeConstructors :: Name -> Q [(Type, Name)]+typeConstructors typeName = do+  info <- reify typeName+  case info of+    (TyConI (DataD _ _ _ _ constructors _)) ->+      mapM go constructors+      where+        go (NormalC name []) =+          fail $ "Constructor " ++ nameBase name ++ " doesn't have any arguments"+        go (NormalC name [(_, type')]) = return (type', name)+        go (NormalC name _) =+          fail $ "Constructor " ++ nameBase name ++ " has more than one argument"+        go _ =+          fail $ "Invalid constructor in " ++ nameBase typeName+    _ -> fail $ nameBase typeName ++ " must be a sum type"++matchConstructor :: [(Type, Name)] -> (Type, Name) -> Q BothConstructors+matchConstructor targetConstructors (type', sourceConstructor) = do+  targetConstructor <-+    maybe+      ( fail $+          "Can't find constructor in target type corresponding to "+            ++ nameBase sourceConstructor+      )+      return+      (lookup type' targetConstructors)+  return $ BothConstructors type' sourceConstructor targetConstructor++mkSerializeFunc :: BothConstructors -> Q Clause+mkSerializeFunc BothConstructors {..} = do+  varName <- newName "value"+  let patternMatch = ConP sourceConstructor [] [VarP varName]+      constructor = AppE (ConE targetConstructor) (VarE varName)+  return $ Clause [patternMatch] (NormalB constructor) []++mkDeserializeFunc :: BothConstructors -> Q Clause+mkDeserializeFunc BothConstructors {..} = do+  varName <- newName "value"+  let patternMatch = ConP targetConstructor [] [VarP varName]+      constructor =+        AppE (ConE 'Just) (AppE (ConE sourceConstructor) (VarE varName))+  return $ Clause [patternMatch] (NormalB constructor) []

@@ -0,0 +1,122 @@+{-# LANGUAGE QuasiQuotes #-}+{-# LANGUAGE TemplateHaskell #-}++module Eventium.TH.SumTypeCodec+  ( mkSumTypeCodec,+    mkSumTypeEmbedding,+  )+where++import Data.Char (toLower)+import Eventium.TH.SumType+import Language.Haskell.TH++-- | This is a template haskell function that creates a 'Codec' between+-- two sum types. The first sum type must be a subset of the second sum type.+-- This is useful in situations where you define all the events in your system+-- in one type, and you want to create sum types that are subsets for each+-- 'Projection'.+--+-- For example, assume we have the following three event types and two sum+-- types holding these events:+--+-- @+--    data EventA = EventA+--    data EventB = EventB+--    data EventC = EventC+--+--    data AllEvents+--      = AllEventsEventA EventA+--      | AllEventsEventB EventB+--      | AllEventsEventC EventC+--+--    data MyEvents+--      = MyEventsEventA EventA+--      | MyEventsEventB EventB+-- @+--+-- In this case, @AllEvents@ holds all the events in our system, and @MyEvents@+-- holds some subset of @AllEvents@. If we run+--+-- @+--    mkSumTypeCodec "myEventsCodec" ''MyEvents ''AllEvents+-- @+--+-- we will produce the following code:+--+-- @+--    -- Encoding function+--    myEventsToAllEvents :: MyEvents -> AllEvents+--    myEventsToAllEvents (MyEventsEventA e) = AllEventsEventA e+--    myEventsToAllEvents (MyEventsEventB e) = AllEventsEventB e+--+--    -- Decoding function+--    allEventsToMyEvents :: AllEvents -> Maybe MyEvents+--    allEventsToMyEvents (AllEventsEventA e) = Just (MyEventsEventA e)+--    allEventsToMyEvents (AllEventsEventB e) = Just (MyEventsEventB e)+--    allEventsToMyEvents _ = Nothing+--+--    -- Codec+--    myEventsCodec :: Codec MyEvents AllEvents+--    myEventsCodec = Codec myEventsToAllEvents allEventsToMyEvents+-- @+mkSumTypeCodec :: String -> Name -> Name -> Q [Dec]+mkSumTypeCodec codecName sourceType targetType = do+  -- Construct the encoding/decoding functions+  let encodeFuncName = firstCharToLower (nameBase sourceType) ++ "To" ++ nameBase targetType+      decodeFuncName = firstCharToLower (nameBase targetType) ++ "To" ++ nameBase sourceType+  -- Generate the sum type converter functions+  encodeDecls <- sumTypeConverter encodeFuncName sourceType targetType+  decodeDecls <- partialSumTypeConverter decodeFuncName targetType sourceType++  -- Construct the codec+  codecTypeDecl <- [t|$(conT $ mkName "Codec") $(conT sourceType) $(conT targetType)|]+  codecExp <- [e|$(conE $ mkName "Codec") $(varE $ mkName encodeFuncName) $(varE $ mkName decodeFuncName)|]+  let codecClause = Clause [] (NormalB codecExp) []++  return $+    [ SigD (mkName codecName) codecTypeDecl,+      FunD (mkName codecName) [codecClause]+    ]+      ++ encodeDecls+      ++ decodeDecls++-- | Like 'mkSumTypeCodec' but generates a 'TypeEmbedding' instead of a+-- 'Codec'. Preferred when the purpose is to embed one sum type into+-- another rather than wire-format encoding.+--+-- For example:+--+-- @+--    mkSumTypeEmbedding "myEventsEmbedding" ''MyEvents ''AllEvents+-- @+--+-- produces:+--+-- @+--    myEventsToAllEvents :: MyEvents -> AllEvents+--    allEventsToMyEvents :: AllEvents -> Maybe MyEvents+--    myEventsEmbedding :: TypeEmbedding MyEvents AllEvents+--    myEventsEmbedding = TypeEmbedding myEventsToAllEvents allEventsToMyEvents+-- @+mkSumTypeEmbedding :: String -> Name -> Name -> Q [Dec]+mkSumTypeEmbedding embeddingName sourceType targetType = do+  let embedFuncName = firstCharToLower (nameBase sourceType) ++ "To" ++ nameBase targetType+      extractFuncName = firstCharToLower (nameBase targetType) ++ "To" ++ nameBase sourceType+  embedDecls <- sumTypeConverter embedFuncName sourceType targetType+  extractDecls <- partialSumTypeConverter extractFuncName targetType sourceType++  embeddingTypeDecl <- [t|$(conT $ mkName "TypeEmbedding") $(conT sourceType) $(conT targetType)|]+  embeddingExp <- [e|$(conE $ mkName "TypeEmbedding") $(varE $ mkName embedFuncName) $(varE $ mkName extractFuncName)|]+  let embeddingClause = Clause [] (NormalB embeddingExp) []++  return $+    [ SigD (mkName embeddingName) embeddingTypeDecl,+      FunD (mkName embeddingName) [embeddingClause]+    ]+      ++ embedDecls+      ++ extractDecls++firstCharToLower :: String -> String+firstCharToLower [] = []+firstCharToLower (x : xs) = toLower x : xs

@@ -1,86 +0,0 @@-{-# LANGUAGE QuasiQuotes #-}-{-# LANGUAGE RecordWildCards #-}-{-# LANGUAGE TemplateHaskell #-}--module Eventium.TH.SumTypeSerializer-  ( mkSumTypeSerializer,-  )-where--import Data.Char (toLower)-import Language.Haskell.TH-import SumTypesX.TH---- | This is a template haskell function that creates a 'Serializer' between--- two sum types. The first sum type must be a subset of the second sum type.--- This is useful in situations where you define all the events in your system--- in one type, and you want to create sum types that are subsets for each--- 'Projection'.------ For example, assume we have the following three event types and two sum--- types holding these events:------ @---    data EventA = EventA---    data EventB = EventB---    data EventC = EventC------    data AllEvents---      = AllEventsEventA EventA---      | AllEventsEventB EventB---      | AllEventsEventC EventC------    data MyEvents---      = MyEventsEventA EventA---      | MyEventsEventB EventB--- @------ In this case, @AllEvents@ holds all the events in our system, and @MyEvents@--- holds some subset of @AllEvents@. If we run------ @---    mkSumTypeSerializer "myEventsSerializer" ''MyEvents ''AllEvents--- @------ we will produce the following code:------ @---    -- Serialization function---    myEventsToAllEvents :: MyEvents -> AllEvents---    myEventsToAllEvents (MyEventsEventA e) = AllEventsEventA e---    myEventsToAllEvents (MyEventsEventB e) = AllEventsEventB e------    -- Deserialization function---    allEventsToMyEvents :: AllEvents -> Maybe MyEvents---    allEventsToMyEvents (AllEventsEventA e) = Just (MyEventsEventA e)---    allEventsToMyEvents (AllEventsEventB e) = Just (MyEventsEventB e)---    allEventsToMyEvents _ = Nothing------    -- Serializer---    myEventsSerializer :: Serializer MyEvents AllEvents---    myEventsSerializer = simpleSerializer myEventsToAllEvents allEventsToMyEvents--- @-mkSumTypeSerializer :: String -> Name -> Name -> Q [Dec]-mkSumTypeSerializer serializerName sourceType targetType = do-  -- Construct the serialization function-  let serializeFuncName = firstCharToLower (nameBase sourceType) ++ "To" ++ nameBase targetType-      deserializeFuncName = firstCharToLower (nameBase targetType) ++ "To" ++ nameBase sourceType-  -- Generate the sum type converter functions-  serializeDecls <- sumTypeConverter serializeFuncName sourceType targetType-  deserializeDecls <- partialSumTypeConverter deserializeFuncName targetType sourceType--  -- Construct the serializer-  serializerTypeDecl <- [t|$(conT $ mkName "Serializer") $(conT sourceType) $(conT targetType)|]-  serializerExp <- [e|$(varE $ mkName "simpleSerializer") $(varE $ mkName serializeFuncName) $(varE $ mkName deserializeFuncName)|]-  let serializerClause = Clause [] (NormalB serializerExp) []--  return $-    [ SigD (mkName serializerName) serializerTypeDecl,-      FunD (mkName serializerName) [serializerClause]-    ]-      ++ serializeDecls-      ++ deserializeDecls--firstCharToLower :: String -> String-firstCharToLower [] = []-firstCharToLower (x : xs) = toLower x : xs

@@ -0,0 +1,49 @@+-- | Defines the 'TypeEmbedding' type for embedding one sum type into another.+--+-- A 'TypeEmbedding' represents a total injection from type @a@ into type @b@+-- and a partial extraction back. This is used when @a@ is conceptually a+-- subset of @b@ — for example, when embedding a per-aggregate event sum type+-- into an application-wide event sum type.+--+-- Unlike 'Eventium.Codec.Codec', a 'TypeEmbedding' carries no+-- event-type-name metadata and is not intended for wire-format conversion.+module Eventium.TypeEmbedding+  ( TypeEmbedding (..),+    composeEmbeddings,+    idEmbedding,+    embeddingToCodec,+  )+where++import Control.Monad+import Eventium.Codec (Codec (..))++-- | A 'TypeEmbedding' describes a total injection from type @a@ into type @b@+-- and a partial extraction back. This is used when @a@ is conceptually a+-- subset of @b@.+--+-- For example, @TypeEmbedding AccountEvent BankEvent@ embeds account-specific+-- events into the application-wide event sum type.+data TypeEmbedding a b = TypeEmbedding+  { -- | Total injection: every @a@ can be embedded into @b@.+    embed :: a -> b,+    -- | Partial extraction: a @b@ value may or may not contain an @a@.+    extract :: b -> Maybe a+  }++-- | Compose two 'TypeEmbedding's: @a@ embeds into @b@, @b@ embeds into @c@.+composeEmbeddings :: TypeEmbedding a b -> TypeEmbedding b c -> TypeEmbedding a c+composeEmbeddings emb1 emb2 =+  TypeEmbedding+    { embed = emb2.embed . emb1.embed,+      extract = emb2.extract >=> emb1.extract+    }++-- | Identity embedding where no type conversion is needed.+idEmbedding :: TypeEmbedding a a+idEmbedding = TypeEmbedding id Just++-- | Convert a 'TypeEmbedding' to a 'Codec' for interoperability with+-- APIs that require a 'Codec'.+embeddingToCodec :: TypeEmbedding a b -> Codec a b+embeddingToCodec (TypeEmbedding e x) = Codec e x

@@ -49,6 +49,8 @@   fromPathPiece = uuidFromText   toPathPiece = uuidToText +-- TODO: move to testing package+ -- | Constructs a valid 'UUID' from an 'Integer' by padding with zeros. Useful -- for testing. --

@@ -0,0 +1,81 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE TemplateHaskell #-}++module Eventium.CodecSpec (spec) where++import Data.Dynamic+import Data.Typeable (typeOf)+import Eventium.Codec+import Eventium.TH.SumTypeCodec+import Eventium.TypeEmbedding+import GHC.Generics+import Test.Hspec++data EventA = EventA deriving (Show, Eq)++data EventB = EventB deriving (Show, Eq)++data EventC = EventC deriving (Show, Eq)++data AllEvents+  = AllEventsEventA EventA+  | AllEventsEventB EventB+  | AllEventsEventC EventC+  deriving (Show, Eq, Generic)++instance EventSumType AllEvents++data MyEvents+  = MyEventsEventA EventA+  | MyEventsEventB EventB+  deriving (Show, Eq, Generic)++instance EventSumType MyEvents++mkSumTypeCodec "myEventsCodec" ''MyEvents ''AllEvents++-- Separate subset type for TypeEmbedding tests to avoid TH name clashes+-- (mkSumTypeCodec and mkSumTypeEmbedding generate the same helper+-- function names for the same type pair)+data SubsetEvents+  = SubsetEventsEventA EventA+  | SubsetEventsEventB EventB+  deriving (Show, Eq)++mkSumTypeEmbedding "subsetEventsEmbedding" ''SubsetEvents ''AllEvents++spec :: Spec+spec = do+  describe "EventSumType" $ do+    it "can serialize events without the constructor" $ do+      dynTypeRep (eventToDyn $ MyEventsEventA EventA) `shouldBe` typeOf EventA+      dynTypeRep (eventToDyn $ MyEventsEventA EventA) `shouldBe` dynTypeRep (eventToDyn $ AllEventsEventA EventA)++    it "can deserialize events with the constructor" $ do+      eventFromDyn (toDyn EventA) `shouldBe` Just (MyEventsEventA EventA)+      eventFromDyn (toDyn EventB) `shouldBe` Just (AllEventsEventB EventB)++      eventFromDyn (eventToDyn $ MyEventsEventA EventA) `shouldBe` Just (AllEventsEventA EventA)+      eventFromDyn (eventToDyn $ AllEventsEventB EventB) `shouldBe` Just (MyEventsEventB EventB)++  describe "mkSumTypeCodec" $ do+    it "can serialize events" $ do+      myEventsCodec.encode (MyEventsEventA EventA) `shouldBe` AllEventsEventA EventA+      myEventsCodec.encode (MyEventsEventB EventB) `shouldBe` AllEventsEventB EventB++    it "can deserialize events" $ do+      myEventsCodec.decode (AllEventsEventA EventA) `shouldBe` Just (MyEventsEventA EventA)+      myEventsCodec.decode (AllEventsEventB EventB) `shouldBe` Just (MyEventsEventB EventB)+      myEventsCodec.decode (AllEventsEventC EventC) `shouldBe` Nothing++  describe "mkSumTypeEmbedding" $ do+    it "can embed events" $ do+      subsetEventsEmbedding.embed (SubsetEventsEventA EventA) `shouldBe` AllEventsEventA EventA+      subsetEventsEmbedding.embed (SubsetEventsEventB EventB) `shouldBe` AllEventsEventB EventB++    it "can extract matching events" $ do+      subsetEventsEmbedding.extract (AllEventsEventA EventA) `shouldBe` Just (SubsetEventsEventA EventA)+      subsetEventsEmbedding.extract (AllEventsEventB EventB) `shouldBe` Just (SubsetEventsEventB EventB)++    it "returns Nothing for non-matching events" $ do+      subsetEventsEmbedding.extract (AllEventsEventC EventC) `shouldBe` Nothing

@@ -0,0 +1,86 @@+{-# LANGUAGE OverloadedStrings #-}++module Eventium.CommandDispatcherSpec (spec) where++import Data.IORef+import Eventium.CommandDispatcher+import Eventium.CommandHandler+import Eventium.ProcessManager (CommandDispatchResult (..), CommandDispatcher (..), RejectionReason (..))+import Eventium.Projection+import Eventium.Store.Class+import Eventium.UUID+import Test.Hspec++-- Minimal test domain+data CounterEvent = Incremented | Decremented deriving (Show, Eq)++data CounterCommand = Increment | Decrement | Unknown deriving (Show, Eq)++data CounterError = AlreadyZero deriving (Show, Eq)++type Counter = Int++counterProjection :: Projection Counter CounterEvent+counterProjection =+  Projection 0 $ \s e -> case e of+    Incremented -> s + 1+    Decremented -> s - 1++counterHandler :: CommandHandler Counter CounterEvent CounterCommand CounterError+counterHandler = CommandHandler decide' counterProjection+  where+    decide' _ Increment = Right [Incremented]+    decide' s Decrement+      | s <= 0 = Left AlreadyZero+      | otherwise = Right [Decremented]+    decide' _ Unknown = Right []++-- | Simple IORef-based event store for testing.+mkTestStore :: IO (VersionedEventStoreWriter IO CounterEvent, VersionedEventStoreReader IO CounterEvent)+mkTestStore = do+  eventsRef <- newIORef ([] :: [VersionedStreamEvent CounterEvent])+  let writer = EventStoreWriter $ \uuid _expected events -> do+        existing <- readIORef eventsRef+        let 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)+  where+    filterByQuery (QueryRange uuid _ _) =+      filter (\(StreamEvent k _ _ _) -> k == uuid)++spec :: Spec+spec = describe "CommandDispatcher" $ do+  describe "commandHandlerDispatcher" $ do+    it "routes command to matching handler and reports success" $ do+      (writer, reader) <- mkTestStore++      let handlers = [mkAggregateHandler counterHandler]+          dispatcher = commandHandlerDispatcher writer reader handlers++      result <- dispatcher.dispatchCommand (uuidFromInteger 1) Increment+      result `shouldBe` CommandSucceeded++    it "reports failure when command is rejected" $ do+      (writer, reader) <- mkTestStore++      let handlers = [mkAggregateHandler counterHandler]+          dispatcher = commandHandlerDispatcher writer reader handlers++      -- Counter starts at 0, Decrement should fail+      result <- dispatcher.dispatchCommand (uuidFromInteger 1) Decrement+      result `shouldBe` CommandFailed (RejectionReason "AlreadyZero")++    it "returns CommandSucceeded when no handler matches" $ do+      (writer, reader) <- mkTestStore++      let handlers = [mkAggregateHandler counterHandler]+          dispatcher = commandHandlerDispatcher writer reader handlers++      -- Unknown returns Right [], so no handler "matches" (produces events)+      result <- dispatcher.dispatchCommand (uuidFromInteger 1) Unknown+      result `shouldBe` CommandSucceeded

@@ -0,0 +1,100 @@+module Eventium.EventHandlerSpec (spec) where++import Data.Functor.Contravariant+import Data.IORef+import Eventium.Codec (Codec (..), DecodeError (..))+import Eventium.EventHandler+import Test.Hspec++-- Helper: create an EventHandler that appends events to an IORef+recordingHandler :: IORef [a] -> EventHandler IO a+recordingHandler ref = EventHandler $ \e -> modifyIORef ref (++ [e])++spec :: Spec+spec = do+  describe "handleEvent" $ do+    it "should process a single event" $ do+      ref <- newIORef ([] :: [Int])+      (recordingHandler ref).handleEvent 42+      readIORef ref `shouldReturn` [42]++  describe "handleEvents" $ do+    it "should process events in order" $ do+      ref <- newIORef ([] :: [Int])+      handleEvents (recordingHandler ref) [1, 2, 3]+      readIORef ref `shouldReturn` [1, 2, 3]++    it "should do nothing for empty list" $ do+      ref <- newIORef ([] :: [Int])+      handleEvents (recordingHandler ref) []+      readIORef ref `shouldReturn` []++  describe "Contravariant instance" $ do+    it "should adapt the event type via contramap" $ do+      ref <- newIORef ([] :: [Int])+      let handler = contramap length (recordingHandler ref)+      handler.handleEvent "hello"+      handler.handleEvent "hi"+      readIORef ref `shouldReturn` [5, 2]++  describe "Semigroup instance" $ do+    it "should run both handlers for each event" $ do+      ref1 <- newIORef ([] :: [Int])+      ref2 <- newIORef ([] :: [Int])+      let combined = recordingHandler ref1 <> recordingHandler ref2+      combined.handleEvent 10+      combined.handleEvent 20+      readIORef ref1 `shouldReturn` [10, 20]+      readIORef ref2 `shouldReturn` [10, 20]++  describe "Monoid instance" $ do+    it "mempty should be a no-op" $ do+      (mempty :: EventHandler IO Int).handleEvent 42+    -- No crash, no observable effect++    it "mconcat should compose multiple handlers" $ do+      ref1 <- newIORef ([] :: [Int])+      ref2 <- newIORef ([] :: [Int])+      ref3 <- newIORef ([] :: [Int])+      let combined = mconcat [recordingHandler ref1, recordingHandler ref2, recordingHandler ref3]+      combined.handleEvent 99+      readIORef ref1 `shouldReturn` [99]+      readIORef ref2 `shouldReturn` [99]+      readIORef ref3 `shouldReturn` [99]++  describe "eventHandlerMapMaybe" $ do+    it "should only handle events matching the filter" $ do+      ref <- newIORef ([] :: [Int])+      let evenOnly n = if even n then Just n else Nothing+          handler = eventHandlerMapMaybe evenOnly (recordingHandler ref)+      handleEvents handler [1, 2, 3, 4, 5, 6]+      readIORef ref `shouldReturn` [2, 4, 6]++    it "should handle no events when filter matches nothing" $ do+      ref <- newIORef ([] :: [Int])+      let noneMatch _ = Nothing :: Maybe Int+          handler = eventHandlerMapMaybe noneMatch (recordingHandler ref)+      handleEvents handler [1, 2, 3]+      readIORef ref `shouldReturn` []++  describe "codecEventHandler" $ do+    it "should deserialize events before handling" $ do+      ref <- newIORef ([] :: [Int])+      let codec = Codec show (\s -> Just (read s :: Int))+          handler = codecEventHandler codec (recordingHandler ref)+      handleEvents handler ["1", "2", "3"]+      readIORef ref `shouldReturn` [1, 2, 3]++    it "should error on deserialization failure" $ do+      ref <- newIORef ([] :: [Int])+      let codec = Codec show (\s -> if s == "bad" then Nothing else Just (read s :: Int))+          handler = codecEventHandler codec (recordingHandler ref)+      handleEvents handler ["1", "bad", "3"] `shouldThrow` (\(DecodeError ctx _) -> ctx == "codecEventHandler")++  describe "lenientCodecEventHandler" $ do+    it "should silently drop events that fail to deserialize" $ do+      ref <- newIORef ([] :: [Int])+      let codec = Codec show (\s -> if s == "bad" then Nothing else Just (read s :: Int))+          handler = lenientCodecEventHandler codec (recordingHandler ref)+      handleEvents handler ["1", "bad", "3"]+      readIORef ref `shouldReturn` [1, 3]

@@ -0,0 +1,69 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE TemplateHaskell #-}++module Eventium.JsonSpec (spec) where++import Data.Aeson+import qualified Data.Aeson.KeyMap as KM+import Eventium.Json (dropPrefix, dropSuffix, unPrefixLower)+import Eventium.Json.TH (deriveJSONUnPrefixLower)+import GHC.Generics (Generic)+import Test.Hspec++data SampleRecord = SampleRecord+  { sampleRecordName :: String,+    sampleRecordAge :: Int+  }+  deriving (Show, Eq, Generic)++deriveJSONUnPrefixLower ''SampleRecord++data AnotherType = AnotherType+  { anotherTypeFoo :: Bool,+    anotherTypeBar :: String+  }+  deriving (Show, Eq, Generic)++deriveJSONUnPrefixLower ''AnotherType++spec :: Spec+spec = describe "Eventium.Json" $ do+  describe "unPrefixLower" $ do+    let opts = unPrefixLower "sampleRecord"+    it "strips prefix and lowercases first char" $ do+      fieldLabelModifier opts "sampleRecordName" `shouldBe` "name"+      fieldLabelModifier opts "sampleRecordAge" `shouldBe` "age"++  describe "dropPrefix" $ do+    it "strips an exact prefix" $ do+      dropPrefix "foo" "fooBar" `shouldBe` "Bar"+      dropPrefix "sample" "sampleRecord" `shouldBe` "Record"++    it "returns remainder when prefix matches exactly" $ do+      dropPrefix "abc" "abcdef" `shouldBe` "def"++  describe "dropSuffix" $ do+    it "strips an exact suffix" $ do+      dropSuffix "Bar" "fooBar" `shouldBe` "foo"++  describe "deriveJSONUnPrefixLower" $ do+    it "produces JSON with stripped and lowercased field names" $ do+      let sample = SampleRecord "Alice" 30+          Object obj = toJSON sample+      KM.lookup "name" obj `shouldBe` Just (String "Alice")+      KM.lookup "age" obj `shouldBe` Just (Number 30)++    it "round-trips through JSON" $ do+      let sample = SampleRecord "Bob" 25+      decode (encode sample) `shouldBe` Just sample++    it "works for different type prefixes" $ do+      let val = AnotherType True "baz"+          Object obj = toJSON val+      KM.lookup "foo" obj `shouldBe` Just (Bool True)+      KM.lookup "bar" obj `shouldBe` Just (String "baz")++    it "round-trips for different types" $ do+      let val = AnotherType False "qux"+      decode (encode val) `shouldBe` Just val

@@ -0,0 +1,72 @@+{-# LANGUAGE OverloadedStrings #-}++module Eventium.ProjectionSpec (spec) where++import Control.Exception (evaluate)+import Data.Functor.Contravariant+import Eventium.Codec (Codec (..), DecodeError (..))+import Eventium.Projection+import Eventium.Store.Types+import Test.Hspec++-- Simple test projection: counter that sums integers+counterProjection :: Projection Int Int+counterProjection = Projection 0 (+)++spec :: Spec+spec = do+  describe "latestProjection" $ do+    it "should compute the correct final state" $ do+      latestProjection counterProjection [1, 2, 3, 4] `shouldBe` 10++    it "should return the seed for an empty event list" $ do+      latestProjection counterProjection [] `shouldBe` 0++  describe "allProjections" $ do+    it "should return all intermediate states" $ do+      allProjections counterProjection [1, 2, 3] `shouldBe` [0, 1, 3, 6]++    it "should return just the seed for an empty event list" $ do+      allProjections counterProjection [] `shouldBe` [0]++  describe "Contravariant instance" $ do+    it "should adapt the event type via contramap" $ do+      let stringProjection = contramap (length :: String -> Int) counterProjection+      latestProjection stringProjection ["hi", "hello", "a"] `shouldBe` 8++  describe "projectionMapMaybe" $ do+    it "should filter events and apply matching ones" $ do+      let evenOnly n = if even n then Just n else Nothing+          evenProjection = projectionMapMaybe evenOnly counterProjection+      latestProjection evenProjection [1, 2, 3, 4, 5, 6] `shouldBe` 12++    it "should return seed when no events match" $ do+      let noneMatch _ = Nothing+          emptyProjection = projectionMapMaybe noneMatch counterProjection+      latestProjection emptyProjection [1, 2, 3] `shouldBe` 0++  describe "codecProjection" $ do+    it "should deserialize and apply events" $ do+      let codec = Codec show (Just . (read :: String -> Int))+          codecProj = codecProjection codec counterProjection+      latestProjection codecProj ["1", "2", "3"] `shouldBe` 6++    it "should error on deserialization failure" $ do+      let codec = Codec show (\s -> if s == "bad" then Nothing else Just (read s :: Int))+          codecProj = codecProjection codec counterProjection+      evaluate (latestProjection codecProj ["1", "bad"]) `shouldThrow` (\(DecodeError ctx _) -> ctx == "codecProjection")++  describe "lenientCodecProjection" $ do+    it "should apply deserialized events and skip failures" $ do+      let codec = Codec show (\s -> if s == "bad" then Nothing else Just (read s :: Int))+          codecProj = lenientCodecProjection codec counterProjection+      latestProjection codecProj ["1", "2", "bad", "3"] `shouldBe` 6++  describe "streamProjectionEventHandler" $ do+    it "should update state and position" $ do+      let sp = streamProjection "key" (0 :: Int) counterProjection+          event = StreamEvent "other" 5 (emptyMetadata "") (10 :: Int)+          sp' = streamProjectionEventHandler sp event+      sp'.state `shouldBe` 10+      sp'.position `shouldBe` 5+      sp'.key `shouldBe` "key"

@@ -1,58 +0,0 @@-{-# LANGUAGE DeriveGeneric #-}-{-# LANGUAGE TemplateHaskell #-}--module Eventium.SerializerSpec (spec) where--import Data.Dynamic-import Data.Typeable (typeOf)-import Eventium.Serializer-import Eventium.TH.SumTypeSerializer-import GHC.Generics-import Test.Hspec--data EventA = EventA deriving (Show, Eq)--data EventB = EventB deriving (Show, Eq)--data EventC = EventC deriving (Show, Eq)--data AllEvents-  = AllEventsEventA EventA-  | AllEventsEventB EventB-  | AllEventsEventC EventC-  deriving (Show, Eq, Generic)--instance EventSumType AllEvents--data MyEvents-  = MyEventsEventA EventA-  | MyEventsEventB EventB-  deriving (Show, Eq, Generic)--instance EventSumType MyEvents--mkSumTypeSerializer "myEventsSerializer" ''MyEvents ''AllEvents--spec :: Spec-spec = do-  describe "EventSumType" $ do-    it "can serialize events without the constructor" $ do-      dynTypeRep (eventToDyn $ MyEventsEventA EventA) `shouldBe` typeOf EventA-      dynTypeRep (eventToDyn $ MyEventsEventA EventA) `shouldBe` dynTypeRep (eventToDyn $ AllEventsEventA EventA)--    it "can deserialize events with the constructor" $ do-      eventFromDyn (toDyn EventA) `shouldBe` Just (MyEventsEventA EventA)-      eventFromDyn (toDyn EventB) `shouldBe` Just (AllEventsEventB EventB)--      eventFromDyn (eventToDyn $ MyEventsEventA EventA) `shouldBe` Just (AllEventsEventA EventA)-      eventFromDyn (eventToDyn $ AllEventsEventB EventB) `shouldBe` Just (MyEventsEventB EventB)--  describe "mkSumTypeSerializer" $ do-    it "can serialize events" $ do-      serialize myEventsSerializer (MyEventsEventA EventA) `shouldBe` AllEventsEventA EventA-      serialize myEventsSerializer (MyEventsEventB EventB) `shouldBe` AllEventsEventB EventB--    it "can deserialize events" $ do-      deserialize myEventsSerializer (AllEventsEventA EventA) `shouldBe` Just (MyEventsEventA EventA)-      deserialize myEventsSerializer (AllEventsEventB EventB) `shouldBe` Just (MyEventsEventB EventB)-      deserialize myEventsSerializer (AllEventsEventC EventC) `shouldBe` Nothing

@@ -0,0 +1,84 @@+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE TemplateHaskell #-}++module Eventium.SumTypeSpec (spec) where++import Eventium.TH.SumType+import Test.Hspec++-- Test payload types++data Alpha = Alpha deriving (Show, Eq)++data Beta = Beta deriving (Show, Eq)++data Gamma = Gamma deriving (Show, Eq)++-- 1. constructSumType with default PrefixTagsWithTypeName+-- Generates: data Prefixed = PrefixedAlpha Alpha | PrefixedBeta Beta | PrefixedGamma Gamma+constructSumType "Prefixed" defaultSumTypeOptions [''Alpha, ''Beta, ''Gamma]++deriving instance Show Prefixed++deriving instance Eq Prefixed++-- 2. constructSumType with AppendTypeNameToTags+-- Generates: data Appended = AlphaAppended Alpha | BetaAppended Beta+constructSumType "Appended" (withTagOptions AppendTypeNameToTags defaultSumTypeOptions) [''Alpha, ''Beta]++deriving instance Show Appended++deriving instance Eq Appended++-- 3. constructSumType with custom ConstructTagName+-- Generates: data Custom = AlphaEvent Alpha | BetaEvent Beta+constructSumType "Custom" (withTagOptions (ConstructTagName (++ "Event")) defaultSumTypeOptions) [''Alpha, ''Beta]++deriving instance Show Custom++deriving instance Eq Custom++-- 4. A subset for converter tests (only Alpha, Beta — Prefixed has all three)+constructSumType "Subset" defaultSumTypeOptions [''Alpha, ''Beta]++deriving instance Show Subset++deriving instance Eq Subset++-- 5. sumTypeConverter: total conversion Subset -> Prefixed+sumTypeConverter "subsetToPrefixed" ''Subset ''Prefixed++-- 6. partialSumTypeConverter: partial conversion Prefixed -> Subset+partialSumTypeConverter "prefixedToSubset" ''Prefixed ''Subset++spec :: Spec+spec = do+  describe "constructSumType" $ do+    describe "PrefixTagsWithTypeName (default)" $ do+      it "generates constructors prefixed with the type name" $ do+        PrefixedAlpha Alpha `shouldBe` PrefixedAlpha Alpha+        PrefixedBeta Beta `shouldBe` PrefixedBeta Beta+        PrefixedGamma Gamma `shouldBe` PrefixedGamma Gamma++    describe "AppendTypeNameToTags" $ do+      it "generates constructors with type name appended" $ do+        AlphaAppended Alpha `shouldBe` AlphaAppended Alpha+        BetaAppended Beta `shouldBe` BetaAppended Beta++    describe "ConstructTagName" $ do+      it "generates constructors using the custom naming function" $ do+        AlphaEvent Alpha `shouldBe` AlphaEvent Alpha+        BetaEvent Beta `shouldBe` BetaEvent Beta++  describe "sumTypeConverter" $ do+    it "converts all source constructors to matching target constructors" $ do+      subsetToPrefixed (SubsetAlpha Alpha) `shouldBe` PrefixedAlpha Alpha+      subsetToPrefixed (SubsetBeta Beta) `shouldBe` PrefixedBeta Beta++  describe "partialSumTypeConverter" $ do+    it "returns Just for constructors present in the target type" $ do+      prefixedToSubset (PrefixedAlpha Alpha) `shouldBe` Just (SubsetAlpha Alpha)+      prefixedToSubset (PrefixedBeta Beta) `shouldBe` Just (SubsetBeta Beta)++    it "returns Nothing for constructors not in the target type" $ do+      prefixedToSubset (PrefixedGamma Gamma) `shouldBe` Nothing