diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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. diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -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) diff --git a/eventium-core.cabal b/eventium-core.cabal --- a/eventium-core.cabal +++ b/eventium-core.cabal @@ -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 diff --git a/src/Eventium.hs b/src/Eventium.hs --- a/src/Eventium.hs +++ b/src/Eventium.hs @@ -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 diff --git a/src/Eventium/Codec.hs b/src/Eventium/Codec.hs new file mode 100644 --- /dev/null +++ b/src/Eventium/Codec.hs @@ -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) +-- <> +-- +-- 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 diff --git a/src/Eventium/CommandDispatcher.hs b/src/Eventium/CommandDispatcher.hs new file mode 100644 --- /dev/null +++ b/src/Eventium/CommandDispatcher.hs @@ -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 diff --git a/src/Eventium/CommandHandler.hs b/src/Eventium/CommandHandler.hs --- a/src/Eventium/CommandHandler.hs +++ b/src/Eventium/CommandHandler.hs @@ -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) diff --git a/src/Eventium/EventBus.hs b/src/Eventium/EventBus.hs deleted file mode 100644 --- a/src/Eventium/EventBus.hs +++ /dev/null @@ -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 diff --git a/src/Eventium/EventHandler.hs b/src/Eventium/EventHandler.hs new file mode 100644 --- /dev/null +++ b/src/Eventium/EventHandler.hs @@ -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 diff --git a/src/Eventium/EventPublisher.hs b/src/Eventium/EventPublisher.hs new file mode 100644 --- /dev/null +++ b/src/Eventium/EventPublisher.hs @@ -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 diff --git a/src/Eventium/EventSubscription.hs b/src/Eventium/EventSubscription.hs new file mode 100644 --- /dev/null +++ b/src/Eventium/EventSubscription.hs @@ -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) diff --git a/src/Eventium/Json.hs b/src/Eventium/Json.hs new file mode 100644 --- /dev/null +++ b/src/Eventium/Json.hs @@ -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 diff --git a/src/Eventium/Json/TH.hs b/src/Eventium/Json/TH.hs new file mode 100644 --- /dev/null +++ b/src/Eventium/Json/TH.hs @@ -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 diff --git a/src/Eventium/ProcessManager.hs b/src/Eventium/ProcessManager.hs --- a/src/Eventium/ProcessManager.hs +++ b/src/Eventium/ProcessManager.hs @@ -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 ++ " " + +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 diff --git a/src/Eventium/Projection.hs b/src/Eventium/Projection.hs --- a/src/Eventium/Projection.hs +++ b/src/Eventium/Projection.hs @@ -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 diff --git a/src/Eventium/ProjectionCache/Cache.hs b/src/Eventium/ProjectionCache/Cache.hs new file mode 100644 --- /dev/null +++ b/src/Eventium/ProjectionCache/Cache.hs @@ -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) diff --git a/src/Eventium/ProjectionCache/Types.hs b/src/Eventium/ProjectionCache/Types.hs --- a/src/Eventium/ProjectionCache/Types.hs +++ b/src/Eventium/ProjectionCache/Types.hs @@ -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 diff --git a/src/Eventium/ReadModel.hs b/src/Eventium/ReadModel.hs new file mode 100644 --- /dev/null +++ b/src/Eventium/ReadModel.hs @@ -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) diff --git a/src/Eventium/ReadModel/Class.hs b/src/Eventium/ReadModel/Class.hs deleted file mode 100644 --- a/src/Eventium/ReadModel/Class.hs +++ /dev/null @@ -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) diff --git a/src/Eventium/Serializer.hs b/src/Eventium/Serializer.hs deleted file mode 100644 --- a/src/Eventium/Serializer.hs +++ /dev/null @@ -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) --- <> --- --- 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 diff --git a/src/Eventium/Store/Class.hs b/src/Eventium/Store/Class.hs --- a/src/Eventium/Store/Class.hs +++ b/src/Eventium/Store/Class.hs @@ -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 diff --git a/src/Eventium/Store/Queries.hs b/src/Eventium/Store/Queries.hs --- a/src/Eventium/Store/Queries.hs +++ b/src/Eventium/Store/Queries.hs @@ -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) diff --git a/src/Eventium/Store/Types.hs b/src/Eventium/Store/Types.hs new file mode 100644 --- /dev/null +++ b/src/Eventium/Store/Types.hs @@ -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 + ) diff --git a/src/Eventium/TH.hs b/src/Eventium/TH.hs --- a/src/Eventium/TH.hs +++ b/src/Eventium/TH.hs @@ -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 diff --git a/src/Eventium/TH/Projection.hs b/src/Eventium/TH/Projection.hs --- a/src/Eventium/TH/Projection.hs +++ b/src/Eventium/TH/Projection.hs @@ -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 diff --git a/src/Eventium/TH/SumType.hs b/src/Eventium/TH/SumType.hs new file mode 100644 --- /dev/null +++ b/src/Eventium/TH/SumType.hs @@ -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) [] diff --git a/src/Eventium/TH/SumTypeCodec.hs b/src/Eventium/TH/SumTypeCodec.hs new file mode 100644 --- /dev/null +++ b/src/Eventium/TH/SumTypeCodec.hs @@ -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 diff --git a/src/Eventium/TH/SumTypeSerializer.hs b/src/Eventium/TH/SumTypeSerializer.hs deleted file mode 100644 --- a/src/Eventium/TH/SumTypeSerializer.hs +++ /dev/null @@ -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 diff --git a/src/Eventium/TypeEmbedding.hs b/src/Eventium/TypeEmbedding.hs new file mode 100644 --- /dev/null +++ b/src/Eventium/TypeEmbedding.hs @@ -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 diff --git a/src/Eventium/UUID.hs b/src/Eventium/UUID.hs --- a/src/Eventium/UUID.hs +++ b/src/Eventium/UUID.hs @@ -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. -- diff --git a/tests/Eventium/CodecSpec.hs b/tests/Eventium/CodecSpec.hs new file mode 100644 --- /dev/null +++ b/tests/Eventium/CodecSpec.hs @@ -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 diff --git a/tests/Eventium/CommandDispatcherSpec.hs b/tests/Eventium/CommandDispatcherSpec.hs new file mode 100644 --- /dev/null +++ b/tests/Eventium/CommandDispatcherSpec.hs @@ -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 diff --git a/tests/Eventium/EventHandlerSpec.hs b/tests/Eventium/EventHandlerSpec.hs new file mode 100644 --- /dev/null +++ b/tests/Eventium/EventHandlerSpec.hs @@ -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] diff --git a/tests/Eventium/JsonSpec.hs b/tests/Eventium/JsonSpec.hs new file mode 100644 --- /dev/null +++ b/tests/Eventium/JsonSpec.hs @@ -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 diff --git a/tests/Eventium/ProjectionSpec.hs b/tests/Eventium/ProjectionSpec.hs new file mode 100644 --- /dev/null +++ b/tests/Eventium/ProjectionSpec.hs @@ -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" diff --git a/tests/Eventium/SerializerSpec.hs b/tests/Eventium/SerializerSpec.hs deleted file mode 100644 --- a/tests/Eventium/SerializerSpec.hs +++ /dev/null @@ -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 diff --git a/tests/Eventium/SumTypeSpec.hs b/tests/Eventium/SumTypeSpec.hs new file mode 100644 --- /dev/null +++ b/tests/Eventium/SumTypeSpec.hs @@ -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