Refactoring of two command handler implementations with F# in the context of event sourcing

Question

I've written some F# code to handle commands in the context of event sourcing which works along the lines below:

• Read Events from a specific event store stream
• Build the current state in the given business / domain context accordingly to the events that has just been read right above
• Decide new events based on the current state + read events + a given command
• Write the event that have been newly "decided" to a given stream
• If there is a version mismatch (leading to a failure when it comes to writing the new events to the given stream), there a resilience policy to retry the whole process a certain number of times with a given number of intervals in between (akin to what you can find in Polly).

There is basically two different implementations, a "standard" one and another one validating if some particular events are considered "not okay" slash errors, which makes the handle function returning a Result<'Ok, 'Error>.

Note: The terminology is a bit off but I've found the terms in quite a few places in the "Event Sourcing world".

The code below works, but it's super-duper redundant and it does feel overly bloated, bear in mind that what is below has already been refactored a couple of times, but I'm not too sure about how I can take this code to a whole new level of clarity.

Not only the code is a bit hard to wrap my own head around, it actually also pains me to see that some paths leading to nowhere cannot be prevented upfront in a declarative way, things like:

| None, Ok _
| Some _, Error _ -> NotImplementedException() |> raise

So what I'm asking is some help to refactor the code below into something easier to reason about, and if possible how can I avoid to process cases that aren't supposed to happen by design.

---

CommandHandlerUtils.fs:

[<RequireQualifiedAccess>]
module My.EventSourcing.CommandHandlerUtils

open System

open FSharpPlus.Data

open My.FSharp.Resilience
open My.Infra.EventStore.Common


let internal getInvalidValueException _ _ = NotImplementedException()

let defaultWrongVersionRetryIntervals = Intervals.Seconds [1; 2; 4]

let createStreamKey aggregateName commandMessage =
    { FriendlyName = aggregateName; FriendlyId = commandMessage.AggregateId.ToString() }

/// Build the "history" states based on data available in the "history" events with the reBuild function
/// Note: "history": means "current" here.
let buildHistoryState reBuild (historyEvents: EventRead<'Data, 'Metadata> list) =
    historyEvents
    |> List.map (fun eventRead -> eventRead.Data)
    |> reBuild

/// Read the available events (ascending versions) from a event store and build the current state on top of them.
/// Both values (ie. events and state) are returned.
let buildHistory reBuild (readEvents: ReadEvents<'Data, 'Metadata>) streamKey =
    async {
        let! historyEvents = readEvents streamKey (All Forward)
        let historyState = buildHistoryState reBuild historyEvents
        return (historyEvents, historyState)
    }

/// Save the given events into an event store stream and return a result containing either:
/// - the events successfully written to the event store stream
/// - the error in case the operation has failed
let saveEvents (tryWriteEvents: TryWriteEvents<_, _, _>) historyEvents events streamKey =
    async {
        let eventsToWrite =
            events
            |> List.map (fun event ->
                { Causation = Root
                  Data = event
                  Metadata = () })
            |> NonEmptyList.ofList

        let writeVersion =
            if List.isEmpty historyEvents
            then Expected 1
            else Expected ((List.last historyEvents).StreamVersion + 1)

        return! tryWriteEvents streamKey writeVersion eventsToWrite
    }

---

GenericCommandHandler.fs:

[<RequireQualifiedAccess>]
module My.EventSourcing.GenericCommandHandler

open FSharpPlus.Data

open My.FSharp.Resilience
open My.Infra.EventStore.Common


// 1. Build history state from readEvents
// 2. Decide new events based on command message + history state
// 3. If any events:
// - a. Try Save / Write new events to given stream (ie. command message) into the Event Store
// - b. Return events + state if writing events did work, throw exception otherwise
// 4. If no events: Return events + state
// Note: default intervals => 4 attempts (1st attempt + 3 retries)
let handleWith aggregateName decide build reBuild readEvents tryWriteEvents commandMessage wrongVersionRetryIntervals =

    let handleCore =
        async {
            let streamKey = CommandHandlerUtils.createStreamKey aggregateName commandMessage
            let! (historyEventReads, historyState) = CommandHandlerUtils.buildHistory reBuild readEvents streamKey
            let events = decide commandMessage.Command historyState
            if List.isEmpty events then
                return (None, events, historyState)
            else
                let state = build historyState events
                let! eventWritesResult =
                    CommandHandlerUtils.saveEvents tryWriteEvents historyEventReads events streamKey
                return (Some eventWritesResult, events, state)
        }

    let acceptOutcome last _ (eventWritesResult, events, state) =
        match eventWritesResult with
        | None
        | Some (Ok _) -> Return (events, state)
        | Some (Error (UnexpectedStreamVersion _)) when not last -> TryAgain
        | Some (Error e) -> e |> WriteEventsException |> raise

    Retry.onValue acceptOutcome CommandHandlerUtils.getInvalidValueException wrongVersionRetryIntervals handleCore

let handle aggregateName decide build reBuild readEvents tryWriteEvents commandMessage =
    handleWith
        aggregateName
        decide
        build
        reBuild
        readEvents
        tryWriteEvents
        commandMessage
        CommandHandlerUtils.defaultWrongVersionRetryIntervals

---

ValidatingGenericCommandHandler.fs:

[<RequireQualifiedAccess>]
module My.EventSourcing.ValidatingGenericCommandHandler

