Bank

The most complete example: a full CQRS banking application. It has multiple aggregates (Account and Customer), a process manager that coordinates money transfers with compensation, a command dispatcher, event publishing, and two styles of read model — one in memory, one persisted to SQLite. It’s also the only example with a test suite.

It brings together Process Managers and Design & Internals.

Composing aggregates

Each aggregate (Account, Customer) defines its own events and commands. The application works with combined sum types, BankEvent and BankCommand, generated by Template Haskell — and embeddings connect the two levels:

constructSumType "BankEvent" (withTagOptions (ConstructTagName (++ "Event")) defaultSumTypeOptions) $
  accountEvents ++ customerEvents

mkSumTypeEmbedding "accountEventEmbedding" ''AccountEvent ''BankEvent
mkSumTypeEmbedding "accountCommandEmbedding" ''AccountCommand ''BankCommand

accountBankProjection :: Projection Account BankEvent
accountBankProjection = embeddedProjection accountEventEmbedding accountProjection

accountBankCommandHandler :: CommandHandler Account BankEvent BankCommand AccountCommandError
accountBankCommandHandler =
  embeddedCommandHandler accountEventEmbedding accountCommandEmbedding accountCommandHandler

embeddedCommandHandler lifts an aggregate handler to the application-wide types; events and commands that don’t belong to it are simply ignored. That’s what makes composition safe — each handler only sees what it understands.

The account handler itself is an ordinary pure decide with validation:

handleAccountCommand account (TransferToAccountAccountCommand cmd)
  | isNothing account.owner = Left AccountNotOpen
  | accountAvailableBalance account - cmd.amount < 0 =
      Left (InsufficientFunds (accountAvailableBalance account))
  | otherwise =
      Right [AccountTransferStartedAccountEvent (AccountTransferStarted cmd.transferId cmd.amount cmd.targetAccount)]

The transfer process manager

A transfer debits one account and credits another — two aggregates. The process manager reacts to AccountTransferStarted by issuing AcceptTransfer to the target, with compensation that fires RejectTransfer on the source if the credit is rejected:

reactTransfer manager (StreamEvent sourceAcct _ _ (AccountTransferStartedEvent evt))
  | isNothing (Map.lookup evt.transferId manager.transferData) =
      [ IssueCommandWithCompensation
          evt.targetAccount
          (AcceptTransferCommand (AcceptTransfer evt.transferId sourceAcct evt.amount))
          id
          ( \(RejectionReason reason) ->
              [ IssueCommand sourceAcct (RejectTransferCommand (RejectTransfer evt.transferId (T.unpack reason))) id ]
          )
      ]
  | otherwise = []

transferProcessManager :: TransferProcessManager
transferProcessManager = ProcessManager transferManagerProjection reactTransfer

The entire saga, including the failure branch, is a pure value.

Two read models

A persisted read model tracks transfers in a SQLite table, with a checkpoint so it resumes after a restart:

transferReadModel :: ReadModel (SqlPersistT IO) BankEvent
transferReadModel =
  ReadModel
    { initialize = void (runMigrationSilent migrateTransfer),
      eventHandler = EventHandler handleTransferEvent,
      checkpointStore = sqliteCheckpointStore (CheckpointName "transfers"),
      reset = deleteWhere ([] :: [Filter TransferEntity])
    }

getTransfersByStatus :: (MonadIO m) => Text -> SqlPersistT m [Entity TransferEntity]
getTransfersByStatus s = selectList [TransferEntityStatus ==. s] []

An in-memory read model — customer→accounts — is just a Projection over the global stream, reusing the account projection and the embedding to fold only the events it cares about:

customerAccountsProjection :: Projection CustomerAccounts (VersionedStreamEvent BankEvent)
customerAccountsProjection = Projection (CustomerAccounts Map.empty Map.empty Map.empty) handleCustomerAccountsEvent

Wiring the write side

Bank.CLI.Store snaps the pieces together. The writer enriches metadata, persists to SQLite, and — via publishingEventStoreWriter — publishes each write synchronously to a handler that both logs the event and drives the process manager through the command dispatcher:

cliEventStoreWriter :: (MonadIO m) => VersionedEventStoreWriter (SqlPersistT m) BankEvent
cliEventStoreWriter =
  publishingEventStoreWriter enrichedWriter (synchronousPublisher eventHandler')
  where
    taggedStore = sqliteTaggedEventStoreWriter defaultSqlEventStoreConfig
    enrichedWriter = metadataEnrichingEventStoreWriter jsonStringCodec taggedStore
    dispatcher =
      commandHandlerDispatcher
        jsonStringCodec
        taggedStore
        cliEventStoreReader
        [mkAggregateHandlerWith formatAccountError accountBankCommandHandler]
    eventHandler' =
      EventHandler (\event -> liftIO $ printJSONPretty (event.key, event.payload))
        <> processManagerEventHandler transferProcessManager cliGlobalEventStoreReader dispatcher

commandHandlerDispatcher routes each command to the aggregate that understands it; <> on EventHandler fans a single write out to both the logger and the process manager.

Testing the domain

Because decide is pure, the domain is tested with no database — just states, commands, and assertions:

let stateAfterStarted = latestProjection accountProjection events
accountAvailableBalance stateAfterStarted `shouldBe` 4
accountCommandHandler.decide stateAfterStarted (DebitAccountAccountCommand (DebitAccount 9 "rent"))
  `shouldBe` Left (InsufficientFunds 4)

Running it

cabal run bank -- --help    # explore the CLI
cabal test examples-bank    # run the test suite

Where to go next

Read the Design & Internals guide to see why the store, publishing, and composition machinery behind this example are built the way they are.