home..
The Playground

Learning by example | github | blog


    import kotlinx.coroutines.flow.*
    import com.fraktalio.fmodel.domain.Decider
    import com.fraktalio.fmodel.application.EventRepository
    import com.fraktalio.fmodel.application.EventSourcingAggregate
    import com.fraktalio.fmodel.application.eventSourcingAggregate
    import com.fraktalio.fmodel.application.StateRepository
    import com.fraktalio.fmodel.application.StateStoredAggregate
    import com.fraktalio.fmodel.application.stateStoredAggregate
    import com.fraktalio.fmodel.application.*

                package com.fraktalio.fmodel.domain
            import kotlinx.coroutines.flow.*

          // ######### Decider #########

            interface I_Decider<in C, in Si, out So, in Ei, out Eo> {
                val decide: (C, Si) -> Flow<Eo>
                val evolve: (Si, Ei) -> So
                val initialState: So
            }

            typealias IDecider<C, S, E> = I_Decider<C, S, S, E, E>
            typealias Decider<C, S, E> = _Decider<C, S, S, E, E>

            data class _Decider<in C, in Si, out So, in Ei, out Eo>(
                override val decide: (C, Si) -> Flow<Eo>,
                override val evolve: (Si, Ei) -> So,
                override val initialState: So
            ) : I_Decider<C, Si, So, Ei, Eo> {

            inline fun <Cn> mapLeftOnCommand(
                crossinline f: (Cn) -> C
            ): _Decider<Cn, Si, So, Ei, Eo> = _Decider(
                decide = { cn, si -> decide(f(cn), si) },
                evolve = { si, ei -> evolve(si, ei) },
                initialState = initialState
            )

            inline fun <Ein, Eon> dimapOnEvent(
                crossinline fl: (Ein) -> Ei, crossinline fr: (Eo) -> Eon
            ): _Decider<C, Si, So, Ein, Eon> = _Decider(
                decide = { c, si -> decide(c, si).map { fr(it) } },
                evolve = { si, ein -> evolve(si, fl(ein)) },
                initialState = initialState
            )

            inline fun <Ein> mapLeftOnEvent(crossinline f: (Ein) -> Ei): _Decider<C, Si, So, Ein, Eo> = dimapOnEvent(f) { it }

            inline fun <Eon> mapOnEvent(crossinline f: (Eo) -> Eon): _Decider<C, Si, So, Ei, Eon> = dimapOnEvent({ it }, f)

            inline fun <Sin, Son> dimapOnState(
                crossinline fl: (Sin) -> Si, crossinline fr: (So) -> Son
            ): _Decider<C, Sin, Son, Ei, Eo> = _Decider(
                decide = { c, sin -> decide(c, fl(sin)) },
                evolve = { sin, ei -> fr(evolve(fl(sin), ei)) },
                initialState = fr(initialState)
            )

            inline fun <Sin> mapLeftOnState(crossinline f: (Sin) -> Si): _Decider<C, Sin, So, Ei, Eo> = dimapOnState(f) { it }

            inline fun <Son> mapOnState(crossinline f: (So) -> Son): _Decider<C, Si, Son, Ei, Eo> = dimapOnState({ it }, f)
            }

            fun <C, Si, So, Ei, Eo, Son> _Decider<C, Si, So, Ei, Eo>.applyOnState(
                ff: _Decider<C, Si, (So) -> Son, Ei, Eo>
            ): _Decider<C, Si, Son, Ei, Eo> = _Decider(
                decide = { c, si -> flowOf(ff.decide(c, si), decide(c, si)).flattenConcat() },
                evolve = { si, ei -> ff.evolve(si, ei)(evolve(si, ei)) },
                initialState = ff.initialState(initialState)
            )

            fun <C, Si, So, Ei, Eo, Son> _Decider<C, Si, So, Ei, Eo>.productOnState(
                fb: _Decider<C, Si, Son, Ei, Eo>
            ): _Decider<C, Si, Pair<So, Son>, Ei, Eo> = applyOnState(fb.mapOnState { b: Son -> { a: So -> Pair(a, b) } })

            inline fun <reified C : C_SUPER, Si, So, reified Ei : Ei_SUPER, Eo : Eo_SUPER, reified C2 : C_SUPER, Si2, So2, reified Ei2 : Ei_SUPER, Eo2 : Eo_SUPER, C_SUPER, Ei_SUPER, Eo_SUPER> _Decider<C?, Si, So, Ei?, Eo>.combine(
                y: _Decider<C2?, Si2, So2, Ei2?, Eo2>
            ): _Decider<C_SUPER, Pair<Si, Si2>, Pair<So, So2>, Ei_SUPER, Eo_SUPER> {

                val deciderX = this.mapLeftOnCommand<C_SUPER> { it as? C }.mapLeftOnState<Pair<Si, Si2>> { pair -> pair.first }
                    .dimapOnEvent<Ei_SUPER, Eo_SUPER>({ it as? Ei }, { it })

                val deciderY = y.mapLeftOnCommand<C_SUPER> { it as? C2 }.mapLeftOnState<Pair<Si, Si2>> { pair -> pair.second }
                    .dimapOnEvent<Ei_SUPER, Eo_SUPER>({ it as? Ei2 }, { it })

                return deciderX.productOnState(deciderY)
            }


           // ######### View #########

            interface I_View<in Si, out So, in E> {
                val evolve: (Si, E) -> So
                val initialState: So
            }

            typealias IView<S, E> = I_View<S, S, E>

            typealias View<S, E> = _View<S, S, E>

            data class _View<in Si, out So, in E>(
                override val evolve: (Si, E) -> So,
                override val initialState: So,
            ) : I_View<Si, So, E> {
                inline fun <En> mapLeftOnEvent(
                    crossinline f: (En) -> E
                ): _View<Si, So, En> = _View(
                    evolve = { si, en -> evolve(si, f(en)) },
                    initialState = initialState
                )

                inline fun <Sin, Son> dimapOnState(
                    crossinline fl: (Sin) -> Si, crossinline fr: (So) -> Son
                ): _View<Sin, Son, E> = _View(
                    evolve = { sin, e -> fr(evolve(fl(sin), e)) },
                    initialState = fr(initialState)
                )

                inline fun <Sin> mapLeftOnState(crossinline f: (Sin) -> Si): _View<Sin, So, E> = dimapOnState(f) { it }

                inline fun <Son> mapOnState(crossinline f: (So) -> Son): _View<Si, Son, E> = dimapOnState({ it }, f)
            }

            fun <Si, So, E, Son> _View<Si, So, E>.applyOnState(
                ff: _View<Si, (So) -> Son, E>
            ): _View<Si, Son, E> = _View(
                evolve = { si, e -> ff.evolve(si, e)(evolve(si, e)) },
                initialState = ff.initialState(initialState)
            )

            fun <Si, So, E, Son> _View<Si, So, E>.productOnState(fb: _View<Si, Son, E>): _View<Si, Pair<So, Son>, E> =
                applyOnState(fb.mapOnState { b: Son -> { a: So -> Pair(a, b) } })

            inline fun <Si, So, reified E : E_SUPER, Si2, So2, reified E2 : E_SUPER, E_SUPER> _View<Si, So, E?>.combine(
                y: _View<Si2, So2, E2?>
            ): _View<Pair<Si, Si2>, Pair<So, So2>, E_SUPER> {

                val viewX = this.mapLeftOnEvent<E_SUPER> { it as? E }.mapLeftOnState<Pair<Si, Si2>> { pair -> pair.first }

                val viewY = y.mapLeftOnEvent<E_SUPER> { it as? E2 }.mapLeftOnState<Pair<Si, Si2>> { pair -> pair.second }

                return viewX.productOnState(viewY)
            }

         // ######### Saga #########

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

            typealias ISaga<AR, A> = I_Saga<AR, A>

            typealias Saga<AR, A> = _Saga<AR, A>

            data class _Saga<in AR, out A>(
                override val react: (AR) -> Flow<A>
            ) : I_Saga<AR, A> {
                inline fun <ARn> mapLeftOnActionResult(crossinline f: (ARn) -> AR): _Saga<ARn, A> =
                    _Saga(react = { arn -> react(f(arn)) })

                inline fun <An> mapOnAction(crossinline f: (A) -> An): _Saga<AR, An> =
                    _Saga(react = { ar -> react(ar).map { f(it) } })
            }

            inline fun <reified AR : AR_SUPER, A : A_SUPER, reified AR2 : AR_SUPER, A2 : A_SUPER, AR_SUPER, A_SUPER> _Saga<AR?, A>.combine(
                y: _Saga<AR2?, A2>
            ): _Saga<AR_SUPER, A_SUPER> {

                val sagaX = this.mapLeftOnActionResult<AR_SUPER> { it as? AR }.mapOnAction<A_SUPER> { it }

                val sagaY = y.mapLeftOnActionResult<AR_SUPER> { it as? AR2 }.mapOnAction<A_SUPER> { it }

                return _Saga(react = { eitherAr -> flowOf(sagaX.react(eitherAr), (sagaY.react(eitherAr))).flattenConcat() })
            }

        

          package com.fraktalio.fmodel.application

      import com.fraktalio.fmodel.domain.IDecider
      import com.fraktalio.fmodel.domain.IView
      import com.fraktalio.fmodel.domain.ISaga
      import kotlinx.coroutines.flow.*


        // ######### Event sourcing aggregate #########

        interface EventRepository<C, E> {
              fun C.fetchEvents(): Flow<E>
              suspend fun E.save(): E
              fun Flow<E>.save(): Flow<E> = map { it.save() }
          }

        interface EventSourcingAggregate<C, S, E> : IDecider<C, S, E>, EventRepository<C, E> {
              fun Flow<E>.computeNewEvents(command: C): Flow<E> = flow {
              val currentState = fold(initialState) { s, e -> evolve(s, e) }
              val resultingEvents = decide(command, currentState)
              emitAll(resultingEvents)
              }
          }

        interface EventSourcingOrchestratingAggregate<C, S, E> : ISaga<E, C>, EventSourcingAggregate<C, S, E> {
            override fun Flow<E>.computeNewEvents(command: C): Flow<E> = flow {
                val currentState = fold(initialState) { s, e -> evolve(s, e) }
                var resultingEvents = decide(command, currentState)

                resultingEvents.flatMapConcat { react(it) }.onEach {
                    val newEvents = flowOf(this@computeNewEvents, resultingEvents).flattenConcat().computeNewEvents(it)
                    resultingEvents = flowOf(resultingEvents, newEvents).flattenConcat()
                }.collect()

                emitAll(resultingEvents)
            }
        }

        fun <C, S, E> EventSourcingAggregate<C, S, E>.handle(command: C): Flow<E> =
          command.fetchEvents().computeNewEvents(command).save()

        fun <C, S, E> EventSourcingAggregate<C, S, E>.handle(commands: Flow<C>): Flow<E> = commands.flatMapConcat { handle(it) }

        fun <C, E> C.publishTo(aggregate: EventSourcingAggregate<C, *, E>): Flow<E> = aggregate.handle(this)
        fun <C, E> Flow<C>.publishTo(aggregate: EventSourcingAggregate<C, *, E>): Flow<E> = aggregate.handle(this)

        fun <C, S, E> eventSourcingAggregate(
            decider: IDecider<C, S, E>,
            eventRepository: EventRepository<C, E>
        ): EventSourcingAggregate<C, S, E> =
            object : EventSourcingAggregate<C, S, E>,
                EventRepository<C, E> by eventRepository,
                IDecider<C, S, E> by decider {}

        fun <C, S, E> eventSourcingOrchestratingAggregate(
            decider: IDecider<C, S, E>,
            eventRepository: EventRepository<C, E>,
            saga: ISaga<E, C>
        ): EventSourcingOrchestratingAggregate<C, S, E> =
            object : EventSourcingOrchestratingAggregate<C, S, E>,
                EventRepository<C, E> by eventRepository,
                IDecider<C, S, E> by decider,
                ISaga<E, C> by saga {}

        // ######### State-stored aggregate #########

        interface StateRepository<C, S> {
            suspend fun C.fetchState(): S?
            suspend fun S.save(): S
        }

        interface StateStoredAggregate<C, S, E> : IDecider<C, S, E>, StateRepository<C, S> {
            suspend fun S?.computeNewState(command: C): S {
                val currentState = this ?: initialState
                val events = decide(command, currentState)
                return events.fold(currentState) { s, e -> evolve(s, e) }
            }
        }

        interface StateStoredOrchestratingAggregate<C, S, E> : ISaga<E, C>, StateStoredAggregate<C, S, E> {
            override suspend fun S?.computeNewState(command: C): S {
                val currentState = this ?: initialState
                val events = decide(command, currentState)
                val newState = events.fold(currentState) { s, e -> evolve(s, e) }
                events.flatMapConcat { react(it) }.onEach { newState.computeNewState(it) }.collect()
                return newState
            }
        }


        suspend fun <C, S, E> StateStoredAggregate<C, S, E>.handle(command: C): S =
            command.fetchState().computeNewState(command).save()

        fun <C, S, E> StateStoredAggregate<C, S, E>.handle(commands: Flow<C>): Flow<S> = commands.map { handle(it) }

        suspend fun <C, S> C.publishTo(aggregate: StateStoredAggregate<C, S, *>): S = aggregate.handle(this)

        fun <C, S> Flow<C>.publishTo(aggregate: StateStoredAggregate<C, S, *>): Flow<S> = aggregate.handle(this)

        fun <C, S, E> stateStoredAggregate(
            decider: IDecider<C, S, E>,
            stateRepository: StateRepository<C, S>
        ): StateStoredAggregate<C, S, E> =
            object : StateStoredAggregate<C, S, E>,
                StateRepository<C, S> by stateRepository,
                IDecider<C, S, E> by decider {}

        fun <C, S, E> stateStoredOrchestratingAggregate(
            decider: IDecider<C, S, E>,
            stateRepository: StateRepository<C, S>,
            saga: ISaga<E, C>
        ): StateStoredOrchestratingAggregate<C, S, E> =
            object : StateStoredOrchestratingAggregate<C, S, E>,
                StateRepository<C, S> by stateRepository,
                IDecider<C, S, E> by decider,
                ISaga<E, C> by saga {}

        // ######### Materialized View #########

        interface ViewStateRepository<E, S> {
            suspend fun E.fetchState(): S?
            suspend fun S.save(): S
        }

        interface MaterializedView<S, E> : IView<S, E>, ViewStateRepository<E, S> {
            fun S?.computeNewState(event: E): S = evolve(this ?: initialState, event)
        }

        suspend fun <S, E> MaterializedView<S, E>.handle(event: E): S = event.fetchState().computeNewState(event).save()

        fun <S, E> MaterializedView<S, E>.handle(events: Flow<E>): Flow<S> = events.map { handle(it) }

        suspend fun <S, E> E.publishTo(materializedView: MaterializedView<S, E>): S = materializedView.handle(this)

        fun <S, E> Flow<E>.publishTo(materializedView: MaterializedView<S, E>): Flow<S> = materializedView.handle(this)


        fun <S, E> materializedView(
            view: IView<S, E>,
            viewStateRepository: ViewStateRepository<E, S>,
        ): MaterializedView<S, E> =
            object : MaterializedView<S, E>,
                ViewStateRepository<E, S> by viewStateRepository,
                IView<S, E> by view {}

        // ######### Saga Manager #########

        interface ActionPublisher<A> {
            suspend fun A.publish(): A
            fun Flow<A>.publish(): Flow<A> = map { it.publish() }
        }

        interface SagaManager<AR, A> : ISaga<AR, A>, ActionPublisher<A> {
            fun AR.computeNewActions(): Flow<A> = react(this)
        }

        fun <AR, A> SagaManager<AR, A>.handle(actionResult: AR): Flow<A> = actionResult.computeNewActions().publish()

        fun <AR, A> SagaManager<AR, A>.handle(actionResults: Flow<AR>): Flow<A> = actionResults.flatMapConcat { handle(it) }

        fun <AR, A> AR.publishTo(sagaManager: SagaManager<AR, A>): Flow<A> = sagaManager.handle(this)

        fun <AR, A> Flow<AR>.publishTo(sagaManager: SagaManager<AR, A>): Flow<A> = sagaManager.handle(this)


        fun <AR, A> sagaManager(
            saga: ISaga<AR, A>,
            actionPublisher: ActionPublisher<A>
        ): SagaManager<AR, A> = object : SagaManager<AR, A>,
            ActionPublisher<A> by actionPublisher,
            ISaga<AR, A> by saga {}


    

    // ################################################################################################
    // ############################################ Domain ############################################
    // ################################################################################################
    // ### - The domain model is fully isolated from the application model and API-related concerns ###
    // ### - It represents a pure declaration of the program logic                                  ###
    // ################################################################################################

    // ############################# The various data of our domain model #############################
    // ################################################################################################

    // ############ Common Values ############
    @JvmInline
    value class Description(val get: String) {
    operator fun plus(param: Description) = Description("${this.get} + ${param.get}")
    operator fun minus(param: Description) = Description("${this.get} - ${param.get}")
    }

    @JvmInline
    value class NumberValue(val get: Int) {
    operator fun plus(param: NumberValue) = NumberValue(this.get + param.get)
    operator fun minus(param: NumberValue) = NumberValue(this.get - param.get)
    }

    // ############ Commands ###############
    abstract class NumberCommand {
        abstract val description: Description
        abstract val value: NumberValue
    }

    sealed class EvenNumberCommand : NumberCommand()

    data class AddEvenNumber(
        override val description: Description,
        override val value: NumberValue
    ) : EvenNumberCommand()

    data class SubtractEvenNumber(
        override val description: Description,
        override val value: NumberValue
    ) : EvenNumberCommand()


    // ############ Events ###############
    abstract class NumberEvent {
        abstract val description: Description
        abstract val value: NumberValue
    }

    sealed class EvenNumberEvent : NumberEvent()

    data class EvenNumberAdded(
        override val description: Description,
        override val value: NumberValue
    ) : EvenNumberEvent()

    data class EvenNumberSubtracted(
        override val description: Description,
        override val value: NumberValue
    ) : EvenNumberEvent()

    // ############ State ###############
    sealed class NumberState {
        abstract val description: Description
        abstract val value: NumberValue
    }

    data class EvenNumberState(
        override val description: Description,
        override val value: NumberValue
    ) : NumberState()

    // ######################### The behaviour of our domain / Program logic ##########################
    // ################################################################################################

    // ############ A Decider ############
    val evenNumberDecider: Decider<EvenNumberCommand?, EvenNumberState, EvenNumberEvent?> =
    Decider(
    initialState = EvenNumberState(Description("Initial state"), NumberValue(0)),
    decide = { c, s ->
    if (c != null && c.value.get > 1000) flow<EvenNumberEvent> { throw UnsupportedOperationException("Sorry") }
    else
    when (c) {
    is AddEvenNumber -> flowOf(EvenNumberAdded(c.description, c.value + s.value))
    is SubtractEvenNumber -> flowOf(EvenNumberSubtracted(c.description, s.value - c.value))
    null -> emptyFlow()
    }
    },
    evolve = { s, e ->
    when (e) {
    is EvenNumberAdded -> EvenNumberState(s.description + e.description, e.value)
    is EvenNumberSubtracted -> EvenNumberState(s.description - e.description, e.value)
    null -> s
    }
    }
    )

    // ################################################################################################
    // ####   The application module orchestrates the execution of the logic by:                   ####
    // ####   - loading current state,                                                             ####
    // ####   - executing domain components and                                                    ####
    // ####   - storing new state                                                                  ####
    // ################################################################################################

    // ################################################################################################
    // ####   Demonstrating two approaches by using single/same Decider (domain model):            ####
    // ####   - Event-sourced systems are storing the Events in immutable storage by only appending ###
    // ####   - State-stored systems are storing the new State by overwriting the previous State   ####
    // ################################################################################################

    // ################################################################################################
    // ######################## 1. Application and Infrastructure (Event-sourced) #####################
    // ################################################################################################

    // ############## A very simple event store/db ##############
    private var evenNumberEventStorage: List<EvenNumberEvent?> = emptyList()

    // ############## Concrete implementation of the EventRepository ##############
    // `EventRepository` interface: https://github.com/fraktalio/fmodel/blob/main/application/src/commonMain/kotlin/com/fraktalio/fmodel/application/EventRepository.kt
    class EvenNumberRepository : EventRepository<EvenNumberCommand?, EvenNumberEvent?> {
    override fun EvenNumberCommand?.fetchEvents(): Flow<EvenNumberEvent?> = evenNumberEventStorage.asFlow()

    override suspend fun EvenNumberEvent?.save(): EvenNumberEvent? {
    evenNumberEventStorage = evenNumberEventStorage.plus(this)
    return this
    }
    }

    // ################################################################################################
    // ######################## 2. Application and Infrastructure (State-stored / Traditional) ########
    // ################################################################################################

    // ############## A very simple state store/db ##############
    private var evenNumberStateStorage: EvenNumberState = EvenNumberState(Description("0"), NumberValue(0))

    // ############## Concrete implementation of the StateRepository ##############
    // `StateRepository` interface: https://github.com/fraktalio/fmodel/blob/main/application/src/commonMain/kotlin/com/fraktalio/fmodel/application/StateRepository.kt
    class EvenNumberStateRepository : StateRepository<EvenNumberCommand?, EvenNumberState> {
    override suspend fun EvenNumberCommand?.fetchState(): EvenNumberState = evenNumberStateStorage

    override suspend fun EvenNumberState.save(): EvenNumberState {
    evenNumberStateStorage = this
    return evenNumberStateStorage
    }
    }

    // ################################################################################################
    // ##################################### Main method/function #####################################
    // ################################################################################################
    suspend fun main(args: Array<String>) {

    // ##########################################
    // ############ 1. Event-sourced ############
    // ##########################################

    val evenNumberESRepository = EvenNumberRepository()

    // An event sourced aggregate
    // `eventSourcingAggregate` factory function: https://github.com/fraktalio/fmodel/blob/main/application/src/commonMain/kotlin/com/fraktalio/fmodel/application/EventSourcingAggregate.kt
    val eventSourcingAggregate = eventSourcingAggregate(
        decider = evenNumberDecider,
        eventRepository = evenNumberESRepository
    )
    // Demonstrating different APIs that are available on EventSourcingAggregate:
    println("### The Event-Sourced approach ###")
    println("new events being stored: ${eventSourcingAggregate.handle(AddEvenNumber(Description("2"), NumberValue(2))).toList()}")
    println("new events being stored: ${eventSourcingAggregate.handle(flowOf(AddEvenNumber(Description("4"), NumberValue(4)), AddEvenNumber(Description("6"), NumberValue(6)))).toList()}")
    println("new events being stored: ${SubtractEvenNumber(Description("8"), NumberValue(8)).publishTo(eventSourcingAggregate).toList()}")
    println("new events being stored: ${flowOf(AddEvenNumber(Description("10"), NumberValue(10)), AddEvenNumber(Description("12"), NumberValue(12))).publishTo(eventSourcingAggregate).toList()}")
    println("final state of the event-storage: ${evenNumberEventStorage}")

    // ##########################################
    // ############ 2. State-stored #############
    // ##########################################

    val evenNumberStateRepository = EvenNumberStateRepository()

    // A state-stored aggregate
    // `stateStoredAggregate` factory function: https://github.com/fraktalio/fmodel/blob/main/application/src/commonMain/kotlin/com/fraktalio/fmodel/application/StateStoredAggregate.kt
    val stateStoredAggregate = stateStoredAggregate(
    decider = evenNumberDecider,
    stateRepository = evenNumberStateRepository
    )
    println("-")
    // Demonstrating different APIs that are available on StateStoredAggregate:
    println("### The State-Stored/Traditional approach ###")
    println("new state being stored: ${stateStoredAggregate.handle(AddEvenNumber(Description("2"), NumberValue(2)))}")
    println("new state being stored: ${stateStoredAggregate.handle(flowOf(AddEvenNumber(Description("4"), NumberValue(4)), AddEvenNumber(Description("6"), NumberValue(6)))).toList()}")
    println("new state being stored: ${SubtractEvenNumber(Description("8"), NumberValue(8)).publishTo(stateStoredAggregate)}")
    println("new state being stored: ${flowOf(AddEvenNumber(Description("10"), NumberValue(10)), AddEvenNumber(Description("12"), NumberValue(12))).publishTo(stateStoredAggregate).toList()}")
    println("final state of the state-storage: ${evenNumberStateStorage}")
    }