Process Manager

[idugalic]

• A Process Manager orchestrates the execution of a series of actions or commands that involve multiple components or systems.
• It tracks the state of the process, manages transitions between different steps, and ensures consistency and coordination among involved components.
• In event-driven systems, it often reacts to events and emits commands, creating a bridge between the event flow and the command flow.

This discussion is an attempt to formalize a Process Manager type.
The type formalizes a pure algorithm, and we do not care about how and from where state/events are coming from (or saved).

The whole idea of this type comes from Jérémie Chassaing, and you can find more info about it here

Domain

/**
 * [Process] is a datatype that represents the central point of control deciding what Action to execute next ([A]).
 *
 * @param AR Action Result type. It can be anything: Event of your systems, Event of another system, HTTP response...
 * @param A Action type. It can be anything: Command of your system, Command of another system, HTTP request, ...
 * @param S State/View type. Represents the State (View/green sticky) that is always backed by Events.
 * @param Ei Event type. These are the new events that this Process manages/creates
 * @property ingest A pure function/lambda that takes action results of type [AR] and input state of type [S] as parameters, and returns the flow of internal events [Flow]<[Ei]>.
 * @property evolve A pure function/lambda that takes input state of type [S] and input event of type [Ei] as parameters, and returns the output/new state [S].
 * @property pending A pure function/lambda that takes input state of type [S], and returns the flow of actions [Flow]<[A]>.
 * @property initialState A starting point / An initial state of type [S].
 */
interface Process<AR, S, Ei, A> {
    val ingest: (AR, S) -> Flow<Ei>
    val evolve: (S, Ei) -> S
    val pending: (S) -> Flow<A>
    val initialState: S
}

1. AR/ActionResult will trigger your Process (manager) by calling ingest: (AR, S) -> Flow<Ei>. AR can be E1/Event of your system, E2/Event of another system, or HTTP response of some service that you are calling. S represents the current state of your Process (a ToDo list). The result is a list/flow of new event types (Ei) that this process can publish.
2. You will use these events (Ei) to evolve the state of your process: evolve: (S, Ei) -> S. This relation between events and state gives us the opportunity to either store the state of this Process (manager) as a series of events (Ei) / event sourced, or as a plain old state (S) / state stored.
3. You can ask/pool your process to give you the next actions you need to take: pending: (S) -> Flow<A>. These actions (A) can be a Command of your system, a Command of another system, HTTP request ...
4. The initial state is your starting point. Usually, it is NULL, but it can be 0 or anything you find to fit your domain.

This type maintains a symmetric relation between the state and events of a single system, enabling a robust integration pattern by loosely coupling systems. Every S/State of any system is always backed out of its events, sharing nothing!

Infrastructure

/**
 * Infrastructure - Action publisher
 *
 * @param A Action
 */
interface ActionPublisher<A, AR> {
    /**
     * Publish actions
     *
     * @receiver [Flow] of Actions of type [A]
     * @return [Flow] of newly published Action Results of type [AR]
     */
    fun Flow<A>.publish(): Flow<AR>
}

/**
 * Infrastructure - State repository.
 *
 * @param AR Action Result
 * @param S State
 */
interface StateRepository<AR, S> {
    /**
     * Fetch state
     *
     * @receiver Action Result of type [AR]
     * @return the State of type [S] or null
     */
    suspend fun AR?.fetchState(): S?


    /**
     * Save state
     *
     * @receiver State of type [S]
     * @return newly saved State of type [S]
     */
    suspend fun S.save(): S
}

/**
 * Infrastructure - Event repository.
 *
 * @param AR Action Result
 * @param Ei Event
 */
interface EventRepository<AR, Ei> {
    /**
     * Fetch events
     *
     * @receiver Action Result of type [AR]
     * @return the flow of events / [Ei]
     */
    fun AR?.fetchEvents(): Flow<Ei>


    /**
     * Save events
     *
     * @receiver flow of events / [Ei]
     * @return newly saved flow of events / [Ei]
     */
    fun Flow<Ei>.save(): Flow<Ei>
}

Application

Two flavors: State Stored and Event Sourced

The algorithm is the same! The way we store the state (eg. a ToDo list) of the Process Manager is different:

• State-stored process manager will durably store the state(ToDo list) in some table/document
• Event-sourced process manager will durably store a list of events that potentially can build that state/projection (ToDo list)

If your system is event-sourced (you use event sourcing), events are the whole truth! Any state/projection you have in your system can evolve from these events! The event-sourced process manager will ensure that integration components align with this fact, otherwise, you might directly evolve the state of your process manager/integration component with events of another system. This is a very robust way of integrating systems. Events are all you need to back up, and there is no need to back up any projection/state.

