packages feed

eventsourcing (empty) → 0.9.0

raw patch · 23 files changed

+3046/−0 lines, 23 filesdep +basedep +deepseqdep +eventsourcingsetup-changed

Dependencies added: base, deepseq, eventsourcing, free, hashable, hedgehog, mtl, pipes, psqueues, stm, tasty, tasty-hedgehog, time, unordered-containers

Files

+ CHANGELOG.md view
@@ -0,0 +1,5 @@+# Revision history for eventsourcing++## 0.9.0 -- 2020-08-16++* First version.
+ LICENSE view
@@ -0,0 +1,5 @@+Copyright 2019-2020 Tom Feron <tho.feron@gmail.com>++Permission to use, copy, modify, and/or distribute this software for any purpose with or without fee is hereby granted, provided that the above copyright notice and this permission notice appear in all copies.++THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ eventsourcing.cabal view
@@ -0,0 +1,77 @@+cabal-version:       >=1.10+name:                eventsourcing+version:             0.9.0+synopsis:            CQRS/ES library.+description:         Backend-agnostic implementation of CQRS/ES.+license:             ISC+license-file:        LICENSE+author:              Tom Feron <tho.feron@gmail.com>+maintainer:          Tom Feron <tho.feron@gmail.com>+build-type:          Simple+extra-source-files:  CHANGELOG.md+homepage:            https://github.com/thoferon/eventsourcing+bug-reports:         https://github.com/thoferon/eventsourcing/issues+category:            Database++source-repository head+  type:     git+  location: git://github.com/thoferon/eventsourcing.git+  subdir:   eventsourcing++library+  hs-source-dirs:    src+  default-language:  Haskell2010+  ghc-options:       -Wall++  exposed-modules:+    Database.CQRS+    Database.CQRS.Error+    Database.CQRS.Event+    Database.CQRS.InMemory+    Database.CQRS.Projection+    Database.CQRS.ReadModel+    Database.CQRS.ReadModel.AggregateStore+    Database.CQRS.Stream+    Database.CQRS.StreamFamily+    Database.CQRS.TabularData+    Database.CQRS.TabularData.Internal+    Database.CQRS.TabularData.Optimisation+    Database.CQRS.TaskManager+    Database.CQRS.Transformer++  build-depends:+    base >=4.12 && <5,+    deepseq >=1.4,+    free >=5,+    hashable,+    mtl,+    pipes >=4.3,+    psqueues >=0.2,+    stm >=2.5,+    time,+    unordered-containers++test-suite unit-tests+  type:              exitcode-stdio-1.0+  main-is:           Main.hs+  hs-source-dirs:    test+  default-language:  Haskell2010+  ghc-options:       -Wall -threaded++  other-modules:+    Database.CQRS.InMemoryTest+    Database.CQRS.TabularDataTest+    Database.CQRS.TransformerTest+    Helpers++  build-depends:+    base,+    deepseq,+    eventsourcing,+    hedgehog >=1.0,+    mtl,+    pipes,+    stm,+    tasty >=1.2,+    tasty-hedgehog >=1.0,+    unordered-containers
+ src/Database/CQRS.hs view
@@ -0,0 +1,60 @@+module Database.CQRS+  ( -- * Events+    Event(..)+  , WritableEvent(..)++    -- * Streams+  , Stream(..)+  , WritableStream(..)+  , EventWithContext(..)+  , EventWithContext'+  , MonadMetadata(..)+  , ConsistencyCheck(..)+  , writeEvent+  , writeEventCc+  , optimistically+  , StreamBounds(..)+  , StreamBounds'+  , afterEvent+  , untilEvent++    -- * Stream families+  , StreamFamily(..)++    -- * Aggregators and projections+  , Aggregator+  , runAggregator+  , Projection+  , runProjection+  , TrackedState(..)+  , TrackingTable(..)+  , InMemoryTrackingTable(..)+  , createInMemoryTrackingTable+  , executeInMemoryActions++    -- * Read models+  , ReadModel(..)++    -- * Transformers+  , Transformer+  , TransformedStream+  , transformStream+  , TransformedStreamFamily+  , transformStreamFamily+  , Transform+  , pushEvent+  , mergeEvents+  , flushEvents+  , failTransformer++  -- * Errors+  , Error(..)+  ) where++import Database.CQRS.Error+import Database.CQRS.Event+import Database.CQRS.Projection+import Database.CQRS.ReadModel+import Database.CQRS.Stream+import Database.CQRS.StreamFamily+import Database.CQRS.Transformer
+ src/Database/CQRS/Error.hs view
@@ -0,0 +1,18 @@+{-# LANGUAGE DeriveGeneric #-}++module Database.CQRS.Error+  ( Error(..)+  ) where++import GHC.Generics++data Error+  = EventWriteError String+  | EventDecodingError String String+  | EventRetrievalError String+  | NewEventsStreamingError String+  | ProjectionError String+  | MigrationError String+  | ConsistencyCheckError String+  | TrackingTableError String+  deriving (Eq, Show, Generic)
+ src/Database/CQRS/Event.hs view
@@ -0,0 +1,50 @@+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE FunctionalDependencies #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE TypeFamilies #-}++-- | This module provides the interface between an event stream and the event+-- type of the stream. An 'Event' is something that can be read from an event+-- stream. It can also be a 'WritableEvent' which, unsurprisingly, means it+-- can be added to the stream. Having two different typeclasses allows us to+-- prevent a subset of events to be written as in the following example.+--+-- @+--  data Current+--  data Deprecated+--+--  data MyEvent a where+--    EventA :: MyEvent Current+--    EventB :: MyEvent Current+--    EventC :: MyEvent Deprecated+--+--  instance Event (MyEvent e) where+--    type DecodingFormat = Aeson.Value+--    decodeEvent = Aeson.eitherDecode+--+--  instance WritableEvent (MyEvent Current) where+--    type EncodingFormat = Aeson.Value+--    encodeEvent = Aeson.encode+--+--  instance TypeError (Text "Cannot write deprecated event")+--      => WritableEvent (MyEvent Deprecated)+-- @++module Database.CQRS.Event+  ( Event(..)+  , WritableEvent(..)+  ) where++-- | Event that can read from some event stream with compatible decoding format+-- and fed to a projection or aggregated in some way.+class Event e where+  -- | Format in which the event is encoded, e.g. 'Data.Aeson.Value'.+  type EncodingFormat e :: *+  decodeEvent :: EncodingFormat e -> Either String e++-- | Event that can be written to an event stream. This is separate from 'Event'+-- to make it possible to restrict the events that can be written with a GADT.+class Event e => WritableEvent e where+  -- | Format from which the event can be decoded, e.g. 'Data.Aeson.Value'.+  encodeEvent :: e -> EncodingFormat e
+ src/Database/CQRS/InMemory.hs view
@@ -0,0 +1,288 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeFamilies #-}++module Database.CQRS.InMemory+  ( Stream+  , emptyStream+  , StreamFamily+  , emptyStreamFamily+  ) where++import Prelude hiding (last, length)++import Control.Exception   (evaluate)+import Control.Monad       (forever, forM, when)+import Control.Monad.Trans (MonadIO(..))+import Data.Foldable       (foldrM)+import System.Mem.Weak     (Weak, deRefWeak, mkWeakPtr)++import qualified Control.Concurrent.STM as STM+import qualified Control.DeepSeq        as Seq+import qualified Control.Monad.Except   as Exc+import qualified Data.Hashable          as Hash+import qualified Data.HashMap.Strict    as HM+import qualified GHC.Conc+import qualified Pipes++import qualified Database.CQRS as CQRS++-- | In-memory event stream.+--+-- It's using STM internally so it's safe to have different thread writing+-- events to the stream concurrently. Its main purpose is for tests but it could+-- be used for production code as well.+data Stream metadata event = Stream+  { events :: InnerStream (event, metadata)+  , last   :: STM.TVar (STM.TMVar (InnerStream (event, metadata)))+  , length :: STM.TVar Integer+  , notify :: CQRS.EventWithContext Integer metadata event -> STM.STM ()+  }++data InnerStream a+  = Cons a (InnerStream a)+  | Future (STM.TMVar (InnerStream a))++-- | Initialise an empty event stream.+emptyStream :: MonadIO m => m (Stream metadata event)+emptyStream =+  liftIO . STM.atomically $ emptyStreamSTM++emptyStreamSTM :: STM.STM (Stream metadata event)+emptyStreamSTM = do+  tmvar  <- STM.newEmptyTMVar+  last   <- STM.newTVar tmvar+  length <- STM.newTVar 0+  let events = Future tmvar+      notify = const $ pure ()+  pure Stream{..}++instance MonadIO m => CQRS.Stream m (Stream metadata event) where+  type EventType       (Stream metadata event) = event+  type EventIdentifier (Stream metadata event) = Integer -- starting at 1+  type EventMetadata   (Stream metadata event) = metadata++  streamEvents = streamStreamEvent++instance+    ( Exc.MonadError CQRS.Error m+    , MonadIO m+    , Seq.NFData event+    , Seq.NFData metadata+    )+    => CQRS.WritableStream m (Stream metadata event) where+  writeEventWithMetadata = streamWriteEventWithMetadata++streamWriteEventWithMetadata+  :: ( CQRS.EventIdentifier (Stream metadata event) ~ Integer+     , Exc.MonadError CQRS.Error m+     , MonadIO m+     , Seq.NFData event+     , Seq.NFData metadata+     )+  => Stream metadata event+  -> event+  -> metadata+  -> CQRS.ConsistencyCheck Integer+  -> m Integer+streamWriteEventWithMetadata Stream{..} event metadata cc =+  (Exc.liftEither =<<) . liftIO $ do+    -- Force exceptions now and in this thread.+    evaluate $ Seq.rnf (event, metadata)+    STM.atomically $ do+      mErr <- case cc of+        CQRS.CheckNoEvents -> do+          len <- STM.readTVar length+          pure $ if len == 0 then Nothing else Just "stream not empty"+        CQRS.CheckLastEvent identifier -> do+          len <- STM.readTVar length+          -- The identifier of the last event is the stream length.+          pure $ if len == identifier+            then Nothing+            else Just "last event identifier doesn't match"+        CQRS.NoConsistencyCheck -> pure Nothing++      case mErr of+        Just err -> pure . Left . CQRS.ConsistencyCheckError $ err+        Nothing -> do+          lastVar <- STM.readTVar last+          newVar  <- STM.newEmptyTMVar+          let innerStream = Cons (event, metadata) (Future newVar)+          STM.putTMVar lastVar innerStream+          STM.writeTVar last newVar+          identifier <- STM.stateTVar length $ \l ->+            let l' = l + 1 in l' `seq` (l', l')+          notify $ CQRS.EventWithContext identifier metadata event+          pure $ Right identifier++streamStreamEvent+  :: MonadIO m+  => Stream metadata event+  -> CQRS.StreamBounds' (Stream metadata event)+  -> Pipes.Producer+      [ Either+          (Integer, String) (CQRS.EventWithContext' (Stream metadata event))+      ] m ()+streamStreamEvent stream bounds = do+    let innerStream = events stream+    go 1 innerStream+  where+    go+      :: MonadIO m+      => Integer+      -> InnerStream (event, metadata)+      -> Pipes.Producer+          [ Either+              (Integer, String) (CQRS.EventWithContext Integer metadata event)+          ] m ()+    go n = \case+      Cons (event, metadata) innerStream -> do+        when (inBounds n) $+          Pipes.yield [Right (CQRS.EventWithContext n metadata event)]+        go (n+1) innerStream+      Future var -> do+        mInnerStream <- liftIO . STM.atomically . STM.tryReadTMVar $ var+        case mInnerStream of+          Nothing -> pure ()+          Just innerStream -> go n innerStream++    inBounds :: Integer -> Bool+    inBounds n =+      maybe True (n >) (CQRS._afterEvent bounds)+      && maybe True (n <=) (CQRS._untilEvent bounds)++-- | A family of in-memory streams.+--+-- There are two things to be aware of when using this type:+--+-- 'getStream' adds a new empty stream to the family regardless of whether it's+-- used or not. If an attacker can make your application call 'getStream' with+-- arbitrary stream identifiers, it can lead to a Denial-of-Service attack.+--+-- Slow consumers of event notifications (from 'allNewEvents') could become an+-- issue if events are written at a higher pace that they can keep up with. In+-- order to avoid the queue of notifications to grow bigger and bigger, it's+-- capped at 100 (hard-coded for now.) This means that writing a new event can+-- be blocked by a full consumer's queue.+data StreamFamily identifier metadata event = StreamFamily+  { hashMap :: STM.TVar (HM.HashMap identifier (Stream metadata event))+  , queues+      :: STM.TVar [Weak (STM.TBQueue+                          ( identifier+                          , CQRS.EventWithContext' (Stream metadata event)+                          ))]+  }++-- | Initialise an empty stream family.+emptyStreamFamily+  -- metadata first as it's the most likely to be passed explicitly.+  :: forall metadata identifier event m. MonadIO m+  => m (StreamFamily identifier metadata event)+emptyStreamFamily =+  liftIO . STM.atomically $ do+    hashMap <- STM.newTVar HM.empty+    queues  <- STM.newTVar []+    pure StreamFamily{..}++instance+    (Eq identifier, Hash.Hashable identifier, MonadIO m)+    => CQRS.StreamFamily m (StreamFamily identifier metadata event) where+  type StreamType (StreamFamily identifier metadata event) =+    Stream metadata event+  type StreamIdentifier (StreamFamily identifier metadata event) =+    identifier++  getStream              = streamFamilyGetStream+  allNewEvents           = streamFamilyAllNewEvents+  latestEventIdentifiers = streamFamilyLatestEventIdentifiers++streamFamilyNotify+  :: StreamFamily identifier metadata event+  -> identifier+  -> CQRS.EventWithContext' (Stream metadata event)+  -> STM.STM ()+streamFamilyNotify StreamFamily{..} identifier event = do+  qs <- STM.readTVar queues+  qs' <- (\f -> foldrM f [] qs) $ \queueWeak acc -> do+    -- This call to unsafeIOToSTM should be safe in this context.+    mQueue <- GHC.Conc.unsafeIOToSTM $ deRefWeak queueWeak+    case mQueue of+      Just queue -> do+        STM.writeTBQueue queue (identifier, event)+        pure $ queueWeak : acc+      Nothing -> pure acc+  STM.writeTVar queues qs'++streamFamilyGetStream+  :: (Eq identifier, Hash.Hashable identifier, MonadIO m)+  => StreamFamily identifier metadata event+  -> identifier+  -> m (Stream metadata event)+streamFamilyGetStream sf@StreamFamily{..} identifier =+  liftIO . STM.atomically $ do+    hm <- STM.readTVar hashMap+    case HM.lookup identifier hm of+      Nothing -> do+        stream <- emptyStreamSTM+        let stream' = stream { notify = streamFamilyNotify sf identifier }+            hm' = HM.insert identifier stream' hm+        STM.writeTVar hashMap hm'+        pure stream'+      Just stream -> pure stream++streamFamilyAllNewEvents+  :: forall identifier metadata event m a. MonadIO m+  => StreamFamily identifier metadata event+  -> m (Pipes.Producer+        [ ( identifier+          , Either+              (Integer, String)+              (CQRS.EventWithContext' (Stream metadata event))+          ) ]+        m a)+streamFamilyAllNewEvents StreamFamily{..} = do+    queue <- initialise+    pure $ producer queue++  where+    initialise+      :: m (STM.TBQueue+             (identifier, CQRS.EventWithContext' (Stream metadata event)))+    initialise = liftIO $ do+      queue <- STM.newTBQueueIO 100+      queueWeak <- mkWeakPtr queue Nothing+      STM.atomically $ STM.modifyTVar' queues (queueWeak :)+      pure queue++    producer+      :: STM.TBQueue+          (identifier, CQRS.EventWithContext' (Stream metadata event))+      -> Pipes.Producer+          [ ( identifier+            , Either+                (Integer, String)+                (CQRS.EventWithContext' (Stream metadata event))+            ) ]+          m a+    producer queue = forever $ do+      xs <- liftIO . STM.atomically $+        (:) <$> STM.readTBQueue queue <*> STM.flushTBQueue queue+      Pipes.yield $ map (fmap Right) xs++streamFamilyLatestEventIdentifiers+  :: MonadIO m+  => StreamFamily identifier metadata event+  -> Pipes.Producer (identifier, Integer) m ()+streamFamilyLatestEventIdentifiers StreamFamily{..} = do+  ids <- liftIO . STM.atomically $ do+    hm <- STM.readTVar hashMap+    let pairs = HM.toList hm+    forM pairs $ \(streamId, stream) -> do+      lastEventId <- STM.readTVar $ length stream+      pure (streamId, lastEventId)++  Pipes.each ids
+ src/Database/CQRS/Projection.hs view
@@ -0,0 +1,391 @@+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE TupleSections #-}+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FunctionalDependencies #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE ScopedTypeVariables #-}++module Database.CQRS.Projection+  ( Aggregator+  , runAggregator+  , Projection+  , runProjection+  , TrackedState(..)+  , TrackingTable(..)+  , InMemoryTrackingTable(..)+  , createInMemoryTrackingTable+  , executeInMemoryActions+  ) where++import Control.Monad+import Control.Monad.Trans (MonadIO(..), lift)+import Data.Hashable (Hashable)+import Data.Tuple (swap)+import Pipes ((>->))++import qualified Data.HashMap.Strict        as HM+import qualified Control.Concurrent.STM     as STM+import qualified Control.Monad.Except       as Exc+import qualified Control.Monad.State.Strict as St+import qualified Pipes++import Database.CQRS.Error+import Database.CQRS.Stream+import Database.CQRS.StreamFamily++-- | Function aggregating a state in memory.+type Aggregator event agg =+  event -> St.State agg ()++-- | Projection returning actions that can be batched and executed.+--+-- This can be used to batch changes to tables in a database for example.+type Projection event st action =+  event -> St.State st [action]++-- | Run an 'Aggregator' on events from a stream starting with a given state and+-- return the new aggregate state, the identifier of the last event processed if+-- any and how many of them were processed.+runAggregator+  :: forall m stream aggregate.+     ( Exc.MonadError Error m+     , Show (EventIdentifier stream)+     , Stream m stream+     )+  => Aggregator (EventWithContext' stream) aggregate+  -> stream+  -> StreamBounds' stream+  -> aggregate+  -> m (aggregate, Maybe (EventIdentifier stream), Int)+runAggregator aggregator stream bounds initState = do+  flip St.execStateT (initState, Nothing, 0) . Pipes.runEffect $+    Pipes.hoist lift (streamEvents stream bounds)+      >-> flatten+      >-> aggregatorPipe++  where+    aggregatorPipe+      :: Pipes.Consumer+          (Either (EventIdentifier stream, String) (EventWithContext' stream))+          (St.StateT (aggregate, Maybe (EventIdentifier stream), Int) m) ()+    aggregatorPipe = forever $ do+      ewc <- Pipes.await >>= \case+        Left (eventId, err) ->+          Exc.throwError $ EventDecodingError (show eventId) err+        Right e -> pure e++      St.modify' $ \(aggregate, _, eventCount) ->+        let aggregate' = St.execState (aggregator  ewc) aggregate+        in (aggregate', Just (identifier ewc), eventCount + 1)++flatten :: Monad m => Pipes.Pipe [a] a m ()+flatten = forever $ Pipes.await >>= Pipes.each++runProjection+  :: forall streamFamily action trackingTable m st.+     ( Exc.MonadError Error m+     , Hashable (StreamIdentifier streamFamily)+     , Ord (EventIdentifier (StreamType streamFamily))+     , Ord (StreamIdentifier streamFamily)+     , Stream m (StreamType streamFamily)+     , StreamFamily m streamFamily+     , TrackingTable m trackingTable+        (StreamIdentifier streamFamily)+        (EventIdentifier (StreamType streamFamily))+        st+     )+  => streamFamily+  -> (StreamIdentifier streamFamily -> st)+     -- ^ Initialise state when no events have been processed yet.+  -> Projection+      (EventWithContext' (StreamType streamFamily))+      st action+  -> trackingTable+  -> (trackingTable+      -> ( st+         , [action]+         , StreamIdentifier streamFamily+         , EventIdentifier (StreamType streamFamily)+         )+      -> m ())+  -- ^ Commit the custom actions. See 'executeSqlActions' for 'SqlAction's.+  -- This consumer is expected to update the tracking table accordingly.+  -> m ()+runProjection streamFamily initState projection trackingTable+              executeActions = do+    newEvents <- allNewEvents streamFamily+    Pipes.runEffect $ do+      latestEventIdentifiers streamFamily >-> catchUp+      newEvents+        >-> groupByStream+        >-> bisectFailingBatches id+              (\batch -> Pipes.runEffect $+                Pipes.for (projectionPipe batch)+                  (lift . executeActions trackingTable))++  where+    catchUp+      :: Pipes.Consumer+          ( StreamIdentifier streamFamily+          , EventIdentifier (StreamType streamFamily)+          ) m ()+    catchUp = forever $ do+      (streamId, eventId) <- Pipes.await+      stream <- lift $ getStream streamFamily streamId+      state <- lift $ getTrackedState trackingTable streamId++      lift . Pipes.runEffect $ case state of+        NeverRan -> catchUp' streamId stream mempty+        SuccessAt lastSuccesfulEventId _+          | lastSuccesfulEventId < eventId ->+              catchUp' streamId stream (afterEvent lastSuccesfulEventId)+          | otherwise -> pure ()+        -- We are catching up, so maybe the executable was restarted and this+        -- stream won't fail this time.+        FailureAt (Just (lastSuccessfulEventId, _)) _ _ ->+          catchUp' streamId stream (afterEvent lastSuccessfulEventId)+        FailureAt Nothing _ _ -> catchUp' streamId stream mempty++    catchUp'+      :: StreamIdentifier streamFamily+      -> StreamType streamFamily+      -> StreamBounds' (StreamType streamFamily)+      -> Pipes.Effect m ()+    catchUp' streamId stream bounds =+      streamEvents stream bounds+        >-> bisectFailingBatches (streamId,)+              (\batch -> Pipes.runEffect $+                Pipes.for (projectionPipe batch)+                  (lift . executeActions trackingTable))++    groupByStream+      :: Pipes.Pipe+          [ ( StreamIdentifier streamFamily+            , Either+                (EventIdentifier (StreamType streamFamily), String)+                (EventWithContext' (StreamType streamFamily))+            ) ]+          ( StreamIdentifier streamFamily+          , [ Either+                (EventIdentifier (StreamType streamFamily), String)+                (EventWithContext' (StreamType streamFamily)) ]+          )+          m ()+    groupByStream = forever $ do+      events <- Pipes.await+      let eventsByStream =+            HM.toList . HM.fromListWith (flip (++)) . map (fmap pure) $ events+      Pipes.each eventsByStream++    -- It must be used with 'Pipes.for' to ensure the execution continues after+    -- the call to 'Pipes.yield'.+    projectionPipe+      :: ( StreamIdentifier streamFamily+         , [ Either+              (EventIdentifier (StreamType streamFamily), String)+              (EventWithContext' (StreamType streamFamily)) ] )+      -> Pipes.Producer+          ( st+          , [action]+          , StreamIdentifier streamFamily+          , EventIdentifier (StreamType streamFamily)+          ) m ()+    projectionPipe (streamId, eEvents) = do+      let filterEventsAfter eventId = filter $ \case+            Left (eventId', _) -> eventId' > eventId+            Right EventWithContext{ identifier = eventId' } ->+              eventId' > eventId++      state <- lift $ getTrackedState trackingTable streamId+      case state of+        NeverRan ->+          coreProjectionPipe streamId (initState streamId) eEvents+        SuccessAt lastSuccesfulEventId st ->+          coreProjectionPipe streamId st+            . filterEventsAfter lastSuccesfulEventId+            $ eEvents+        -- If this batch has events lower than the identifier for which the+        -- projection has failed, it means this is part of the catch-up phase.+        -- It happens after a restart, so the code might have changed and we+        -- should retry.+        FailureAt mSuccess eventId _ -> do+          let check = \case+                Left (eventId', _) -> eventId' <= eventId+                Right EventWithContext{..} -> identifier <= eventId+          when (any check eEvents) $+            case mSuccess of+              Nothing ->+                coreProjectionPipe streamId (initState streamId) eEvents+              Just (lastSuccesfulEventId, st) ->+                coreProjectionPipe streamId st+                  . filterEventsAfter lastSuccesfulEventId+                  $ eEvents++    coreProjectionPipe+      :: StreamIdentifier streamFamily+      -> st+      -> [ Either+            (EventIdentifier (StreamType streamFamily), String)+            (EventWithContext' (StreamType streamFamily)) ]+      -> Pipes.Producer+          ( st+          , [action]+          , StreamIdentifier streamFamily+          , EventIdentifier (StreamType streamFamily)+          ) m ()+    coreProjectionPipe streamId st eEvents = do+      -- "Healthy" events up until the first error if any. We want to process+      -- the events before throwing the error so that chunking as no effect on+      -- semantics.+      let (events, mFirstError) = stopOnLeft eEvents+          (st', actions) =+            fmap mconcat+              . swap+              . flip St.runState st+              . mapM projection+              $ events++      unless (null actions) $ do+        -- There is a last event, otherwise actions would be empty.+        let latestEventId = identifier . last $ events+        Pipes.yield (st', actions, streamId, latestEventId)++      case mFirstError of+        Nothing -> pure ()+        Just (eventId, err) -> do+          lift $ upsertError trackingTable streamId eventId err+          pure ()++    bisectFailingBatches+      :: (a+          -> ( StreamIdentifier streamFamily+             , [ Either+                  (EventIdentifier (StreamType streamFamily), String)+                  (EventWithContext' (StreamType streamFamily)) ]))+      -> (( StreamIdentifier streamFamily+          , [ Either+                (EventIdentifier (StreamType streamFamily), String)+                (EventWithContext' (StreamType streamFamily)) ])+          -> m ())+      -> Pipes.Consumer a m ()+    bisectFailingBatches split f = forever $ do+        input <- Pipes.await+        let (streamId, batch) = split input+        void . lift $ go streamId batch++      where+        go+          :: StreamIdentifier streamFamily+          -> [ Either+                (EventIdentifier (StreamType streamFamily), String)+                (EventWithContext' (StreamType streamFamily)) ]+          -> m Bool+        go streamId batch =+          Exc.catchError+            (f (streamId, batch) >> pure True)+            (\err -> case (err, batch) of+              (ProjectionError _, _ : _ : _) -> do+                let len = length batch+                    (batch1, batch2) =+                      splitAt (floor @Double (fromIntegral len / 2)) batch+                succeeded <- go streamId batch1+                if succeeded+                  then go streamId batch2+                  else pure False++              (ProjectionError err', [Left (eventId, _)]) -> do+                upsertError trackingTable streamId eventId err'+                pure False++              (ProjectionError err', [Right EventWithContext{..}]) -> do+                upsertError trackingTable streamId identifier err'+                pure False++              _ -> Exc.throwError err)++-- | Return all the 'Right' elements before the first 'Left' and the value of+-- the first 'Left'.+stopOnLeft :: [Either a b] -> ([b], Maybe a)+stopOnLeft = go id+  where+    go :: ([b] -> [b]) -> [Either a b] -> ([b], Maybe a)+    go f = \case+      [] -> (f [], Nothing)+      Left err : _ -> (f [], Just err)+      Right x : xs -> go (f . (x:)) xs++data TrackedState identifier st+  = NeverRan+  | SuccessAt identifier st+  | FailureAt (Maybe (identifier, st)) identifier String+    -- ^ Last succeeded at, failed at.+  deriving (Eq, Show)++class+    TrackingTable m table streamId eventId st+    | table -> streamId, table -> eventId, table -> st where+  getTrackedState :: table -> streamId -> m (TrackedState eventId st)+  upsertError     :: table -> streamId -> eventId -> String -> m ()++newtype InMemoryTrackingTable streamId eventId st+  = InMemoryTrackingTable+      (STM.TVar (HM.HashMap+                  streamId (Maybe (eventId, st), Maybe (eventId, String))))++instance+    (Hashable streamId, MonadIO m, Ord streamId)+    => TrackingTable m (InMemoryTrackingTable streamId eventId st)+        streamId eventId st where++  getTrackedState (InMemoryTrackingTable tvar) streamId =+    liftIO . STM.atomically $ do+      hm <- STM.readTVar tvar+      pure $ case HM.lookup streamId hm of+        Nothing -> NeverRan+        Just (Nothing, Nothing) -> NeverRan -- This actually can't happen.+        Just (successPair, Just (eventId, err)) ->+          FailureAt successPair eventId err+        Just (Just (eventId, st), Nothing) -> SuccessAt eventId st++  upsertError (InMemoryTrackingTable tvar) streamId eventId err =+    liftIO . STM.atomically . STM.modifyTVar tvar $+      HM.alter+        (\case+          Nothing -> Just (Nothing, Just (eventId, err))+          Just (mSuccess, _) -> Just (mSuccess, Just (eventId, err)))+        streamId++createInMemoryTrackingTable+  :: MonadIO m => m (InMemoryTrackingTable streamId eventId st)+createInMemoryTrackingTable =+  liftIO . fmap InMemoryTrackingTable . STM.newTVarIO $ HM.empty++executeInMemoryActions+  :: ( Exc.MonadError Error m+     , Hashable streamId+     , MonadIO m+     , Ord streamId+     )+  => STM.TVar projected+  -> (projected -> action -> Either Error projected)+     -- ^ For tabular data actions, use `applyTabularDataAction`.+  -> InMemoryTrackingTable streamId eventId st+  -> (st, [action], streamId, eventId)+  -> m ()+executeInMemoryActions+    tvar applyAction (InMemoryTrackingTable trackingTVar)+    (st, actions, streamId, eventId) =++  Exc.liftEither <=< liftIO . STM.atomically $ do+    projected <- STM.readTVar tvar++    case foldM applyAction projected actions of+      Left err -> pure $ Left err++      Right projected' -> do+        STM.writeTVar tvar projected'+        STM.modifyTVar' trackingTVar $+          HM.insert streamId (Just (eventId, st), Nothing)+        pure $ Right ()
+ src/Database/CQRS/ReadModel.hs view
@@ -0,0 +1,12 @@+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE TypeFamilies #-}++module Database.CQRS.ReadModel+  ( ReadModel(..)+  ) where++class ReadModel f model where+  type ReadModelQuery    model :: *+  type ReadModelResponse model :: *++  query :: model -> ReadModelQuery model -> f (ReadModelResponse model)
+ src/Database/CQRS/ReadModel/AggregateStore.hs view
@@ -0,0 +1,205 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE UndecidableInstances #-}++module Database.CQRS.ReadModel.AggregateStore+  ( AggregateStore+  , makeAggregateStore+  , Response(..)+  ) where++import Control.Monad.Trans (MonadIO(..))+import Data.Hashable (Hashable)++import qualified Control.Monad.Except   as Exc+import qualified Control.Concurrent.STM as STM+import qualified Data.HashPSQ           as HashPSQ+import qualified Data.Time              as T++import qualified Database.CQRS as CQRS++data Response eventId aggregate = Response+  { lastEventId     :: Maybe eventId+  , aggregate       :: aggregate+  , eventCount      :: Int -- ^ Number of events processed in this fetch.+  , totalEventCount :: Int -- ^ Total number of events making this aggregate.+  }++data AggregateStore streamFamily aggregate = AggregateStore+  { streamFamily     :: streamFamily+  , aggregator       :: CQRS.Aggregator+                          (CQRS.EventWithContext'+                            (CQRS.StreamType streamFamily))+                          aggregate+  , initialAggregate :: CQRS.StreamIdentifier streamFamily -> aggregate+  , cache            :: Cache+                          (CQRS.StreamIdentifier streamFamily)+                          (CQRS.EventIdentifier+                            (CQRS.StreamType streamFamily))+                          aggregate+  , lagTolerance     :: T.NominalDiffTime+  }++makeAggregateStore+  :: MonadIO m+  => streamFamily+  -> CQRS.Aggregator+      (CQRS.EventWithContext' (CQRS.StreamType streamFamily))+      aggregate+  -> (CQRS.StreamIdentifier streamFamily -> aggregate)+  -> T.NominalDiffTime -- ^ Lag tolerance.+  -> Int -- ^ Maximum number of elements in the cache.+  -> m (AggregateStore streamFamily aggregate)+makeAggregateStore streamFamily aggregator initialAggregate lagTolerance+                   maxSize = do+  cache <- liftIO . STM.atomically $ do+    cachedValues <- STM.newTVar HashPSQ.empty+    size <- STM.newTVar 0+    pure Cache{..}+  pure AggregateStore{..}++instance+    ( CQRS.StreamFamily m streamFamily+    , CQRS.Stream m (CQRS.StreamType streamFamily)+    , Exc.MonadError CQRS.Error m+    , Hashable (CQRS.StreamIdentifier streamFamily)+    , MonadIO m+    , Ord (CQRS.EventIdentifier (CQRS.StreamType streamFamily))+    , Ord (CQRS.StreamIdentifier streamFamily)+    , Show (CQRS.EventIdentifier (CQRS.StreamType streamFamily))+    )+    => CQRS.ReadModel m (AggregateStore streamFamily aggregate) where++  type ReadModelQuery (AggregateStore streamFamily aggregate) =+    CQRS.StreamIdentifier streamFamily++  type ReadModelResponse (AggregateStore streamFamily aggregate) =+    Response (CQRS.EventIdentifier (CQRS.StreamType streamFamily)) aggregate++  query = aggregateStoreQuery++aggregateStoreQuery+  :: forall m streamFamily aggregate.+     ( CQRS.StreamFamily m streamFamily+     , CQRS.Stream m (CQRS.StreamType streamFamily)+     , Exc.MonadError CQRS.Error m+     , Hashable (CQRS.StreamIdentifier streamFamily)+     , MonadIO m+     , Ord (CQRS.EventIdentifier (CQRS.StreamType streamFamily))+     , Ord (CQRS.StreamIdentifier streamFamily)+     , Show (CQRS.EventIdentifier (CQRS.StreamType streamFamily))+     )+  => AggregateStore streamFamily aggregate+  -> CQRS.StreamIdentifier streamFamily+  -> m (Response+        (CQRS.EventIdentifier (CQRS.StreamType streamFamily)) aggregate)+aggregateStoreQuery AggregateStore{..} streamId = do+    hpsq <- liftIO . STM.atomically . STM.readTVar . cachedValues $ cache+    now  <- liftIO T.getCurrentTime++    case HashPSQ.lookup streamId hpsq of+      Just (lastUpToDateTime, item@CacheItem{..}) -> do+        if now < T.addUTCTime lagTolerance lastUpToDateTime+          then+            pure Response+              { lastEventId = Just cachedLastEventId+              , aggregate = cachedAggregate+              , eventCount = 0+              , totalEventCount = cachedEventCount+              }+          else getAggregate now (Just item)+      Nothing -> getAggregate now Nothing++  where+    getAggregate+      :: T.UTCTime+      -> Maybe (CacheItem+                (CQRS.EventIdentifier (CQRS.StreamType streamFamily))+                aggregate)+      -> m (Response+            (CQRS.EventIdentifier (CQRS.StreamType streamFamily))+            aggregate)+    getAggregate now mPrevious = do+      let (initAggregate, bounds, eventCount) = case mPrevious of+            Just CacheItem{..} ->+              ( cachedAggregate+              , CQRS.afterEvent cachedLastEventId+              , cachedEventCount+              )+            Nothing -> (initialAggregate streamId, mempty, 0)++      stream <- CQRS.getStream streamFamily streamId+      (aggregate, mEventId, processedEventCount) <-+        CQRS.runAggregator aggregator stream bounds initAggregate++      let totalEventCount = eventCount + processedEventCount+          mkCacheItem lastEventId = CacheItem+            { cachedAggregate   = aggregate+            , cachedLastEventId = lastEventId+            , cachedEventCount  = totalEventCount+            }++      lastEventId <- liftIO $ case (mEventId, mPrevious) of+        (Nothing, Nothing) -> pure Nothing+        (Nothing, Just CacheItem{..}) -> do+          addValueToCache cache streamId now (mkCacheItem cachedLastEventId)+          pure $ Just cachedLastEventId+        (Just lastEventId, _) -> do+          addValueToCache cache streamId now (mkCacheItem lastEventId)+          pure $ Just lastEventId++      pure Response+        { lastEventId+        , aggregate+        , eventCount = processedEventCount+        , totalEventCount+        }++data CacheItem eventId aggregate = CacheItem+  { cachedAggregate   :: aggregate+  , cachedLastEventId :: eventId+  , cachedEventCount  :: Int+  }++data Cache streamId eventId aggregate = Cache+  { cachedValues+      :: STM.TVar (HashPSQ.HashPSQ+                    streamId T.UTCTime (CacheItem eventId aggregate))+  , size    :: STM.TVar Int+  , maxSize :: Int+  }++addValueToCache+  :: ( Hashable streamId+     , Ord eventId+     , Ord streamId+     )+  => Cache streamId eventId aggregate+  -> streamId+  -> T.UTCTime+  -> CacheItem eventId aggregate+  -> IO ()+addValueToCache Cache{..} streamId now item =+  STM.atomically $ do+    hpsq <- STM.readTVar cachedValues+    currentSize <- STM.readTVar size++    let (newSize, hpsq') = (\f -> HashPSQ.alter f streamId hpsq) $ \case+          Nothing -> (currentSize + 1, Just (now, item))+          Just current@(_, currentItem)+            | cachedLastEventId currentItem > cachedLastEventId item ->+                (currentSize, Just current)+            | otherwise -> (currentSize, Just (now, item))++        (newSize', hpsq'')+          | newSize > maxSize = (newSize - 1, HashPSQ.deleteMin hpsq')+          | otherwise = (newSize, hpsq')++    STM.writeTVar size newSize'+    STM.writeTVar cachedValues hpsq''
+ src/Database/CQRS/Stream.hs view
@@ -0,0 +1,185 @@+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DeriveTraversable #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE UndecidableInstances #-}++-- | Events can be written to a stream and can be streamed from it.++module Database.CQRS.Stream+  ( Stream(..)+  , WritableStream(..)+  , EventWithContext(..)+  , EventWithContext'+  , MonadMetadata(..)+  , ConsistencyCheck(..)+  , writeEvent+  , writeEventCc+  , optimistically+  , StreamBounds(..)+  , StreamBounds'+  , afterEvent+  , untilEvent+  ) where++import GHC.Generics++import qualified Control.Monad.Except as Exc+import qualified Pipes++import Database.CQRS.Error++class Stream f stream where+  -- | Type of the events contained in that stream.+  type EventType stream :: *++  -- | Type of unique identifiers for events in the stream.+  --+  -- There must be a total order on identifiers so they can be sorted.+  type EventIdentifier stream :: *++  -- | Depending on the store, this structure can contain the creation date, a+  -- correlation ID, etc.+  type EventMetadata stream :: *++  -- | Stream all the events within some bounds in arbitrary batches.+  --+  -- Events must be streamed from lowest to greatest identifier. If the back-end+  -- is fetching events in batches, they can be returned in the same way to+  -- improve performace. If the event can't be decoded, a 'Left' should be+  -- returned instead with the identifier and an error message.+  streamEvents+    :: stream+    -> StreamBounds' stream+    -> Pipes.Producer+        [ Either+            (EventIdentifier stream, String)+            (EventWithContext' stream)+        ] f ()++-- | A condition to check before inserting a new event in a stream.+--+-- This can be used to enforce consistency by checking that no new events were+-- inserted since some validation has been performed and therefore that the+-- validations are still sound.+data ConsistencyCheck identifier+  = NoConsistencyCheck -- ^ Always write the new event.+  | CheckNoEvents -- ^ There are no events in that stream.+  | CheckLastEvent identifier+    -- ^ The latest event's identifier matches.++class Stream f stream => WritableStream f stream where+  -- | Append the event to the stream and return the identifier.+  --+  -- The identifier must be greater than the previous events' identifiers.+  --+  -- The function must throw 'ConsistencyCheckError' if the check fails.+  writeEventWithMetadata+    :: stream+    -> EventType stream+    -> EventMetadata stream+    -> ConsistencyCheck (EventIdentifier stream)+    -> f (EventIdentifier stream)++-- | Once added to the stream, an event is adorned with an identifier and some+-- metadata.+data EventWithContext identifier metadata event = EventWithContext+  { identifier :: identifier+  , metadata   :: metadata+  , event      :: event+  } deriving (Eq, Show, Generic)++type EventWithContext' stream+  = EventWithContext+      (EventIdentifier stream)+      (EventMetadata stream)+      (EventType stream)++-- | The event metadata come from the current "environment".+class MonadMetadata metadata m where+  getMetadata :: m metadata++instance Monad m => MonadMetadata () m where+  getMetadata = pure ()++-- | Get the metadata from the environment, append the event to the store and+-- return the identifier.+writeEvent+  :: (Monad m, MonadMetadata (EventMetadata stream) m, WritableStream m stream)+  => stream+  -> EventType stream+  -> m (EventIdentifier stream)+writeEvent stream ev = do+  md <- getMetadata+  writeEventWithMetadata stream ev md NoConsistencyCheck++-- | Get the metadata from the environment, validate the consistency check,+-- append the event to the store and return its identifier.+writeEventCc+  :: (Monad m, MonadMetadata (EventMetadata stream) m, WritableStream m stream)+  => stream+  -> EventType stream+  -> ConsistencyCheck (EventIdentifier stream)+  -> m (EventIdentifier stream)+writeEventCc stream ev cc = do+  md <- getMetadata+  writeEventWithMetadata stream ev md cc++-- | Execute an action and retry indefinitely as long as it throws+-- 'ConsistencyCheckError'.+--+-- This makes it possible to have Optimistic Concurrency Control when writing+-- events by getting the aggregate and using 'writeEventCc' or+-- 'writeEventWithMetadata' inside the action passed to 'optimistically'.+--+-- /!\ It does NOT create a transaction when you can write several events. You+-- should only use this to write a single event!+optimistically :: Exc.MonadError Error m => m a -> m a+optimistically f = do+  f `Exc.catchError` \case+    ConsistencyCheckError _ -> optimistically f+    e -> Exc.throwError e++-- | Lower/upper bounds of an event stream.+--+-- The 'Semigroup' instance returns bounds for the intersection of the two+-- streams.+data StreamBounds identifier = StreamBounds+  { _afterEvent :: Maybe identifier -- ^ Exclusive.+  , _untilEvent :: Maybe identifier -- ^ Inclusive.+  }+  deriving (Functor, Foldable, Traversable)++type StreamBounds' stream = StreamBounds (EventIdentifier stream)++instance+    forall identifier. Ord identifier+    => Semigroup (StreamBounds identifier) where+  sb1 <> sb2 =+    StreamBounds+      { _afterEvent = combine _afterEvent max+      , _untilEvent = combine _untilEvent min+      }+    where+      combine+        :: (StreamBounds identifier -> Maybe b) -> (b -> b -> b) -> Maybe b+      combine proj merge =+        case (proj sb1, proj sb2) of+          (mx, Nothing) -> mx+          (Nothing, my) -> my+          (Just x, Just y) -> Just $ merge x y++instance Ord identifier => Monoid (StreamBounds identifier) where+  mempty = StreamBounds Nothing Nothing++-- | After the event with the given identifier, excluding it.+afterEvent :: Ord identifier => identifier -> StreamBounds identifier+afterEvent i = mempty { _afterEvent = Just i }++-- | Until the event with the given identifier, including it.+untilEvent :: Ord identifier => identifier -> StreamBounds identifier+untilEvent i = mempty { _untilEvent = Just i }
+ src/Database/CQRS/StreamFamily.hs view
@@ -0,0 +1,65 @@+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE TypeFamilies #-}++module Database.CQRS.StreamFamily+  ( StreamFamily(..)+  ) where++import qualified Pipes++import Database.CQRS.Stream++-- | A stream family is a collection of streams of the same type.+--+-- For example, a table in a relational database can contain events for all+-- aggregates of a certain type. Each aggregate has its own stream of events but+-- they are all stored in the same table. That table is a stream family indexed+-- by aggregate ID.+class StreamFamily f fam where+  -- | Type of the streams contained in this stream family.+  type StreamType fam :: *++  -- | Identifier for a specific stream, e.g. an aggregate ID.+  type StreamIdentifier fam :: *++  -- | Get the stream corresponding to a given identifier.+  getStream :: fam -> StreamIdentifier fam -> f (StreamType fam)++  -- | Initialise and return a producer of newly-created events from *all*+  -- streams in arbitrary batches. If an event can't be decoded, the decoding+  -- error is returned instead.+  --+  -- Events should appear in the correct order within a given stream but not+  -- necessarily in-between them, i.e. two events belonging to different streams+  -- won't necessarily be ordered according to the chronological history.+  --+  -- It is okay for events to be sent more than one time as long as the order+  -- is respected within each stream if it makes the implementation easier and+  -- prevents the loss of some events.+  --+  -- How events are batched together is up to the implementation as long as the+  -- order is respected.+  --+  -- It is okay for batches to be empty to signal that there are currently no+  -- new notifications. This is important for migrations, so they know they have+  -- processed all events.+  allNewEvents+    :: fam+    -> f (Pipes.Producer+          [ ( StreamIdentifier fam+            , Either+                (EventIdentifier (StreamType fam), String)+                (EventWithContext' (StreamType fam))+            ) ]+          f a)++  -- | Stream the identifier of the latest events for each stream in the family.+  --+  -- It is a snapshot of the last event identifiers at the time the producer is+  -- called. It is meant to be used by projections to catch up to the latest+  -- state.+  latestEventIdentifiers+    :: fam+    -> Pipes.Producer+        (StreamIdentifier fam, EventIdentifier (StreamType fam))+        f ()
+ src/Database/CQRS/TabularData.hs view
@@ -0,0 +1,268 @@+{-# LANGUAGE QuantifiedConstraints #-}+{-# LANGUAGE TypeFamilyDependencies #-}+{-# LANGUAGE AllowAmbiguousTypes #-}+{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE FunctionalDependencies #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE PatternSynonyms #-}+{-# LANGUAGE PolyKinds #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE TypeOperators #-}+{-# LANGUAGE UndecidableInstances #-}++-- | This module provides a backend-agnostic abstraction on top on tabular data+-- allowing projections to be decoupled from their storage.+--+-- A 'Table' is a list (if using 'Flat') or a hash map+-- (if using 'WithUniqueKeys') of 'Tuple's.++-- 'TabularDataAction's can be performed on a 'Table' in memory or a table in a+-- database with the help of an adaptor translating them in commands.+-- For example, `eventsourcing-postgresql` has a function+-- `fromTabularDataActions`.+--+-- @+-- type UserCols =+--   'WithUniqueKey+--     '[ '("user_id", Int)]+--        -- Key columns. (The backtick and the space are important for it to be+--        -- parsed correctly since the list only has one element.)+--     ['("email", String), '("admin", Bool)] -- Other columns.+--+-- type User f = Tuple f UserCols+--+-- completeUser :: User Identity+-- completeUser = 3 ~: "admin@example.com" ~: True ~: empty+--+-- incompleteUser :: User Last+-- incompleteUser =+--   field @UserCols @"admin" True+--   <> field @UserCols @"email" "admin@example.com"+--+-- userConditions :: User Conditions+-- userConditions =+--   ffield @UserCols @"admin" (equal True)+--   <> ffield @UserCols @"user_id" (lowerThan 100)+--+-- userStrings :: [(String, String)]+-- userStrings = toList @Show (maybe "NULL" show . getLast) incompleteUser+-- -- [("user_id", "NULL"), ("email", "\"admin@example.com\""), ("admin", "True")]+-- @++module Database.CQRS.TabularData+  ( TabularDataAction(..)+  , Condition(..)+  , Conditions(..)+  , equal+  , notEqual+  , lowerThan+  , lowerThanOrEqual+  , greaterThan+  , greaterThanOrEqual+  , Update(..)+  , set+  , plus+  , minus+  , Columns(..)+  , Tuple+  , Flatten+  , FlatTuple(..)+  , pattern (:~)+  , empty+  , Table+  , field+  , ffield+  , MergeSplitTuple(..)+  , applyTabularDataAction+  , GetKeyAndConditions+  , AllColumns+  , toList+  ) where++import Control.Monad (foldM)+import Data.Hashable (Hashable(..))+import Data.Kind (Type)+import Data.List (foldl')++import qualified Control.Monad.Except       as Exc+import qualified Control.Monad.Identity     as Id+import qualified Control.Monad.State.Strict as St+import qualified Data.HashMap.Strict        as HM++import Database.CQRS.TabularData.Internal++import qualified Database.CQRS as CQRS++equal :: Eq a => a -> Conditions a+equal = Conditions . pure . Equal++notEqual :: Eq a => a -> Conditions a+notEqual = Conditions . pure . NotEqual++lowerThan :: Ord a => a -> Conditions a+lowerThan = Conditions . pure . LowerThan++lowerThanOrEqual :: Ord a => a -> Conditions a+lowerThanOrEqual = Conditions . pure . LowerThanOrEqual++greaterThan :: Ord a => a -> Conditions a+greaterThan = Conditions . pure . GreaterThan++greaterThanOrEqual :: Ord a => a -> Conditions a+greaterThanOrEqual = Conditions . pure . GreaterThanOrEqual++set :: a -> Update a+set = Set++plus :: Num a => a -> Update a+plus = Plus++minus :: Num a => a -> Update a+minus = Minus++-- | Action on tabular data with an index.+--+-- Its purpose is to be used by an 'EffectfulProjection' to create persisting+-- backend-agnostic projections.+data TabularDataAction (cols :: Columns) where+  Insert :: Tuple Id.Identity cols -> TabularDataAction cols+  Update :: Tuple Update cols -> Tuple Conditions cols -> TabularDataAction cols+  Upsert+    :: Tuple Id.Identity ('WithUniqueKey keyCols cols)+    -> TabularDataAction ('WithUniqueKey keyCols cols)+    -- ^ Insert a new row or update the row with the same key if it exists.+  Delete :: Tuple Conditions cols -> TabularDataAction cols++deriving instance+  AllColumns Show (Flatten cols) => Show (TabularDataAction cols)++-- | In-memory table that supports 'TabularDataAction'.+-- See 'applyTabularDataAction'.+type family Table (cols :: Columns) :: Type where+  Table ('WithUniqueKey keyCols cols) =+    HM.HashMap (FlatTuple Id.Identity keyCols) (FlatTuple Id.Identity cols)+  Table ('Flat cols) = [FlatTuple Id.Identity cols]++class ApplyTabularDataAction f (cols :: Columns) where+  -- | Apply some 'TabularDataAction' on an in-memory table and return a new+  -- table.+  applyTabularDataAction+    :: Table cols -> TabularDataAction cols -> f (Table cols)++instance Applicative f => ApplyTabularDataAction f ('Flat cols) where+  applyTabularDataAction tbl = pure . \case+    Insert tuple -> tuple : tbl+    Update updates conditions ->+      map (\tuple ->+            if tuple `matches` conditions+              then update updates tuple+              else tuple) tbl+    Delete conditions -> filter (`matches` conditions) tbl++instance+    ( AllColumns Show keyCols+    , Exc.MonadError CQRS.Error m+    , GetKeyAndConditions keyCols cols+    , Hashable (FlatTuple Id.Identity keyCols)+    , Ord (FlatTuple Id.Identity keyCols)+    , MergeSplitTuple keyCols cols+    )+    => ApplyTabularDataAction m ('WithUniqueKey keyCols cols) where++  applyTabularDataAction tbl = \case+    Insert tuple -> do+      let (keyTuple, otherTuple) = splitTuple tuple+          op = \case+            Nothing -> pure $ Just otherTuple+            Just _ -> Exc.throwError . CQRS.ProjectionError $+              "duplicate key on insert: " ++ show keyTuple+      HM.alterF op keyTuple tbl++    Update updates conditions ->+      case getKeyAndConditions conditions of+        Just (keyTuple, otherConditions) ->+          case HM.lookup keyTuple tbl of+            Nothing -> pure tbl+            Just otherTuple+              | otherTuple `matches` otherConditions -> do+                  let (keyTuple', otherTuple') =+                        splitTuple . update updates . mergeTuple+                          $ (keyTuple, otherTuple)+                  if keyTuple == keyTuple'+                    then pure $ HM.insert keyTuple otherTuple' tbl+                    else+                      case HM.lookup keyTuple' tbl of+                        Nothing ->+                          pure . HM.delete keyTuple+                            . HM.insert keyTuple' otherTuple' $ tbl+                        Just _ ->+                          Exc.throwError . CQRS.ProjectionError $+                            "duplicate key on update: " ++ show keyTuple'+              | otherwise -> pure tbl++        -- It traverses the hash map collecting changing values if the key+        -- doesn't change but the row matches the condition. When the key+        -- changes, it keeps track of the key that has to be deleted and the new+        -- row that has to be inserted. After the traversal, it deletes all the+        -- rows in the first list and inserts all the rows from the second+        -- checking for duplicated keys.+        Nothing -> do+          let step keyTuple otherTuple = do+                let merged = mergeTuple (keyTuple, otherTuple)+                if merged `matches` conditions+                  then do+                    let (keyTuple', otherTuple') =+                          splitTuple $ update updates merged+                    if keyTuple == keyTuple'+                      then pure $ Just otherTuple'+                      else do+                        St.modify' $ \(tbd, tbi) ->+                          (keyTuple : tbd, (keyTuple', otherTuple') : tbi)+                        pure $ Just otherTuple+                  else pure $ Just otherTuple++              (toBeDeleted, toBeInserted) =+                St.execState (HM.traverseWithKey step tbl) ([], [])++              tbl' = foldl' (flip HM.delete) tbl toBeDeleted++          foldM+            (\t (keyTuple, otherTuple) ->+              case HM.lookup keyTuple t of+                Just _ -> Exc.throwError . CQRS.ProjectionError $+                  "duplicate key on update: " ++ show keyTuple+                Nothing -> pure . HM.insert keyTuple otherTuple $ t+            )+            tbl' toBeInserted++    Upsert tuple -> do+      let (keyTuple, otherTuple) = splitTuple tuple+      pure $ HM.insert keyTuple otherTuple tbl++    Delete conditions ->+      case getKeyAndConditions conditions of+        Just (keyTuple, otherConditions) -> do+          let op = \case+                Nothing -> Nothing+                Just otherTuple+                  | otherTuple `matches` otherConditions -> Nothing+                  | otherwise -> Just otherTuple+          pure . HM.alter op keyTuple $ tbl++        Nothing -> do+          let op keyTuple otherTuple+                | mergeTuple (keyTuple, otherTuple) `matches` conditions =+                    Nothing+                | otherwise = Just otherTuple+          pure . HM.mapMaybeWithKey op $ tbl
+ src/Database/CQRS/TabularData/Internal.hs view
@@ -0,0 +1,299 @@+{-# LANGUAGE PatternSynonyms #-}+{-# LANGUAGE AllowAmbiguousTypes #-}+{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE FunctionalDependencies #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE UndecidableInstances #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE KindSignatures #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE PolyKinds #-}+{-# LANGUAGE QuantifiedConstraints #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE TypeOperators #-}++module Database.CQRS.TabularData.Internal where++import Data.Hashable (Hashable(..))+import Data.Kind (Constraint, Type)+import Data.Proxy (Proxy(..))+import GHC.TypeLits (symbolVal, KnownSymbol, Symbol)++import qualified Control.Monad.Identity as Id++data Condition a where+  Equal              :: Eq a => a -> Condition a+  NotEqual           :: Eq a =>  a -> Condition a+  LowerThan          :: Ord a => a -> Condition a+  LowerThanOrEqual   :: Ord a => a -> Condition a+  GreaterThan        :: Ord a => a -> Condition a+  GreaterThanOrEqual :: Ord a => a -> Condition a++deriving instance Show a => Show (Condition a)+deriving instance Eq (Condition a)++newtype Conditions a+  = Conditions { getConditions :: [Condition a] }+  deriving newtype (Eq, Show)++instance Semigroup (Conditions a) where+  Conditions cs <> Conditions cs' = Conditions $ cs <> cs'++instance Monoid (Conditions a) where+  mempty = Conditions []++instance Eq a => Wrapper Conditions a where+  wrap = Conditions . pure . Equal++-- | Kind of types that describe columns of a table.+--+-- This is not intended as a type. It's promoted to a kind.+--+-- Use 'Flat' for simple tables and 'WithUniqueKey' if you want to be able to+-- do upserts and/or store tuples in memory with a hash map instead of a list.+data Columns where+  Flat :: [(Symbol, Type)] -> Columns+  WithUniqueKey :: [(Symbol, Type)] -> [(Symbol, Type)] -> Columns++type family Flatten (cols :: k) :: [(Symbol, Type)] where+  Flatten ('WithUniqueKey (col ': keyCols) cols) =+    col ': Flatten ('WithUniqueKey keyCols cols)+  Flatten ('WithUniqueKey '[] cols) = cols+  Flatten ('Flat cols) = cols++-- | A named tuple representing a row in the table.+type Tuple f cols = FlatTuple f (Flatten cols)++data FlatTuple :: (Type -> Type) -> [(Symbol, Type)] -> Type where+  Nil  :: FlatTuple f '[]+  Cons :: f a -> FlatTuple f cols -> FlatTuple f ('(sym, a) ': cols)++pattern (:~)+  :: a+  -> FlatTuple Id.Identity cols+  -> FlatTuple Id.Identity ('(sym, a) ': cols)+pattern x :~ xs = Cons (Id.Identity x) xs++infixr 5 :~++empty :: FlatTuple f '[]+empty = Nil++instance Eq (FlatTuple f '[]) where+  Nil == Nil = True++instance+    (Eq (f a), Eq (FlatTuple f cols))+    => Eq (FlatTuple f ('(sym, a) ': cols)) where+  Cons x xs == Cons y ys = x == y && xs == ys++instance Ord (FlatTuple f '[]) where+  compare Nil Nil = EQ++instance+    (Ord (f a), Ord (FlatTuple f cols))+    => Ord (FlatTuple f ('(sym, a) ': cols)) where+  compare (Cons x xs) (Cons y ys) =+    case compare x y of+      EQ -> compare xs ys+      res -> res++instance Hashable (FlatTuple f '[]) where+  hashWithSalt salt Nil = salt++instance+    (Hashable (f a), Hashable (FlatTuple f cols))+    => Hashable (FlatTuple f ('(sym, a) ': cols)) where+  hashWithSalt salt (Cons x xs) =+    hashWithSalt (hashWithSalt salt x) xs++instance+    ( AllColumns Show cols, forall a. Show a => Show (f a) )+    => Show (FlatTuple f cols) where+  show = show . toList @Show (\name value -> show (name, value))++instance Semigroup (FlatTuple f '[]) where+  Nil <> Nil = Nil++instance+    (Semigroup (f a), Semigroup (FlatTuple f cols))+    => Semigroup (FlatTuple f ('(sym, a) ': cols)) where+  Cons x xs <> Cons y ys = Cons (x <> y) (xs <> ys)++instance Monoid (FlatTuple f '[]) where+  mempty = Nil++instance+    (Monoid (f a), Monoid (FlatTuple f xs))+    => Monoid (FlatTuple f ('(sym, a) ': xs)) where+  mempty = Cons mempty mempty++class Field f (sym :: Symbol) a (cols :: [(Symbol, Type)]) | cols sym -> a where+  cfield :: proxy sym -> f a -> FlatTuple f cols++instance Monoid (FlatTuple f cols) => Field f sym a ('(sym, a) : cols) where+  cfield _ x = Cons x mempty++instance+    {-# OVERLAPPABLE #-}+    (Monoid (f b), Field f sym a cols)+    => Field f sym a ('(sym', b) : cols) where+  cfield proxy x = Cons mempty (cfield proxy x)++class Wrapper f a where+  wrap :: a -> f a++-- | Create a tuple with the given field set to the given value wrapped into+-- @f@. In practice, @f@ is 'Conditions' or 'Update'.+--+-- It is meant to be used together with @TypeApplications@, e.g.+-- @+-- field @"field_name" value+-- @+field+  :: forall cols sym f a.+     (Field f sym a (Flatten cols), Wrapper f a)+  => a -> Tuple f cols+field value = cfield (Proxy :: Proxy sym) (wrap value)++-- | Create a tuple with the given field set to the given "wrapped" value.+--+-- It is more flexible than 'field' but less convenient to use if the goal is to+-- simply wrap the value inside the 'Applicative'. In particular, it can be used+-- with 'Conditions' such as+-- @+-- ffield @"email" (equal "someone@example.com")+-- @+ffield+  :: forall cols sym f a. Field f sym a (Flatten cols)+  => f a -> Tuple f cols+ffield fvalue = cfield (Proxy :: Proxy sym) fvalue++class MergeSplitTuple keyCols cols where+  mergeTuple+    :: (FlatTuple f keyCols, FlatTuple f cols)+    -> Tuple f ('WithUniqueKey keyCols cols)++  splitTuple+    :: Tuple f ('WithUniqueKey keyCols cols)+    -> (FlatTuple f keyCols, FlatTuple f cols)++instance MergeSplitTuple '[] cols where+  mergeTuple (Nil, tuple) = tuple+  splitTuple tuple = (Nil, tuple)++instance+    MergeSplitTuple keyCols cols+    => MergeSplitTuple (a ': keyCols) cols where+  mergeTuple (Cons x xs, tuple') = Cons x (mergeTuple (xs, tuple'))+  splitTuple (Cons x xs) =+    let (tuple, tuple') = splitTuple xs in (Cons x tuple, tuple')++type family AllColumns+    (cs :: Type -> Constraint) (cols :: [(Symbol, Type)]) :: Constraint where+  AllColumns _ '[] = ()+  AllColumns cs ('(sym, a) ': cols) =+    (cs a, KnownSymbol sym, AllColumns cs cols)++-- | Transform a tuple into a list of pairs given a function to transform the+-- field values.+--+-- @cs@ is some constraint that the values need to satisfy. For example,+-- @+-- toList @Show (\name value -> (name, maybe "NULL" show (getLast value)))+--   :: Tuple Last cols -> [(String, String)]+-- @+toList+  :: forall cs cols f b. AllColumns cs cols+  => (forall a. cs a => String -> f a -> b) -> FlatTuple f cols -> [b]+toList f = \case+    Nil -> []+    pair@(Cons _ _) -> go Proxy pair+  where+    go+      :: (KnownSymbol sym, cs a, AllColumns cs cols')+      => Proxy sym -> FlatTuple f ('(sym, a) ': cols') -> [b]+    go proxy (Cons x xs) =+      f (symbolVal proxy) x : toList @cs f xs++-- | Used to optimise operations on the in-memory storage. When we want to+-- update or delete rows based on some conditions that would match one row+-- matching its key, it's more efficient to use 'HM.alter' instead of traversing+-- the hash map.+class GetKeyAndConditions keyCols cols where+  getKeyAndConditions+    :: Tuple Conditions ('WithUniqueKey keyCols cols)+    -> Maybe (FlatTuple Id.Identity keyCols, FlatTuple Conditions cols)++instance GetKeyAndConditions '[] cols where+  getKeyAndConditions conds = Just (Nil, conds)++instance+    GetKeyAndConditions keyCols cols+    => GetKeyAndConditions ('(sym, a) ': keyCols) cols where+  getKeyAndConditions (Cons cond conds) =+    case getConditions cond of+      [Equal x] -> do+        (tuple, otherConditions) <- getKeyAndConditions conds+        pure (x :~ tuple, otherConditions)+      _ -> Nothing++matches+  :: FlatTuple Id.Identity cols+  -> FlatTuple Conditions cols+  -> Bool+matches = curry $ \case+  (Nil, Nil) -> True+  (Cons (Id.Identity x) xs, Cons (Conditions conds) ys) ->+    all (matchesCond x) conds && xs `matches` ys++matchesCond :: a -> Condition a -> Bool+matchesCond x = \case+  Equal y -> x == y+  NotEqual y -> x /= y+  LowerThan y -> x < y+  LowerThanOrEqual y -> x <= y+  GreaterThan y -> x > y+  GreaterThanOrEqual y -> x >= y++data Update a where+  NoUpdate :: Update a+  Set      :: a -> Update a+  Plus     :: Num a => a -> Update a+  Minus    :: Num a => a -> Update a++deriving instance Show a => Show (Update a)++instance Semigroup (Update a) where+  u1 <> u2 =+    case u2 of+      NoUpdate -> u1+      _ -> u2++instance Monoid (Update a) where+  mempty = NoUpdate++instance Wrapper Update a where+  wrap = Set++update+  :: FlatTuple Update cols+  -> FlatTuple Id.Identity cols+  -> FlatTuple Id.Identity cols+update = curry $ \case+  (Nil, Nil) -> Nil+  (Cons NoUpdate xs, Cons y ys) -> Cons y (update xs ys)+  (Cons (Set x) xs, Cons _ ys) -> Cons (pure x) (update xs ys)+  (Cons (Plus x) xs, Cons (Id.Identity y) ys) ->+    Cons (pure (y+x)) (update xs ys)+  (Cons (Minus x) xs, Cons (Id.Identity y) ys) ->+    Cons (pure (y-x)) (update xs ys)
+ src/Database/CQRS/TabularData/Optimisation.hs view
@@ -0,0 +1,108 @@+{-# LANGUAGE UndecidableInstances #-}+{-# LANGUAGE InstanceSigs #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE KindSignatures #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE ScopedTypeVariables #-}++module Database.CQRS.TabularData.Optimisation+  ( Optimiser+  , Optimisable(..)+  ) where++import qualified Control.Monad.Identity as Id++import Database.CQRS.TabularData+import Database.CQRS.TabularData.Internal++-- TODO: Consecutive updates with the same conditions should be merged.+-- TODO: Inserts that will be updated later on should insert the final version.+--       => Reordering of the actions to put the insert at the end.++type Optimiser cols = [TabularDataAction cols] -> [TabularDataAction cols]++class Optimisable (cols :: Columns) where+  optimiseActions :: Optimiser cols+  optimiseActions = optimiseInsertBeforeDelete++  optimiseInsertBeforeDelete :: Optimiser cols++instance Optimisable ('Flat cols) where+  optimiseInsertBeforeDelete :: Optimiser ('Flat cols)+  optimiseInsertBeforeDelete =+      foldr+        (\action actions ->+          case action of+            Insert tuple | isInsertBeforeDelete tuple actions -> actions+            _ -> action: actions)+        []++    where+      isInsertBeforeDelete+        :: Tuple Id.Identity ('Flat cols)+        -> [TabularDataAction ('Flat cols)]+        -> Bool+      isInsertBeforeDelete tuple = \case+        [] -> False+        Delete conds : actions ->+          tuple `matches` conds || isInsertBeforeDelete tuple actions+        Update updates conds : actions+          | tuple `matches` conds ->+              let tuple' = update updates tuple+              in isInsertBeforeDelete tuple' actions+          | otherwise -> isInsertBeforeDelete tuple actions+        Insert _ : actions -> isInsertBeforeDelete tuple actions++instance+    ( Eq (FlatTuple Id.Identity keyCols)+    , MergeSplitTuple keyCols cols+    )+    => Optimisable ('WithUniqueKey keyCols cols) where++  optimiseInsertBeforeDelete+    ::+      ( Eq (FlatTuple Id.Identity keyCols)+      , MergeSplitTuple keyCols cols+      )+    => Optimiser ('WithUniqueKey keyCols cols)+  optimiseInsertBeforeDelete =+      foldr+        (\action actions ->+          case action of+            Insert tuple | isInsertBeforeDelete tuple actions -> actions+            _ -> action: actions)+        []++    where+      isInsertBeforeDelete+        :: Tuple Id.Identity ('WithUniqueKey keyCols cols)+        -> [TabularDataAction ('WithUniqueKey keyCols cols)]+        -> Bool+      isInsertBeforeDelete tuple = \case+        [] -> False+        Delete conds : actions ->+          tuple `matches` conds || isInsertBeforeDelete tuple actions+        Upsert tuple' : actions -> isUpsertBeforeDelete tuple tuple' actions+        Update updates conds : actions+          | tuple `matches` conds ->+              let tuple' = update updates tuple+              in isInsertBeforeDelete tuple' actions+          | otherwise -> isInsertBeforeDelete tuple actions+        Insert _ : actions -> isInsertBeforeDelete tuple actions++      isUpsertBeforeDelete+        :: Tuple Id.Identity ('WithUniqueKey keyCols cols)+        -> Tuple Id.Identity ('WithUniqueKey keyCols cols)+        -> [TabularDataAction ('WithUniqueKey keyCols cols)]+        -> Bool+      isUpsertBeforeDelete tuple tuple' actions =+        let keyTuple, keyTuple' :: FlatTuple Id.Identity keyCols+            _otherTuple, _otherTuple' :: FlatTuple Id.Identity cols+            (keyTuple, _otherTuple) = splitTuple tuple+            (keyTuple', _otherTuple') = splitTuple tuple'+        in+        isInsertBeforeDelete+          (if keyTuple == keyTuple' then tuple' else tuple)+          actions
+ src/Database/CQRS/TaskManager.hs view
@@ -0,0 +1,144 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TupleSections #-}++module Database.CQRS.TaskManager+  ( TaskManager+  , Task(..)+  , runTaskManager+  ) where++import Control.Concurrent (threadDelay)+import Control.Monad (forever, when)+import Control.Monad.Trans+import Data.Hashable (Hashable)+import Data.List (find, sortOn)+import Data.Maybe (isNothing, mapMaybe)+import Pipes ((>->))++import qualified Control.Monad.Except       as Exc+import qualified Control.Monad.State.Strict as St+import qualified Data.Time                  as T+import qualified Pipes++import qualified Database.CQRS                          as CQRS+import qualified Database.CQRS.ReadModel.AggregateStore as CQRS.AS++-- | A task as accumulated by a 'TaskManager'.+data Task action = Task+  { action :: action+  , from   :: Maybe T.UTCTime -- ^ Time from which the task can be run.+  }++-- | Projection aggregating a list of tasks from a stream of events together+-- with some state.+type TaskManager event action st =+  CQRS.Aggregator event ([Task action], st)++-- | Repeatedly loop through the streams and run the tasks.+--+-- If it finds no work, it waits 5 minutes before the next run or less if it+-- knows of an earlier planned task.+runTaskManager+  :: forall streamFamily action st m.+     ( Exc.MonadError CQRS.Error m+     , CQRS.Stream m (CQRS.StreamType streamFamily)+     , CQRS.StreamFamily m streamFamily+     , Hashable (CQRS.StreamIdentifier streamFamily)+     , MonadIO m+     , Ord (CQRS.EventIdentifier (CQRS.StreamType streamFamily))+     , Ord (CQRS.StreamIdentifier streamFamily)+     , Show (CQRS.EventIdentifier (CQRS.StreamType streamFamily))+     )+  => streamFamily+  -> (CQRS.StreamIdentifier streamFamily -> st)+     -- ^ Initial state of the task manager for any given stream.+  -> TaskManager+      (CQRS.EventWithContext' (CQRS.StreamType streamFamily))+      action st+  -> (action -> m ())+  -> m ()+runTaskManager streamFamily mkInitState taskManager runAction = do+  as <- CQRS.AS.makeAggregateStore+    streamFamily taskManager (([],) . mkInitState) 0 1000++  forever $ do+    -- Next run in 5 minutes (unless changed.)+    initialNextRun <- T.addUTCTime (5 * 60) <$> liftIO T.getCurrentTime++    mNextRun <-+      flip St.execStateT (Just initialNextRun) . Pipes.runEffect $ do+        Pipes.hoist lift (CQRS.latestEventIdentifiers streamFamily)+        >-> streamProcessor as++    case mNextRun of+      Nothing -> pure () -- Do not wait.+      Just nextRun -> liftIO $ do+        now <- T.getCurrentTime+        when (now < nextRun) $+          threadDelay . (1000000*) . round $ T.diffUTCTime nextRun now++  where+    streamProcessor+      :: CQRS.AS.AggregateStore streamFamily ([Task action], st)+      -> Pipes.Consumer+          ( CQRS.StreamIdentifier streamFamily+          , CQRS.EventIdentifier (CQRS.StreamType streamFamily)+          )+          (St.StateT (Maybe T.UTCTime) m) ()+    streamProcessor as = do+      -- We don't actually care about the latest event identifier as it might+      -- have changed. We used the aggregate store to fetch the latest version.+      (streamId, _) <- Pipes.await++      lift . lift $ runSomeImmediateTasks as streamId 5+      mNextTaskTime <- lift . lift $ runPlannedTasks as streamId+      tasks <- lift . lift $ fst . CQRS.AS.aggregate <$> CQRS.query as streamId++      -- Do not wait after this run if there are some immediate tasks left.+      when (any (isNothing . from) tasks) $+        St.modify' $ const Nothing++      -- Wait less time if we know that there is an earlier upcoming task.+      case mNextTaskTime of+        Just t -> St.modify' $ fmap (min t)+        Nothing -> pure ()++    -- Run n immediate tasks.+    -- It refetches the tasks after each run since executing one task can change+    -- the list of tasks itself.+    runSomeImmediateTasks+      :: CQRS.AS.AggregateStore streamFamily ([Task action], st)+      -> CQRS.StreamIdentifier streamFamily+      -> Int+      -> m ()+    runSomeImmediateTasks as streamId n+      | n > 0 = do+          tasks <- fst . CQRS.AS.aggregate <$> CQRS.query as streamId++          case find (isNothing . from) tasks of+            Nothing -> pure ()+            Just task -> do+              runAction . action $ task+              runSomeImmediateTasks as streamId (n-1)++      | otherwise = pure ()++    -- Run planned tasks and return the time of the earliest task planned in+    -- the future if any.+    runPlannedTasks+      :: CQRS.AS.AggregateStore streamFamily ([Task action], st)+      -> CQRS.StreamIdentifier streamFamily+      -> m (Maybe T.UTCTime)+    runPlannedTasks as streamId = do+      tasks <- fst . CQRS.AS.aggregate <$> CQRS.query as streamId+      now <- liftIO T.getCurrentTime++      case sortOn fst . mapMaybe (\Task{..} -> (, action) <$> from) $ tasks of+        (t, action) : _+          | now < t -> do+              runAction action+              runPlannedTasks as streamId+          | otherwise -> pure $ Just t+        _ -> pure Nothing
+ src/Database/CQRS/Transformer.hs view
@@ -0,0 +1,288 @@+{-# LANGUAGE DeriveFunctor #-}+{-# LANGUAGE ExistentialQuantification #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE TupleSections #-}+{-# LANGUAGE TypeFamilies #-}++module Database.CQRS.Transformer+  ( -- * Transformed stream+    Transformer+  , TransformedStream+  , transformStream++    -- * Transformed stream family+  , TransformedStreamFamily+  , transformStreamFamily++    -- * Transform monad+  , Transform+  , pushEvent+  , mergeEvents+  , flushEvents+  , failTransformer+  ) where++import Control.Monad (forM_)+import Control.Monad.Trans (lift)+import Data.Functor ((<&>))+import Data.Hashable (Hashable)++import qualified Data.HashMap.Strict as HM+import qualified Control.Monad.Free  as Free+import qualified Control.Monad.State as St+import qualified Pipes+import qualified Pipes.Prelude       as Pipes++import qualified Database.CQRS.Stream       as CQRS+import qualified Database.CQRS.StreamFamily as CQRS++type Transformer inputEvent eventId event =+  inputEvent -> Transform eventId event ()++data TransformedStream m identifier metadata event =+  forall stream. CQRS.Stream m stream => TransformedStream+    { transformer+        :: Transformer+            (Either+              (CQRS.EventIdentifier stream, String)+              (CQRS.EventWithContext' stream))+            identifier (CQRS.EventWithContext identifier metadata event)+    , inputStream :: stream+    , reverseIdentifier :: identifier -> m (CQRS.EventIdentifier stream)+    }++transformStream+  :: CQRS.Stream m stream+  => Transformer+      (Either+        (CQRS.EventIdentifier stream, String)+        (CQRS.EventWithContext' stream))+      identifier (CQRS.EventWithContext identifier metadata event)+  -> (identifier -> m (CQRS.EventIdentifier stream))+  -> stream+  -> TransformedStream m identifier metadata event+transformStream transformer reverseIdentifier inputStream =+  TransformedStream{..}++instance+    Monad m+    => CQRS.Stream m (TransformedStream m identifier metadata event) where+  type EventType (TransformedStream m identifier metadata event) = event+  type EventIdentifier (TransformedStream m identifier metadata event) =+    identifier+  type EventMetadata (TransformedStream m identifier metadata event) = metadata++  streamEvents = transformedStreamStreamEvents++transformedStreamStreamEvents+  :: Monad m+  => TransformedStream m identifier metadata event+  -> CQRS.StreamBounds identifier+  -> Pipes.Producer+      [ Either+          (identifier, String)+          (CQRS.EventWithContext identifier metadata event)+      ] m ()+transformedStreamStreamEvents TransformedStream{..} bounds = do+  inputBounds <- lift $ traverse reverseIdentifier bounds+  let inputs = Pipes.hoist lift $ CQRS.streamEvents inputStream inputBounds++  finalEvents <-+    (\f -> Pipes.foldM f (pure []) pure inputs) $ \waitingEvents batch -> do+      let (batches, waitingEvents') =+            runTransform waitingEvents . mapM_ transformer $ batch+      Pipes.each $ batches <&> \case+        Left err -> [Left err]+        Right b -> map Right b++      -- If there are too many waiting events, we flush them anyway to avoid+      -- using too much memmory.+      case drop 100 waitingEvents' of+        [] -> pure waitingEvents'+        _ -> do+          Pipes.yield $ map Right waitingEvents'+          pure []++  Pipes.yield $ map Right finalEvents++data TransformedStreamFamily m streamId eventId metadata event =+  forall streamFamily.+    ( Hashable (CQRS.StreamIdentifier streamFamily)+    , Ord (CQRS.StreamIdentifier streamFamily)+    , CQRS.Stream m (CQRS.StreamType streamFamily)+    , CQRS.StreamFamily m streamFamily+    )+  => TransformedStreamFamily+    { inputStreamFamily :: streamFamily+    , streamTransformer+        :: Transformer+            (Either+              (CQRS.EventIdentifier (CQRS.StreamType streamFamily), String)+              (CQRS.EventWithContext' (CQRS.StreamType streamFamily)))+            eventId (CQRS.EventWithContext eventId metadata event)+    , makeStreamIdentifier+        :: CQRS.StreamIdentifier streamFamily -> m streamId+    , reverseStreamIdentifier+        :: streamId -> m (CQRS.StreamIdentifier streamFamily)+    , makeEventIdentifier+        :: CQRS.EventIdentifier (CQRS.StreamType streamFamily) -> m eventId+    , reverseEventIdentifier+        :: eventId -> m (CQRS.EventIdentifier (CQRS.StreamType streamFamily))+    }++transformStreamFamily+  :: forall m streamId eventId metadata event streamFamily.+     ( Hashable (CQRS.StreamIdentifier streamFamily)+     , Ord (CQRS.StreamIdentifier streamFamily)+     , CQRS.Stream m (CQRS.StreamType streamFamily)+     , CQRS.StreamFamily m streamFamily+     )+  => Transformer+      (Either+        (CQRS.EventIdentifier (CQRS.StreamType streamFamily), String)+        (CQRS.EventWithContext' (CQRS.StreamType streamFamily)))+      eventId (CQRS.EventWithContext eventId metadata event)+  -> (CQRS.StreamIdentifier streamFamily -> m streamId)+  -> (streamId -> m (CQRS.StreamIdentifier streamFamily))+  -> (CQRS.EventIdentifier (CQRS.StreamType streamFamily) -> m eventId)+  -> (eventId -> m (CQRS.EventIdentifier (CQRS.StreamType streamFamily)))+  -> streamFamily+  -> TransformedStreamFamily m streamId eventId metadata event+transformStreamFamily streamTransformer+                      makeStreamIdentifier reverseStreamIdentifier+                      makeEventIdentifier reverseEventIdentifier+                      inputStreamFamily =+  TransformedStreamFamily{..}++instance+    Monad m+    => CQRS.StreamFamily m+        (TransformedStreamFamily m streamId eventId metadata event) where++  type StreamType (TransformedStreamFamily m streamId eventId metadata event) =+    TransformedStream m eventId metadata event++  type StreamIdentifier+        (TransformedStreamFamily m streamId eventId metadata event) = streamId++  getStream = transformedStreamFamilyGetStream+  allNewEvents = transformedStreamFamilyAllNewEvents+  latestEventIdentifiers = transformedStreamFamilyLatestEventIdentifiers++transformedStreamFamilyGetStream+  :: Monad m+  => TransformedStreamFamily m streamId eventId metadata event+  -> streamId+  -> m (TransformedStream m eventId metadata event)+transformedStreamFamilyGetStream TransformedStreamFamily{..} streamId = do+  inputStreamId <- reverseStreamIdentifier streamId+  inputStream <- CQRS.getStream inputStreamFamily inputStreamId+  pure $ transformStream streamTransformer reverseEventIdentifier inputStream++transformedStreamFamilyAllNewEvents+  :: Monad m+  => TransformedStreamFamily m streamId eventId metadata event+  -> m (Pipes.Producer+          [ ( streamId+            , Either+                (eventId, String)+                (CQRS.EventWithContext eventId metadata event)+            ) ]+          m a)+transformedStreamFamilyAllNewEvents TransformedStreamFamily{..} = do+  inputNewEvents <- CQRS.allNewEvents inputStreamFamily+  pure $ Pipes.for inputNewEvents $ \batch -> do+    let eventsByStream =+          HM.toList . HM.fromListWith (++) . map (fmap pure) $ batch+    forM_ eventsByStream $ \(inputStreamId, events) -> do+      streamId <- lift $ makeStreamIdentifier inputStreamId+      let (batches, waitingEvents) =+            runTransform [] . mapM_ streamTransformer $ events+      Pipes.each $ (batches ++ [Right waitingEvents]) <&> \case+        Left err -> [(streamId, Left err)]+        Right b -> map ((streamId,) . Right) b++transformedStreamFamilyLatestEventIdentifiers+  :: Monad m+  => TransformedStreamFamily m streamId eventId metadata event+  -> Pipes.Producer (streamId, eventId) m ()+transformedStreamFamilyLatestEventIdentifiers TransformedStreamFamily{..} = do+  let inputIds = CQRS.latestEventIdentifiers inputStreamFamily+  Pipes.for inputIds $ \(inputStreamId, inputEventId) -> do+    streamId <- lift $ makeStreamIdentifier inputStreamId+    eventId  <- lift $ makeEventIdentifier  inputEventId+    Pipes.yield (streamId, eventId)++data TransformF eventId event a+  = PushEvent a event+  | MergeEvents ([event] -> (a, [event]))+  | FlushEvents a+  | Failure a eventId String+  deriving Functor++-- | Monad in which you can push, merge and flush events.+type Transform eventId event = Free.Free (TransformF eventId event)++-- | Run the transformation starting with some waiting events and returning+-- batches of events to flush and a new list of waiting events to be fed to the+-- next call. If the transformer fails, return the error instead.+--+-- All given events will be processed except in case of failure.+runTransform+  :: [event] -> Transform eventId event ()+  -> ([Either (eventId, String) [event]], [event])+runTransform lastWaitingEvents =+    flip St.execState ([], lastWaitingEvents) . Free.foldFree interpreter++  where+    interpreter+      :: TransformF eventId event a+      -> St.State ([Either (eventId, String) [event]], [event]) a+    interpreter = \case+      PushEvent x event -> do+        St.modify $ \(batches, waitingEvents) ->+          (batches, waitingEvents ++ [event])+        pure x++      MergeEvents f ->+        St.state $ \(batches, waitingEvents) ->+          let (x, waitingEvents') = f waitingEvents+              st = (batches, waitingEvents')+          in (x, st)++      FlushEvents x -> do+        St.modify $ \(batches, waitingEvents) ->+          (batches ++ [Right waitingEvents], [])+        pure x++      Failure x eventId err -> do+        St.modify $ \(batches, waitingEvents) ->+          (batches ++ [Right waitingEvents, Left (eventId, err)], [])+        pure x++-- | Push a new event at the end of the queue.+pushEvent :: event -> Transform eventId event ()+pushEvent = Free.liftF . PushEvent ()++-- | Apply a function to the queue of event returning a value and a new queue,+-- sets the queue to the new one and return the value.+--+-- The intent is to allow a new event to be merged in a previous one if possible+-- to make the new event stream more compact.+mergeEvents :: ([event] -> (a, [event])) -> Transform eventId event a+mergeEvents = Free.liftF . MergeEvents++-- | Flush the queue so it can be processed downstream, e.g. sent to a message+-- broker.+--+-- Flushing may also occur automatically.+flushEvents :: Transform eventId event ()+flushEvents = Free.liftF $ FlushEvents ()++-- | Flush the events and push an error downstream.+failTransformer :: eventId -> String -> Transform eventId event ()+failTransformer eventId err = Free.liftF $ Failure () eventId err
+ test/Database/CQRS/InMemoryTest.hs view
@@ -0,0 +1,136 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE TypeApplications #-}++module Database.CQRS.InMemoryTest+  ( tests+  ) where++import Control.Concurrent (forkIO, killThread, threadDelay)+import Control.Concurrent.MVar+import Control.Monad (forever, replicateM_, void, zipWithM_)+import Control.Monad.Trans (liftIO)+import Data.Foldable (for_, traverse_)+import Data.List (nub)+import Data.Traversable (for)+import Hedgehog hiding (collect)+import Pipes ((<-<))+import System.Mem (performGC)+import Test.Tasty+import Test.Tasty.Hedgehog++import qualified Control.Monad.Except as Exc+import qualified Hedgehog.Gen         as Gen+import qualified Hedgehog.Range       as Range+import qualified Pipes+import qualified Pipes.Prelude        as Pipes++import Helpers (collect)++import qualified Database.CQRS          as CQRS+import qualified Database.CQRS.InMemory as CQRS.InMem++tests :: TestTree+tests = testGroup "In memory"+  [ testProperty "All events are written exactly once" $+      property writtenExactlyOnce+  , testProperty "Writing an event sends a notification" $+      property writingSendsNotification+  , testProperty+      "GC'ing notification stream removes it from the stream family" $+      withTests 1 $ property gcRemovesListener+  ]++writtenExactlyOnce :: PropertyT IO ()+writtenExactlyOnce = do+  streamIds <- forAll $+    nub <$> Gen.list (Range.linear 1 100) (Gen.int (Range.linear 1 1000))+  chunks <- for streamIds $ \streamId -> do+    events <- forAll $+      Gen.list (Range.linear 1 100) (Gen.int (Range.linear 1 100))+    mvar   <- liftIO newEmptyMVar+    pure (streamId, events, mvar)++  streamFamily <- CQRS.InMem.emptyStreamFamily @()++  mvars <- liftIO $ for chunks $ \(streamId, events, mvar) -> do+    _ <- forkIO $ do+      stream <- CQRS.getStream streamFamily streamId+      _ <- Exc.runExceptT $ traverse_ (CQRS.writeEvent stream) events+      putMVar mvar ()+    pure mvar++  liftIO $ traverse_ takeMVar mvars++  lastEventIdentifiers <-+    collect $ CQRS.latestEventIdentifiers streamFamily++  length chunks === length lastEventIdentifiers++  for_ lastEventIdentifiers $ \(streamId, eventId) -> do+    stream <- CQRS.getStream streamFamily streamId+    events <- collect $ CQRS.streamEvents stream mempty+    annotateShow events+    length events === fromInteger eventId++writingSendsNotification :: PropertyT IO ()+writingSendsNotification = do+  streamFamily <- CQRS.InMem.emptyStreamFamily @()+  notifStream <- CQRS.allNewEvents streamFamily++  events <- forAll $ Gen.list (Range.linear 1 1000) $+    (,)+      <$> Gen.integral (Range.linear 1 10)+      <*> Gen.double (Range.linearFrac 100 10000)++  _ <- liftIO . forkIO $+    for_ events $ \(streamId, event) -> do+      stream <- CQRS.getStream streamFamily (streamId :: Int)+      void . Exc.runExceptT $ CQRS.writeEvent stream (event :: Double)++  notifications <- collect $+    Pipes.take (length events)+      <-< forever (Pipes.each =<< Pipes.await)+      <-< notifStream++  (\f -> zipWithM_ f events notifications) $+    \(streamId, event) (streamId', ewc) -> do+      let Right CQRS.EventWithContext{ CQRS.event = event' } = ewc+      streamId === streamId'+      event    === event'++gcRemovesListener :: PropertyT IO ()+gcRemovesListener = do+  streamFamily <- CQRS.InMem.emptyStreamFamily @()+  stream <- CQRS.getStream streamFamily ()++  notifStreamMVar <- liftIO newEmptyMVar+  liftIO $ putMVar notifStreamMVar . Just =<< CQRS.allNewEvents streamFamily++  -- Fill the queue.+  replicateM_ 100 . Exc.runExceptT $ CQRS.writeEvent stream ()++  do+    -- Trying to write a new event should fail.+    mvar <- liftIO newEmptyMVar+    threadId <- liftIO . forkIO $ do+      void . Exc.runExceptT $ CQRS.writeEvent stream () -- This should block.+      putMVar mvar () -- This should not happen.+    liftIO $ threadDelay 1000++    failed <- liftIO $ isEmptyMVar mvar+    liftIO $ killThread threadId+    failed === True++  liftIO $ modifyMVar_ notifStreamMVar $ \_ -> pure Nothing+  liftIO performGC++  do+    mvar <- liftIO newEmptyMVar+    threadId <- liftIO . forkIO $ do+      void . Exc.runExceptT $ CQRS.writeEvent stream () -- This should not block.+      putMVar mvar () -- This should happen.+    liftIO $ threadDelay 1000++    failed <- liftIO $ isEmptyMVar mvar+    liftIO $ killThread threadId+    failed === False
+ test/Database/CQRS/TabularDataTest.hs view
@@ -0,0 +1,239 @@+{-# LANGUAGE GADTs #-}+{-# LANGUAGE OverloadedLists #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE PatternSynonyms #-}+{-# LANGUAGE TupleSections #-}+{-# LANGUAGE TypeApplications #-}++module Database.CQRS.TabularDataTest+  ( tests+  ) where++import Control.DeepSeq (NFData)+import Control.Concurrent (ThreadId, forkIO, killThread, threadDelay)+import Control.Monad+import Control.Monad.Trans (MonadIO(..))+import Data.List (nub)+import GHC.Generics (Generic)+import Hedgehog+import Test.Tasty+import Test.Tasty.Hedgehog++import qualified Control.Concurrent.STM     as STM+import qualified Control.Monad.Except       as Exc+import qualified Control.Monad.State.Strict as St+import qualified Data.HashMap.Strict        as HM+import qualified Hedgehog.Gen               as Gen+import qualified Hedgehog.Range             as Range++import Database.CQRS.TabularData (pattern (:~), empty)+import Helpers (interleaveSeqs)++import qualified Database.CQRS             as CQRS+import qualified Database.CQRS.InMemory    as CQRS.InMem+import qualified Database.CQRS.TabularData as CQRS.Tab++tests :: TestTree+tests = testGroup "Database.CQRS.TabularData"+  [ testProperty "Projection using tabular data behaves like aggregator" $+      property tabularDataVsAggregator+  , testProperty "Projection can be restarted" $ property restartingProjection+  ]++tabularDataVsAggregator :: PropertyT IO ()+tabularDataVsAggregator = do+  streamIds <- forAll $+    nub <$> Gen.list (Range.linear 1 10) (Gen.int (Range.linear 1 1000))++  events <- forAll $+    foldM interleaveSeqs [] <=< forM streamIds $ \streamId ->+      Gen.list (Range.linear 1 1000) ((streamId,) <$> testEventGen)++  n <- forAll $ Gen.int (Range.linear 0 (length events))+  let (beforeEvents, afterEvents) = splitAt n events++  streamFamily <- liftIO CQRS.InMem.emptyStreamFamily++  trackingTable <- CQRS.createInMemoryTrackingTable+  table <- liftIO $ STM.newTVarIO []++  liftIO . forM_ beforeEvents $ \(streamId, event) -> do+    stream <- CQRS.getStream streamFamily streamId+    void . Exc.runExceptT $ CQRS.writeEvent stream event++  threadId <- liftIO . forkIO . void .  Exc.runExceptT $ do+    CQRS.runProjection streamFamily id goodProjection trackingTable $+      CQRS.executeInMemoryActions table CQRS.Tab.applyTabularDataAction++  liftIO . forM_ afterEvents $ \(streamId, event) -> do+    stream <- CQRS.getStream streamFamily streamId+    void . Exc.runExceptT $ CQRS.writeEvent stream event++  forM_ streamIds $+    checkConsistency True streamFamily trackingTable table threadId++  liftIO . killThread $ threadId++checkConsistency+  :: Bool+  -> CQRS.InMem.StreamFamily Int () TestEvent+  -> CQRS.InMemoryTrackingTable Int Integer Int+  -> STM.TVar (CQRS.Tab.Table ProjectedCols)+  -> ThreadId+  -> Int+  -> PropertyT IO ()+checkConsistency+    shouldRetry streamFamily trackingTable table threadId streamId = do+  projected <- liftIO $ STM.readTVarIO table++  stream <- CQRS.getStream streamFamily streamId+  eExpected <- Exc.runExceptT $+    CQRS.runAggregator aggregator stream mempty (streamId, NoRow)+  let mResult = HM.lookup (streamId :~ empty) projected++  trackedState <- liftIO $ CQRS.getTrackedState trackingTable streamId+  annotateShow trackedState++  case (mResult, eExpected) of+    (Nothing, Right ((_, NoRow), _, _)) -> pure ()+    ( Just (counter :~ CQRS.Tab.Nil),+      Right ((_, SuccessState counter'), _, _) )+      | counter == counter' -> pure ()+    ( Just (counter :~ CQRS.Tab.Nil),+      Right ((_, FailedState counter'), _, _) )+      | counter == counter' -> pure ()+    _ | shouldRetry -> do+          liftIO . threadDelay $ 10 * 1000 -- Wait 10ms for the projection.+          checkConsistency+            False streamFamily trackingTable table threadId streamId+      | otherwise -> do+          liftIO . killThread $ threadId+          annotateShow mResult+          annotateShow eExpected+          failure++restartingProjection :: PropertyT IO ()+restartingProjection = do+  events <- forAll $ Gen.list (Range.linear 1 100) testEventGen+  n <- forAll $ Gen.int (Range.linear 0 (length events))+  let (beforeEvents, afterEvents) = splitAt n events++  streamFamily <- liftIO CQRS.InMem.emptyStreamFamily++  trackingTable <- CQRS.createInMemoryTrackingTable+  -- Set a dummy row in the table to make the bad projection crash.+  table <- liftIO $ STM.newTVarIO [(0 :~ empty, 0 :~ empty)]++  liftIO . forM_ beforeEvents $ \event -> do+    stream <- CQRS.getStream streamFamily 1+    void . Exc.runExceptT $ CQRS.writeEvent stream event++  threadId <- liftIO . forkIO . void . Exc.runExceptT $+    CQRS.runProjection streamFamily id badProjection trackingTable $+      CQRS.executeInMemoryActions table CQRS.Tab.applyTabularDataAction++  liftIO . forM_ afterEvents $ \event -> do+    stream <- CQRS.getStream streamFamily 1+    void . Exc.runExceptT $ CQRS.writeEvent stream event++  liftIO . killThread $ threadId+  threadId' <- liftIO . forkIO . void . Exc.runExceptT $+    CQRS.runProjection streamFamily id goodProjection trackingTable $+      CQRS.executeInMemoryActions table CQRS.Tab.applyTabularDataAction++  checkConsistency True streamFamily trackingTable table threadId' 1++  liftIO . killThread $ threadId'++data TestEvent+  = CreatedEvent+  | IncrementedEvent+  | ResetEvent+  | DeletedEvent+  deriving (Bounded, Enum, Eq, Generic, NFData, Show)++testEventGen :: Gen TestEvent+testEventGen = Gen.enumBounded++type ProjectedCols =+  'CQRS.Tab.WithUniqueKey+    '[ '("stream_id", Int)]+    '[ '("counter", Int)]++goodProjection+  :: CQRS.Projection+      (CQRS.EventWithContext Integer () TestEvent)+      Int+      (CQRS.Tab.TabularDataAction ProjectedCols)+goodProjection CQRS.EventWithContext{..} = do+  streamId <- St.get++  pure $ case event of+    CreatedEvent ->+      [ CQRS.Tab.Insert $ streamId :~ 0 :~ empty+      ]++    IncrementedEvent ->+      [ CQRS.Tab.Update+          (CQRS.Tab.ffield @ProjectedCols @"counter" (CQRS.Tab.plus 1))+          (CQRS.Tab.field @ProjectedCols @"stream_id" streamId)+      ]++    ResetEvent ->+      [ CQRS.Tab.Upsert $ streamId :~ 0 :~ empty+      ]++    DeletedEvent ->+      [ CQRS.Tab.Delete $+          CQRS.Tab.ffield @ProjectedCols @"stream_id" (CQRS.Tab.equal streamId)+      ]++badProjection+  :: CQRS.Projection+      (CQRS.EventWithContext Integer () TestEvent)+      Int+      (CQRS.Tab.TabularDataAction ProjectedCols)+badProjection CQRS.EventWithContext{..} = do+  streamId <- St.get++  pure $ case event of+    CreatedEvent ->+      [ CQRS.Tab.Insert $ streamId :~ 0 :~ empty+      ]++    IncrementedEvent ->+      [ CQRS.Tab.Insert $ 0 :~ 0 :~ empty -- Make it fail.+      ]++    ResetEvent ->+      [ CQRS.Tab.Upsert $ streamId :~ 0 :~ empty+      ]++    DeletedEvent ->+      [ CQRS.Tab.Delete $+          CQRS.Tab.ffield @ProjectedCols @"stream_id" (CQRS.Tab.equal streamId)+      ]++data ExpectedResult+  = NoRow+  | FailedState Int+  | SuccessState Int+  deriving Show++aggregator+  :: CQRS.Aggregator+      (CQRS.EventWithContext Integer () TestEvent)+      (Int, ExpectedResult)+aggregator CQRS.EventWithContext{..} =+  St.modify' $ \(streamId, current) ->+    (streamId,) $ case (event, current) of+      (_, FailedState _) -> current+      (CreatedEvent, NoRow) -> SuccessState 0+      (CreatedEvent, SuccessState c) -> FailedState c+      (IncrementedEvent, NoRow) -> NoRow+      (IncrementedEvent, SuccessState c) -> SuccessState (c + 1)+      (ResetEvent, _) -> SuccessState 0+      (DeletedEvent, _) -> NoRow
+ test/Database/CQRS/TransformerTest.hs view
@@ -0,0 +1,146 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE KindSignatures #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE TupleSections #-}+{-# LANGUAGE TypeApplications #-}++module Database.CQRS.TransformerTest+  ( tests+  ) where++import Control.DeepSeq (NFData)+import Control.Monad (forM_, unless, void, when)+import Control.Monad.Trans (liftIO)+import Data.Maybe (maybeToList)+import GHC.Generics (Generic)+import Hedgehog hiding (collect)+import Test.Tasty+import Test.Tasty.Hedgehog++import qualified Control.Monad.Except as Exc+import qualified Hedgehog.Gen         as Gen+import qualified Hedgehog.Range       as Range++import Helpers (collect, interleaveSeqs)++import qualified Database.CQRS          as CQRS+import qualified Database.CQRS.InMemory as CQRS.InMem++tests :: TestTree+tests = testGroup "Transformer"+  [ testProperty "Events are dropped and merged accordingly" $+      property eventsDroppedAndMerged+  ]++data TestEvent+  = CreatedEvent String+  | RenamedEvent String+  | SuffixAddedEvent String -- We want to get rid of this one.+  | DeletedEvent+  deriving (Eq, Show, Generic)+  deriving anyclass (NFData)++testEventGen :: Gen TestEvent+testEventGen = do+  name <- Gen.string (Range.linear 0 10) Gen.alphaNum+  Gen.element+    [ RenamedEvent name+    , SuffixAddedEvent name+    ]++testEventSequenceGen :: Gen [TestEvent]+testEventSequenceGen = do+  events <- Gen.list (Range.linear 1 1000) testEventGen+  name <- Gen.string (Range.linear 0 10) Gen.alphaNum+  deletedEvent <- Gen.element [Nothing, Just DeletedEvent]+  pure $ CreatedEvent name : events ++ maybeToList deletedEvent++isCreatedEvent :: CQRS.EventWithContext Integer () TestEvent -> Bool+isCreatedEvent = \case+  CQRS.EventWithContext _ _ (CreatedEvent _) -> True+  _ -> False++isRenamedEvent :: CQRS.EventWithContext Integer () TestEvent -> Bool+isRenamedEvent = \case+  CQRS.EventWithContext _ _ (RenamedEvent _) -> True+  _ -> False++isDeletedEvent :: CQRS.EventWithContext Integer () TestEvent -> Bool+isDeletedEvent = \case+  CQRS.EventWithContext _ _ DeletedEvent -> True+  _ -> False++transformer+  :: CQRS.Transformer+      (Either (Integer, String) (CQRS.EventWithContext Integer () TestEvent))+      Integer (CQRS.EventWithContext Integer () TestEvent)+transformer = \case+  Left (i, err) -> CQRS.failTransformer i err+  Right ewc@(CQRS.EventWithContext _ () event) ->+    case event of+      CreatedEvent _ -> CQRS.pushEvent ewc++      RenamedEvent name -> do+        changed <- CQRS.mergeEvents $ \events ->+          case break isCreatedEvent events of+            (levs, CQRS.EventWithContext eventId _ (CreatedEvent _) : revs) ->+              let ewc' = CQRS.EventWithContext eventId () (CreatedEvent name)+              in (True, levs ++ ewc' : revs)+            _ -> (False, events)+        unless changed $ CQRS.pushEvent ewc++      SuffixAddedEvent suffix -> do+        changed <- CQRS.mergeEvents $ \events ->+          case break (\e -> isCreatedEvent e || isRenamedEvent e) events of+            (levs, CQRS.EventWithContext eventId _ (CreatedEvent n) : revs) ->+              let ewc' =+                    CQRS.EventWithContext eventId ()+                      (CreatedEvent (n ++ suffix))+              in (True, levs ++ ewc' : revs)+            (levs, CQRS.EventWithContext eventId _ (RenamedEvent n) : revs) ->+              let ewc' =+                    CQRS.EventWithContext eventId ()+                      (RenamedEvent (n ++ suffix))+              in (True, levs ++ ewc' : revs)+            _ -> (False, events)+        unless changed $ CQRS.pushEvent ewc++      DeletedEvent ->+        CQRS.mergeEvents $ \events ->+          case break isCreatedEvent events of+            (_, CQRS.EventWithContext _ _ (CreatedEvent _) : _) ->+              ((), [])+            _ -> ((), [ewc])++eventsDroppedAndMerged :: PropertyT IO ()+eventsDroppedAndMerged = do+  inputEvents <- forAll $ do+    seq1 <- map (1,) <$> testEventSequenceGen+    seq2 <- map (2,) <$> testEventSequenceGen+    interleaveSeqs seq1 seq2++  inputStreamFamily <- liftIO CQRS.InMem.emptyStreamFamily+  let transformedStreamFamily =+        CQRS.transformStreamFamily @IO @Int+          transformer pure pure pure pure inputStreamFamily++  liftIO . forM_ inputEvents $ \(streamId, event) -> do+    stream <- CQRS.getStream inputStreamFamily streamId+    void . Exc.runExceptT $ CQRS.writeEvent stream event++  stream1 <- liftIO $ CQRS.getStream transformedStreamFamily 1+  stream2 <- liftIO $ CQRS.getStream transformedStreamFamily 2++  outputEvents1 <- liftIO . collect $ CQRS.streamEvents stream1 mempty+  outputEvents2 <- liftIO . collect $ CQRS.streamEvents stream2 mempty++  forM_ [outputEvents1, outputEvents2] $ \outputEvents -> do+    forM_ outputEvents $ \batch -> do+      when (any (either (const False) isCreatedEvent) batch) $+        length batch === 1+      when (any (either (const False) isDeletedEvent) batch) $+        length batch === 1
+ test/Helpers.hs view
@@ -0,0 +1,38 @@+{-# LANGUAGE FlexibleContexts #-}++module Helpers+  ( collect+  , interleaveSeqs+  ) where++import Control.Monad (forever)+import Control.Monad.Trans (lift)+import Hedgehog hiding (collect)+import Pipes ((>->))++import qualified Control.Monad.Writer as W+import qualified Hedgehog.Gen         as Gen+import qualified Pipes++collect :: Monad m => Pipes.Producer a m () -> m [a]+collect producer =+    fmap snd . W.runWriterT . Pipes.runEffect $+      Pipes.hoist lift producer >-> collector+  where+    collector :: W.MonadWriter [a] m => Pipes.Consumer a m ()+    collector = forever $ do+      x <- Pipes.await+      W.tell [x]++interleaveSeqs :: [a] -> [a] -> Gen [a]+interleaveSeqs = go id+  where+    go :: ([a] -> [a]) -> [a] -> [a] -> Gen [a]+    go acc [] [] = pure $ acc []+    go acc [] ys = pure $ acc ys+    go acc xs [] = pure $ acc xs+    go acc (x:xs) (y:ys) =+      Gen.choice+        [ go (acc . (x:)) xs (y:ys)+        , go (acc . (y:)) (x:xs) ys+        ]
+ test/Main.hs view
@@ -0,0 +1,17 @@+import System.IO+import Test.Tasty++import qualified Database.CQRS.InMemoryTest    as InMem+import qualified Database.CQRS.TabularDataTest as TabularData+import qualified Database.CQRS.TransformerTest as Transformer++main :: IO ()+main = do+  -- `bazel test` doesn't set LANG and co.+  hSetEncoding stdout utf8+  hSetEncoding stderr utf8+  defaultMain $ testGroup "All tests"+    [ InMem.tests+    , TabularData.tests+    , Transformer.tests+    ]