foyer-dsi-assure-2035/core/src/lu/foyer/EventSourcing.scala

package lu.foyer

import zio.*
import java.util.UUID
import zio.schema.Schema

final case class Entity[T](entityId: UUID, data: T, version: Long)
final case class Event[T](entityId: UUID, data: T, eventId: UUID)

trait StateRepository[Data] extends Repository[Entity[Data], UUID]
trait EventRepository[Data] extends Repository[Event[Data], UUID]:
  def fetchOne(entityId: UUID, eventId: UUID): Task[Option[Event[Data]]]
  def fetchMany(entityId: UUID, page: Page): Task[Paged[Event[Data]]]

trait Reducer[Event, State]:
  def fromEmpty: PartialFunction[Event, State]
  def fromState: PartialFunction[(State, Event), State]

  def reduce(event: Event): Option[State] =
    fromEmpty.lift(event)

  def reduce(state: State, event: Event): Option[State] =
    fromState.lift((state, event))

trait CommandHandler[+Command, +Event, +State]:
  def name: String
  def isCreate: Boolean
  inline def isUpdate: Boolean = !isCreate
  def commandSchema: Schema[?]

trait CommandHandlerCreate[Command: Schema, Event] extends CommandHandler[Command, Event, Nothing]:
  def onCommand(entityId: UUID, command: Command): Task[Event]
  val isCreate      = true
  val commandSchema = summon[Schema[Command]]

trait CommandHandlerUpdate[Command: Schema, Event, State]
    extends CommandHandler[Command, Event, State]:
  def onCommand(entityId: UUID, state: State, command: Command): Task[Event]
  val isCreate      = false
  val commandSchema = summon[Schema[Command]]

class CommandEngine[Command, Event, State](
  val handlers: List[CommandHandler[Command, Event, State]],
  reducer: Reducer[Event, State],
  val eventRepo: EventRepository[Event],
  val stateRepo: StateRepository[State]):

  def handleCommand(command: Command, name: String, entityId: UUID)
    : Task[(lu.foyer.Event[Event], lu.foyer.Entity[State])] =
    for
      handler <- ZIO
                   .succeed(handlers.find(_.name == name))
                   .someOrFail(new IllegalArgumentException(s"No handler found for command $name"))
      entityOption            <- stateRepo.fetchOne(entityId)
      (event, newStateOption) <- transition(command, name, entityId, entityOption, handler)
      newState <-
        ZIO
          .succeed(newStateOption)
          .someOrFail(new IllegalArgumentException("Reducer cannot resolve state transition"))
      newEntity = Entity(entityId, newState, entityOption.map(_.version).getOrElse(1))
      _ <- if entityOption.isEmpty then stateRepo.insert(newEntity.entityId, newEntity)
           else stateRepo.update(newEntity.entityId, newEntity)
      eventEntity <- Random.nextUUID.map(Event(newEntity.entityId, event, _))
      _           <- eventRepo.insert(eventEntity.eventId, eventEntity)
    yield (eventEntity, newEntity)

  private def transition(
    command: Command,
    name: String,
    entityId: UUID,
    entityOption: Option[Entity[State]],
    handler: CommandHandler[Command, Event, State]
  ): Task[(Event, Option[State])] = (entityOption, handler) match
    case (None, h) if !h.isUpdate =>
      h.asInstanceOf[CommandHandlerCreate[Command, Event]]
        .onCommand(entityId, command)
        .map(event => (event, reducer.reduce(event)))
    case (Some(entity), h) if h.isUpdate =>
      h.asInstanceOf[CommandHandlerUpdate[Command, Event, State]]
        .onCommand(entityId, entity.data, command)
        .map(event => (event, reducer.reduce(entity.data, event)))
    case (Some(_), h) if !h.isUpdate =>
      ZIO.fail(
        new IllegalArgumentException(s"State already exists when applying create command $name")
      )
    case (None, h) if h.isUpdate =>
      ZIO.fail(new IllegalArgumentException(s"No state found to apply the update command $name"))
    case _ => ZIO.fail(new IllegalArgumentException("Impossible state"))
end CommandEngine

object CommandEngine:
  def layer[Command: Tag, Event: Tag, State: Tag] =
    ZLayer.fromFunction(CommandEngine[Command, Event, State](_, _, _, _))