data class EventSourcedProcessManager<AR, S, Ei, A>(
    private val process: Process<AR, S, Ei, A>,
    private val actionPublisher: ActionPublisher<A, AR>,
    private val eventRepository: EventRepository<AR, Ei>
) : Process<AR, S, Ei, A> by process, ActionPublisher<A, AR> by actionPublisher,
    EventRepository<AR, Ei> by eventRepository {

    /**
     * Handles the action result of type [AR], reactively.
     * Useful in event-driven scenarios where the events/actionResults are streamed to your Process Manager
     *
     * @param actionResult Action Result represent the outcome of some action you want to handle in some way. It can be an Event from another system, event from your system or HTTP response of a kind.
     */
    suspend fun handle(actionResult: AR) {
        // Fetch current events by the action result
        val events = actionResult.fetchEvents()
        // Construct current state by folding current events
        val state = events.fold(initialState) { s, e -> evolve(s, e) }
        // Produce new events/state by ingesting action result, given the current state. Save it!
        val newEvents = ingest(actionResult, state).save()
        // Construct a new state by folding new events on top of the current state
        val newState = newEvents.fold(state) { s, e -> evolve(s, e) }
        // Check if there are pending action and publish them
        // Additionally, `publish` function can synchronously return new flow of action results that you can react on.
        pending(newState).publish().collect { ar -> handle(ar) }
    }

    /**
     * Pools the current process manager state to check for pending Actions/Tasks
     * Useful for manual/pooling scenarios where you ask process manger if there are pending actions
     */
    suspend fun pool() {
        val events = null.fetchEvents()
        val state = events.fold(initialState) { s, e -> evolve(s, e) }
        pending(state).publish().collect { ar ->
            ingest(ar, state).save()
        }
    }
}

data class StateStoredProcessManager<AR, S, Ei, A>(
    private val process: Process<AR, S, Ei, A>,
    private val actionPublisher: ActionPublisher<A, AR>,
    private val stateRepository: StateRepository<AR, S>
) : Process<AR, S, Ei, A> by process, ActionPublisher<A, AR> by actionPublisher,
    StateRepository<AR, S> by stateRepository {

    /**
     * Handles the action result of type [AR], reactively.
     * Useful in event-driven scenarios where the events/actionResults are streamed to your Process Manager
     *
     * @param actionResult Action Result represent the outcome of some action you want to handle in some way. It can be an Event from another system, event from your system or HTTP response of a kind.
     */
    suspend fun handle(actionResult: AR) {
        // Fetch current state by the action result
        val state = actionResult.fetchState() ?: initialState
        // Produce new events by ingesting action result, given the current state.
        val newEvents = ingest(actionResult, state)
        // Construct a new state by folding new events on top of the current state. Save the state!
        val newState = newEvents.fold(state, evolve).save()
        // Check if there are pending action and publish them
        // Additionally, `publish` function can synchronously return new flow of action results that you can react on.
        pending(newState).publish().collect { ar -> handle(ar) }
    }

    /**
     * Pools the current process manager state to check for pending Actions/Tasks
     * Useful for manual/pooling scenarios where you ask process manger if there are pending actions
     */
    suspend fun pool() {
        val state = (null.fetchState() ?: initialState)
        pending(state).publish().collect { ar ->
            val events = ingest(ar, state)
            events.fold(state, evolve).save()
        }
    }
}

[idugalic]

Process = Decider + Saga

By looking closely at the signature of the Process component we can see the resemblance to a Decider component, where ingest=decide and only thing that is missing is pending in decider:

interface Process<AR, S, Ei, A> {
    val ingest: (AR, S) -> Flow<Ei>
    val evolve: (S, Ei) -> S
    val pending: (S) -> Flow<A>
    val initialState: S
}

interface Decider<in C, S, E> {
    val decide: (C, S) -> Flow<E>
    val evolve: (S, E) -> S
    val initialState: S
}

interface Saga<in AR, out A> {
    val react: (AR) -> Flow<A>
}

The react/Saga plays an important role: We can compose react and decide to achieve ingest. AR is (contra)mapped to A/Command by react function and then given to the decide` to make new decisions/Events.

You do not have a pending, but the question is if there is a need for it. You already have very specific ToDo events in place that represent that list! So, you can use the same react function to map Event/AR of your system to Commands/A of another.

[idugalic]

Do we need to encode the Process type in FModel?

[idugalic]

Pay attention to evolve: (S, E) -> S

How you evolve the state of your system out of events is extremely important! It becomes more important in the context of the integration of two systems. In this case, you want to differentiate event and state types: E1, E2, S1, S2.

A naive integration component would do something like this:
(S1, E1) -> S1 and (S1, E2) -> S1
Don't do it! Remember, this S1 is used in decide function to make new decisions/events decide: (C1, S1) -> Flow<E1>. These decisions/events are now influenced by foreign events/E2 (S1, E2) -> S1. It leaks! It couples!

[idugalic]

Simple is hard!

Keeping the number of functions to 3 was hard!

1. decide: (C, S) -> Flow<E>
2. evolve: (S, E) -> S
3. react: (AR) -> Flow<A>

Keeping the number of types to 3 was hard!

1. C=Command
2. E=Event
3. S=State

Wrapping these functions in 3 robust components (that will not leak or couple) was hard!

1. Decide
2. View
3. Saga/React

Enabling switching from state-stored to event-sourced systems, or the other way around, enabling adoption, and minimizing the risk was hard!

1. EventSourced system
2. StateStored system
3. Both