open System

open My.FSharp.Resilience
open My.Infra.EventStore.Common


// 1. Build history state from readEvents
// 2. Decide new events based on command message + history state
// 3. If any events:
// - A. Events are wrapped with Ok
//   - a. Try Save / Write new events to given stream (ie. command message) into the Event Store
//   - b. Return events + state if writing events did work, throw exception otherwise
// - B. Return events wrapped with Error
// 4. If no events: Return (no) events + history state wrapped with Ok
// Note: default intervals => 4 attempts (1st attempt + 3 retries)
let handleWith aggregateName decide build reBuild readEvents tryWriteEvents commandMessage wrongVersionRetryIntervals =

    let handleCore =
        async {
            let streamKey = CommandHandlerUtils.createStreamKey aggregateName commandMessage
            let! (historyEventReads, historyState) = CommandHandlerUtils.buildHistory reBuild readEvents streamKey
            let eventsResult = decide commandMessage.Command historyState
            match eventsResult with
            | Error events ->
                return (None, Error events, historyState)
            | Ok events when List.isEmpty events ->
                return (None, Ok events, historyState)
            | Ok events ->
                let state = build historyState events
                let! writeEventsResult =
                    CommandHandlerUtils.saveEvents tryWriteEvents historyEventReads events streamKey
                return (Some writeEventsResult, Ok events, state)
        }

    let acceptOutcome last _ (eventWritesResult, eventsResult, state) =
        match eventWritesResult, eventsResult with
        | None, Error events -> events |> Error |> Return
        | Some (Ok _), Ok events -> (events, state) |> Ok |> Return
        | Some (Error (UnexpectedStreamVersion _)), Ok _ when not last -> TryAgain
        | Some (Error e), Ok _ -> e |> WriteEventsException |> raise
        | None, Ok _
        | Some _, Error _ -> NotImplementedException() |> raise

    Retry.onValue acceptOutcome CommandHandlerUtils.getInvalidValueException wrongVersionRetryIntervals handleCore

let handle aggregateName decide build reBuild readEvents tryWriteEvents message =
    handleWith
        aggregateName
        decide
        build
        reBuild
        readEvents
        tryWriteEvents
        message
        CommandHandlerUtils.defaultWrongVersionRetryIntervals
```

Answer 1

Using types to represent both successful results and error scenarios is generally a very good move, and F# certainly makes it easier to do this than other languages. As I think you've found though, it can get quite complicated as the number of scenarios grows! You're on the right track, but I'm going to suggest one thing you can do to simplify the approach and make it clearer in a couple of places.

As you've identified, there are a couple of branches of code that in theory should not happen:

1. The command is not valid but we still end up writing events to the store.
2. The command is valid and produces events but we do not attempt to write anything to the store.

These "impossible" branches stem from the handleCore function, from which we're returning a value that represents more options than are theoretically possible. To put it another way, the knowledge about what can actually happen inside this function is not being captured in its return type. We can tidy things up a great deal by defining the return value from the command handlers in a more formal way. I'll focus on the validating command handler since it's the more complex of the two:

type CommandHandlerResult =
    /// The command was not valid or could not be applied to the existing state.
    | CommandValidationFailed of error:ValidationError
    /// The command was applied successfully to the existing state, producing
    /// zero or more events and a new state.
    | CommandApplied of newEvents:Event list * newState:State
    /// The command was valid, but there was an error when writing new events to the store.
    | FailedToWriteEvents of error:WriteError

This is obviously fairly verbose, but that's kind of the point, because their intent is now much clearer than the nested combinations of Ok/Error/Some/None, which ultimately gave us more flexibility than we wanted or needed. This already should make some difference to the readability of the handleCore function:

let handleCore =
    async {
        // ...
        match decide commandMessage.Command historyState with
        | Error error -> return CommandValidationFailed error
        // Command was applied, but no change was required, so no need to
        // write any events because there are none!
        | Ok [] -> return CommandApplied ([], historyState)
        | Ok newEvents ->
            // Attempt to write the new events to the store.
            let! writeResult =
                CommandHandlerUtils.saveEvents tryWriteEvents historyEventReads newEvents streamKey
            match writeResult with
            | Error error -> return StateUpdateFailed error
            | Ok _ ->
                let newState = build historyState events
                return CommandApplied (newEvents, newState)
    }

Note that the return value from the decide function is no longer passed through as part of the handleCore result, we've instead transformed it completely into a CommandHandlerResult and have therefore captured the outcome of handleCore much more clearly. The real improvement in readability comes when handling the result of handleCore in the acceptOutcome function:

let acceptOutcome last _ (commandHandlerResult: CommandHandlerResult) =
    match commandHandlerResult with
    | CommandValidationFailed error -> error |> Error |> Return
    | CommandApplied (events, state) -> (events, state) |> Ok |> Return
    | FailedToWriteEvents (UnexpectedStreamVersion _) when not last -> TryAgain
    | FailedToWriteEvents writeError -> writeError |> WriteEventsException |> raise

In my opinion this is now much easier to understand. There are no "impossible" routes and you don't really need to know a great deal about what the command handler is doing because the outcome is encoded in a very descriptive way by its return type.

Generally speaking, defining return types as discriminated unions can be a great way to deal with complex error/failure scenarios where Option<'T> or Result<'T, 'TFailure> are not quite flexible or descriptive enough.