refactoring: Masive refactor to build the V2

docker/parts/docker-compose-traefik.yaml

@@ -1,6 +1,6 @@
 services:
   traefik:
-    image: traefik:3.3.4
+    image: traefik:3.7.9
     command:
       - "--api.insecure=true"
       - "--api.dashboard=true"

src/main/kotlin/eventDemo/contexts/auth/application/eventStores/UserEventStoreRepository.kt

@@ -19,6 +19,6 @@ class UserEventStoreRepository(
   }
 
   override fun save(user: User) {
-    eventStore.publish(user.recordedEvents)
+    eventStore.append(user.recordedEvents)
   }
 }

src/main/kotlin/eventDemo/contexts/auth/domain/User.kt

@@ -11,9 +11,8 @@ data class User(
   val username: String,
   val password: String,
   val version: Int,
+  val recordedEvents: Set<UserEvent>,
 ) {
-  val recordedEvents: Set<UserEvent> = emptySet()
-
   companion object {
     fun createNewUser(
       username: String,
@@ -27,6 +26,7 @@ data class User(
         username = event.username,
         password = event.password,
         version = event.version,
+        recordedEvents = setOf(event),
       )
 
     fun loadFromHistory(events: Set<UserEvent>): User? =

src/main/kotlin/eventDemo/contexts/auth/domain/events/NewUserCreatedEvent.kt

@@ -4,7 +4,9 @@ import eventDemo.libs.eventSource.EventId
 import eventDemo.sharedKernel.UserId
 import kotlinx.datetime.Clock
 import kotlinx.datetime.Instant
+import kotlinx.serialization.Serializable
 
+@Serializable
 class NewUserCreatedEvent(
   val username: String,
   val password: String,

src/main/kotlin/eventDemo/contexts/auth/domain/events/UserEvent.kt

@@ -2,5 +2,7 @@ package eventDemo.contexts.auth.domain.events
 
 import eventDemo.libs.eventSource.Event
 import eventDemo.sharedKernel.UserId
+import kotlinx.serialization.Serializable
 
+@Serializable
 sealed interface UserEvent : Event<UserId>

src/main/kotlin/eventDemo/contexts/auth/infrastructure/configure/ConfigureAuthDI.kt

@@ -5,7 +5,7 @@ import eventDemo.contexts.auth.application.eventStores.UserRepository
 import eventDemo.contexts.auth.application.ports.UserEventStore
 import eventDemo.contexts.auth.application.ports.UserProjectionRepository
 import eventDemo.contexts.auth.infrastructure.persistence.eventStore.UserEventStoreInPostgresql
-import eventDemo.contexts.auth.infrastructure.persistence.projection.UsedProjectionRepositoryInPostgresql
+import eventDemo.contexts.auth.infrastructure.persistence.projection.UserProjectionRepositoryInPostgresql
 import org.koin.core.module.Module
 import org.koin.core.module.dsl.singleOf
 import org.koin.dsl.bind
@@ -13,5 +13,5 @@ import org.koin.dsl.bind
 fun Module.configureAuthDi() {
   singleOf(::UserEventStoreRepository) bind UserRepository::class
   singleOf(::UserEventStoreInPostgresql) bind UserEventStore::class
-  singleOf(::UsedProjectionRepositoryInPostgresql) bind UserProjectionRepository::class
+  singleOf(::UserProjectionRepositoryInPostgresql) bind UserProjectionRepository::class
 }

src/main/kotlin/eventDemo/contexts/auth/infrastructure/configure/ConfigureAuthRoutes.kt

@@ -2,7 +2,6 @@ package eventDemo.contexts.auth.infrastructure.configure
 
 import eventDemo.configuration.configuration
 import eventDemo.contexts.auth.application.eventStores.UserEventStoreRepository
-import eventDemo.contexts.auth.application.eventStores.UserRepository
 import eventDemo.contexts.auth.application.ports.UserProjectionRepository
 import eventDemo.contexts.auth.infrastructure.rest.createUserRoute
 import eventDemo.contexts.auth.infrastructure.rest.loginRoute

src/main/kotlin/eventDemo/contexts/auth/infrastructure/configure/ConfigureKtorAuth.kt

@@ -12,11 +12,7 @@ import io.ktor.server.auth.authentication
 import io.ktor.server.auth.jwt.JWTPrincipal
 import io.ktor.server.auth.jwt.jwt
 import io.ktor.server.response.respond
-import io.ktor.server.routing.post
-import io.ktor.server.routing.routing
-import kotlinx.serialization.json.Json
 import java.util.Date
-import java.util.UUID
 
 fun Application.configureKtorAuth() {
   val jwtSecret = environment.config.configuration.jwtSecret

src/main/kotlin/eventDemo/contexts/auth/infrastructure/persistence/eventStore/UserEventStoreInPostgresql.kt

@@ -18,5 +18,5 @@ class UserEventStoreInPostgresql(
     dataSource,
     { Json.encodeToString(it) },
     { Json.decodeFromString(it) },
-    "auth.user_event_store",
+    "auth.user_event_stream",
   )

src/main/kotlin/eventDemo/contexts/auth/infrastructure/persistence/projection/UserProjectionRepositoryInPostgresql.kt

@@ -7,7 +7,7 @@ import eventDemo.sharedKernel.UserId
 import java.util.UUID
 import javax.sql.DataSource
 
-class UsedProjectionRepositoryInPostgresql(
+class UserProjectionRepositoryInPostgresql(
   val dataSource: DataSource,
 ) : UserProjectionRepository {
   override fun getByUsername(username: String): UserProjection? =

src/main/kotlin/eventDemo/contexts/auth/infrastructure/rest/UserCreateRoute.kt

@@ -1,7 +1,5 @@
 package eventDemo.contexts.auth.infrastructure.rest
 
-import com.password4j.Hash
-import com.password4j.Password
 import eventDemo.contexts.auth.application.eventStores.UserRepository
 import eventDemo.contexts.auth.domain.User
 import eventDemo.contexts.auth.infrastructure.hashPassword

src/main/kotlin/eventDemo/contexts/game/application/command/handlers/GameEventManager.kt

@@ -6,6 +6,8 @@ import eventDemo.contexts.game.application.ports.GameEventBus
 import eventDemo.contexts.game.domain.events.GameEvent
 import eventDemo.contexts.game.domain.game.gameState.Game
 import eventDemo.libs.command.Command
+import eventDemo.libs.eventSource.eventStore.VersionConflictException
+import io.github.oshai.kotlinlogging.KotlinLogging
 import kotlin.reflect.KClass
 
 sealed interface CommandHandler<C : Command> {
@@ -16,12 +18,15 @@ abstract class GameEventManager(
   private val gameRepository: GameRepository,
   private val gameEventBus: GameEventBus,
 ) {
+  private val logger = KotlinLogging.logger {}
+
   fun GameCommand.getGame(): Game =
     gameRepository.get(payload.aggregateId) ?: error("Game not found")
 
   fun GameEvent.getGame(): Game =
     gameRepository.get(aggregateId) ?: error("Game not found")
 
+  @Throws(VersionConflictException::class)
   protected fun Game.saveEvents(): Game {
     gameRepository.save(this)
     return this
@@ -42,4 +47,20 @@ abstract class GameEventManager(
       throw CommandException(message)
     }
   }
+
+  protected fun <T> retry(
+    mapAttempts: Int = 5,
+    block: () -> T,
+  ): T =
+    try {
+      block()
+    } catch (e: VersionConflictException) {
+      if (mapAttempts > 0) {
+        logger.warn { "retry after version conflict (attempts left: $mapAttempts)" }
+        retry(mapAttempts - 1, block)
+      } else {
+        logger.error { "Version conflict retry failed" }
+        throw e
+      }
+    }
 }

src/main/kotlin/eventDemo/contexts/game/application/command/handlers/JoinTheGameHandler.kt

@@ -1,5 +1,6 @@
 package eventDemo.contexts.game.application.command.handlers
 
+import eventDemo.contexts.auth.application.eventStores.UserRepository
 import eventDemo.contexts.game.application.command.models.JoinTheGameCommand
 import eventDemo.contexts.game.application.eventStores.GameRepository
 import eventDemo.contexts.game.application.ports.GameEventBus
@@ -11,16 +12,20 @@ import eventDemo.contexts.game.domain.game.gameState.GameCreated
 class JoinTheGameHandler(
   gameRepository: GameRepository,
   gameEventBus: GameEventBus,
+  private val userRepository: UserRepository,
 ) : GameEventManager(gameRepository, gameEventBus),
   CommandHandler<JoinTheGameCommand> {
   override fun handle(command: JoinTheGameCommand) {
-    command
+    val user = userRepository.get(command.userId) ?: error("User with id ${command.userId} doesn't exist")
-      .getGame()
+    retry {
-      .isStatusOrFail(GameCreated::class, "The game is started")
+      command
-      .userJoinTheGame(
+        .getGame()
-        userId = command.userId,
+        .isStatusOrFail(GameCreated::class, "The game is started")
-        name = "Name${CharArray(6) { ('A'..'Z').random() }.concatToString()}",
+        .userJoinTheGame(
-      ).saveEvents()
+          userId = command.userId,
-      .publishEvents()
+          name = user.username,
+        ).saveEvents()
+        .publishEvents()
+    }
   }
 }

src/main/kotlin/eventDemo/contexts/game/application/eventStores/GameEventStoreRepository.kt

@@ -3,6 +3,7 @@ package eventDemo.contexts.game.application.eventStores
 import eventDemo.contexts.game.application.ports.GameEventStore
 import eventDemo.contexts.game.domain.game.GameId
 import eventDemo.contexts.game.domain.game.gameState.Game
+import eventDemo.libs.eventSource.eventStore.VersionConflictException
 
 class GameEventStoreRepository(
   val eventStore: GameEventStore,
@@ -18,7 +19,8 @@ class GameEventStoreRepository(
     return events.let { Game.loadFromHistory(it) }
   }
 
+  @Throws(VersionConflictException::class)
   override fun save(game: Game) {
-    eventStore.publish(game.recordedEvents)
+    eventStore.append(game.recordedEvents)
   }
 }

src/main/kotlin/eventDemo/contexts/game/application/eventStores/GameRepository.kt

@@ -4,10 +4,12 @@ import eventDemo.contexts.game.domain.game.GameId
 import eventDemo.contexts.game.domain.game.gameState.Game
 import eventDemo.contexts.game.domain.game.gameState.GameCreated
 import eventDemo.contexts.game.domain.game.gameState.GameInit
+import eventDemo.libs.eventSource.eventStore.VersionConflictException
 
 interface GameRepository {
   fun get(id: GameId): Game?
 
+  @Throws(VersionConflictException::class)
   fun save(game: Game)
 
   fun getOrCreate(gameId: GameId): Game =

src/main/kotlin/eventDemo/contexts/game/application/ports/GameListRepository.kt

@@ -3,5 +3,12 @@ package eventDemo.domain.event.projection
 import eventDemo.contexts.game.infrastructure.persistence.projections.models.GameList
 
 interface GameListRepository {
-  fun getList(): List<GameList>
+  fun getList(
+    limit: Int = 100,
+    offset: Int = 0,
+  ): List<GameList>
+
+  fun save(gameList: GameList)
+
+  fun subscribeToBus()
 }

src/main/kotlin/eventDemo/contexts/game/application/projections/GameListBuilder.kt

@@ -11,7 +11,7 @@ import eventDemo.contexts.game.domain.events.PlayerReadyEvent
 import eventDemo.contexts.game.domain.events.PlayerWinEvent
 import eventDemo.contexts.game.infrastructure.persistence.projections.models.GameList
 
-fun GameList.apply(event: GameEvent): GameList =
+fun GameList.applyEvent(event: GameEvent): GameList =
   when (event) {
     is GameCreatedEvent -> {
       this
@@ -32,7 +32,7 @@ fun GameList.apply(event: GameEvent): GameList =
 
     is PlayerWinEvent -> {
       copy(
-        winners = winners + players.filter { it.id == event.playerId },
+        winners = winners,
         status = GameList.Status.FINISH,
       )
     }
@@ -52,6 +52,4 @@ fun GameList.apply(event: GameEvent): GameList =
     is DrawFilledWithDiscardEvent -> {
       this
     }
-  }.copy(
+  }
-    lastEventVersion = event.version,
-  )

src/main/kotlin/eventDemo/contexts/game/domain/game/gameState/GameCreated.kt

@@ -15,7 +15,6 @@ import eventDemo.contexts.game.domain.game.errors.DeckMissingCardsException
 import eventDemo.contexts.game.domain.game.errors.NeedMorePlayersToStartGameException
 import eventDemo.contexts.game.domain.game.errors.ThePlayerIsNotInTheGameException
 import eventDemo.sharedKernel.UserId
-import kotlin.collections.plus
 
 data class GameCreated(
   override val aggregateId: GameId,

src/main/kotlin/eventDemo/contexts/game/infrastructure/configuration/injections/infrastructure/ConfigureDIInfrastructure.kt

@@ -8,7 +8,7 @@ import eventDemo.contexts.game.application.ports.GameEventStore
 import eventDemo.contexts.game.application.ports.GameProjectionBus
 import eventDemo.contexts.game.infrastructure.persistence.eventBus.GameEventBusInRabbinMQ
 import eventDemo.contexts.game.infrastructure.persistence.eventStore.GameEventStoreInPostgresql
-import eventDemo.contexts.game.infrastructure.persistence.projections.GameListRepositoryInRedis
+import eventDemo.contexts.game.infrastructure.persistence.projections.GameListRepositoryInMemory
 import eventDemo.contexts.game.infrastructure.persistence.projections.bus.GameProjectionBusInRabbitMQ
 import eventDemo.domain.event.projection.GameListRepository
 import org.koin.core.module.Module
@@ -22,5 +22,5 @@ fun Module.configureGameDIInfrastructure() {
   singleOf(::CommandSubscriber)
   singleOf(::GameChannelsSubscriber)
   singleOf(::GameCommandHandlerDispatcher)
-  singleOf(::GameListRepositoryInRedis) bind GameListRepository::class
+  singleOf(::GameListRepositoryInMemory) bind GameListRepository::class
 }

src/main/kotlin/eventDemo/contexts/game/infrastructure/configuration/listener/ConfigureGameListener.kt

@@ -1,9 +1,9 @@
 package eventDemo.contexts.game.infrastructure.configuration.listener
 
-import eventDemo.contexts.game.infrastructure.persistence.projections.GameListRepositoryInRedis
+import eventDemo.domain.event.projection.GameListRepository
 import org.koin.core.Koin
 
 fun Koin.configureProjectionListener() {
-  get<GameListRepositoryInRedis>()
+  get<GameListRepository>()
     .subscribeToBus()
 }

src/main/kotlin/eventDemo/contexts/game/infrastructure/persistence/projections/GameListRepositoryInMemory.kt

@@ -1,43 +1,49 @@
 package eventDemo.contexts.game.infrastructure.persistence.projections
 
 import eventDemo.contexts.game.application.ports.GameEventBus
+import eventDemo.contexts.game.application.ports.GameEventStore
 import eventDemo.contexts.game.application.ports.GameProjectionBus
-import eventDemo.contexts.game.application.projections.apply
+import eventDemo.contexts.game.application.projections.applyEvent
 import eventDemo.contexts.game.domain.game.GameId
 import eventDemo.contexts.game.infrastructure.persistence.projections.models.GameList
 import eventDemo.domain.event.projection.GameListRepository
-import eventDemo.libs.eventSource.projection.ProjectionRepositoryInMemory
 import io.github.oshai.kotlinlogging.withLoggingContext
 
 /**
  * Manages [projections][GameList], their building and publication in the [bus][GameProjectionBus].
  */
-class GameListRepositoryInMemory : GameListRepository {
+class GameListRepositoryInMemory(
-  private val projectionsRepository =
+  val gameEventStore: GameEventStore,
-    ProjectionRepositoryInMemory(
+  val projectionBus: GameProjectionBus,
-      applyToProjection = GameList::apply,
+  val eventBus: GameEventBus,
-      initialStateBuilder = { aggregateId: GameId -> GameList(aggregateId) },
+) : GameListRepository {
-    )
+  val projections: MutableMap<GameId, GameList> = mutableMapOf()
 
-  fun subscribeToBus(
+  override fun getList(
-    projectionBus: GameProjectionBus,
+    limit: Int,
-    eventBus: GameEventBus,
+    offset: Int,
-  ) {
+  ): List<GameList> =
+    projections
+      .values
+      .drop(offset)
+      .take(limit)
+
+  override fun save(gameList: GameList) {
+    projections[gameList.aggregateId] = gameList
+  }
+
+  override fun subscribeToBus() {
     // On new event was received, build projection and publish it to the projection bus
     eventBus.subscribe { event ->
       withLoggingContext("event" to event.toString()) {
-        projectionsRepository
+        gameEventStore
-          .applyAndSave(event)
+          .getStream(event.aggregateId)
+          .readAll()
+          .fold(GameList(event.aggregateId)) { acc, event ->
+            acc.applyEvent(event)
+          }.also { save(it) }
           .also { projectionBus.publish(it) }
       }
     }
   }
-
-  /**
-   * Get the last version of the [GameState] from the all eventStream.
-   *
-   * It fetches it from the local cache if possible, otherwise it builds it.
-   */
-  override fun getList(): List<GameList> =
-    projectionsRepository.getList()
 }

src/main/kotlin/eventDemo/contexts/game/infrastructure/persistence/projections/GameListRepositoryInRedis.kt

@@ -1,49 +0,0 @@
-package eventDemo.contexts.game.infrastructure.persistence.projections
-
-import eventDemo.contexts.game.application.ports.GameEventBus
-import eventDemo.contexts.game.application.ports.GameProjectionBus
-import eventDemo.contexts.game.application.projections.apply
-import eventDemo.contexts.game.domain.game.GameId
-import eventDemo.contexts.game.infrastructure.persistence.projections.models.GameList
-import eventDemo.domain.event.projection.GameListRepository
-import eventDemo.libs.eventSource.projection.ProjectionRepositoryInRedis
-import io.github.oshai.kotlinlogging.withLoggingContext
-import kotlinx.serialization.json.Json
-import redis.clients.jedis.UnifiedJedis
-
-/**
- * Manages [projections][GameList], their building and publication in the [bus][GameProjectionBus].
- */
-class GameListRepositoryInRedis(
-  jedis: UnifiedJedis,
-  val projectionBus: GameProjectionBus,
-  val eventBus: GameEventBus,
-) : GameListRepository {
-  private val projectionsRepository =
-    ProjectionRepositoryInRedis(
-      initialStateBuilder = { aggregateId: GameId -> GameList(aggregateId) },
-      projectionClass = GameList::class,
-      projectionToJson = { Json.encodeToString(GameList.serializer(), it) },
-      jsonToProjection = { Json.decodeFromString(GameList.serializer(), it) },
-      applyToProjection = GameList::apply,
-      jedis = jedis,
-    )
-
-  fun subscribeToBus() {
-    eventBus.subscribe { event ->
-      withLoggingContext("event" to event.toString()) {
-        projectionsRepository
-          .applyAndSave(event)
-          .also { projectionBus.publish(it) }
-      }
-    }
-  }
-
-  /**
-   * Get the last version of the [GameState] from the all eventStream.
-   *
-   * It fetches it from the local cache if possible, otherwise it builds it.
-   */
-  override fun getList(): List<GameList> =
-    projectionsRepository.getList()
-}

src/main/kotlin/eventDemo/contexts/game/infrastructure/persistence/projections/models/GameList.kt

@@ -2,7 +2,6 @@ package eventDemo.contexts.game.infrastructure.persistence.projections.models
 
 import eventDemo.contexts.game.domain.game.GameId
 import eventDemo.contexts.game.domain.game.Player
-import eventDemo.libs.eventSource.projection.Projection
 import kotlinx.serialization.Serializable
 
 /**
@@ -10,11 +9,10 @@ import kotlinx.serialization.Serializable
  */
 @Serializable
 data class GameList(
-  override val aggregateId: GameId,
+  val aggregateId: GameId,
-  override val lastEventVersion: Int = 0,
   val status: Status = Status.OPENING,
   val players: Set<Player> = emptySet(),
-  val winners: Set<Player> = emptySet(),
+  val winners: Set<Player.PlayerId> = emptySet(),
 ) : GameProjection {
   enum class Status {
     OPENING,

src/main/kotlin/eventDemo/contexts/game/infrastructure/persistence/projections/models/GameProjection.kt

@@ -1,8 +1,6 @@
 package eventDemo.contexts.game.infrastructure.persistence.projections.models
 
-import eventDemo.contexts.game.domain.game.GameId
-import eventDemo.libs.eventSource.projection.Projection
 import kotlinx.serialization.Serializable
 
 @Serializable
-sealed interface GameProjection : Projection<GameId>
+sealed interface GameProjection

src/main/kotlin/eventDemo/libs/bus/BusInRabbitMQ.kt

@@ -7,6 +7,7 @@ import com.rabbitmq.client.ConnectionFactory
 import com.rabbitmq.client.DefaultConsumer
 import com.rabbitmq.client.Envelope
 import io.github.oshai.kotlinlogging.KotlinLogging
+import io.github.oshai.kotlinlogging.withLoggingContext
 import io.ktor.utils.io.core.toByteArray
 import kotlinx.coroutines.runBlocking
 
@@ -43,15 +44,17 @@ class BusInRabbitMQ<E>(
   }
 
   override fun publish(item: E) {
-    connection
+    withLoggingContext("item" to item.toString()) {
-      .createChannel()
+      connection
-      .basicPublish(
+        .createChannel()
-        exchangeName,
+        .basicPublish(
-        routingKey,
+          exchangeName,
-        AMQP.BasicProperties(),
+          routingKey,
-        objectToString(item).toByteArray(),
+          AMQP.BasicProperties(),
-      )
+          objectToString(item).toByteArray(),
-    logger.info { "Item sent to the bus" }
+        )
+      logger.info { "Item sent to the bus" }
+    }
   }
 
   override fun subscribe(block: (E) -> Unit): Bus.Subscription {

src/main/kotlin/eventDemo/libs/eventSource/Event.kt

@@ -1,6 +1,8 @@
 package eventDemo.libs.eventSource
 
+import eventDemo.libs.serializer.UUIDSerializer
 import kotlinx.datetime.Instant
+import kotlinx.serialization.Serializable
 import java.util.UUID
 
 /**
@@ -23,6 +25,8 @@ interface Event<ID : AggregateId> {
 }
 
 @JvmInline
+@Serializable
 value class EventId(
+  @Serializable(with = UUIDSerializer::class)
   val id: UUID = UUID.randomUUID(),
 )

src/main/kotlin/eventDemo/libs/eventSource/eventStore/EventStore.kt

@@ -7,11 +7,13 @@ import io.github.oshai.kotlinlogging.withLoggingContext
 interface EventStore<E : Event<ID>, ID : AggregateId> {
   fun getStream(aggregateId: ID): EventStream<E, ID>
 
-  fun publish(event: E) =
+  @Throws(VersionConflictException::class)
+  fun append(event: E) =
     withLoggingContext("event" to event.toString()) {
-      getStream(event.aggregateId).publish(event)
+      getStream(event.aggregateId).append(event)
     }
 
-  fun publish(events: Set<E>) =
+  @Throws(VersionConflictException::class)
-    events.forEach { publish(it) }
+  fun append(events: Set<E>) =
+    events.forEach { append(it) }
 }

src/main/kotlin/eventDemo/libs/eventSource/eventStore/EventStream.kt

@@ -2,7 +2,6 @@ package eventDemo.libs.eventSource.eventStore
 
 import eventDemo.libs.eventSource.AggregateId
 import eventDemo.libs.eventSource.Event
-import eventDemo.libs.eventSource.projection.Projection
 import io.github.oshai.kotlinlogging.withLoggingContext
 
 /**
@@ -12,13 +11,14 @@ interface EventStream<E : Event<ID>, ID : AggregateId> {
   val aggregateId: ID
 
   /** Publishes a single event to the event stream */
-  fun publish(event: E)
+  @Throws(VersionConflictException::class)
+  fun append(event: E)
 
   /** Publishes multiple events to the event stream */
-  fun publish(vararg events: E) {
+  fun append(vararg events: E) {
     events.forEach {
       withLoggingContext("event" to it.toString()) {
-        publish(it)
+        append(it)
       }
     }
   }
@@ -31,14 +31,12 @@ interface EventStream<E : Event<ID>, ID : AggregateId> {
 
   fun readVersionBetween(version: IntRange): Set<E>
 
-  fun <P : Projection<*>> readVersionBetween(
-    projection: P?,
-    event: E,
-  ): Set<E> =
-    readVersionBetween(((projection?.lastEventVersion ?: 0) + 1)..event.version)
-
   fun getByVersion(version: Int): E? =
     readVersionBetween(version..version).firstOrNull()
 
   fun exist(): Boolean
 }
+
+class VersionConflictException(
+  event: Event<*>,
+) : RuntimeException("Version conflict: ${event.version}")

src/main/kotlin/eventDemo/libs/eventSource/eventStore/EventStreamInMemory.kt

@@ -17,7 +17,7 @@ class EventStreamInMemory<E : Event<ID>, ID : AggregateId>(
   private val logger = KotlinLogging.logger {}
   private val events: Queue<E> = ConcurrentLinkedQueue()
 
-  override fun publish(event: E) {
+  override fun append(event: E) {
     if (event.aggregateId != aggregateId) {
       throw EventStreamPublishException(
         "You cannot publish this event in this stream because it has a different aggregateId!",

src/main/kotlin/eventDemo/libs/eventSource/eventStore/EventStreamInPostgresql.kt

@@ -5,6 +5,7 @@ import eventDemo.libs.eventSource.Event
 import io.github.oshai.kotlinlogging.KotlinLogging
 import io.github.oshai.kotlinlogging.withLoggingContext
 import org.postgresql.util.PGobject
+import org.postgresql.util.PSQLException
 import javax.sql.DataSource
 
 /**
@@ -21,29 +22,39 @@ class EventStreamInPostgresql<E : Event<ID>, ID : AggregateId>(
 ) : EventStream<E, ID> {
   private val logger = KotlinLogging.logger {}
 
-  override fun publish(event: E) {
+  override fun append(event: E) {
     withLoggingContext("event" to event.toString()) {
       if (event.aggregateId != aggregateId) {
         throw EventStreamPublishException(
           "You cannot publish this event in this stream because it has a different aggregateId!",
         )
       }
-      dataSource.connection.use { connection ->
+      try {
-        connection
+        dataSource.connection.use { connection ->
-          .prepareStatement(
+          connection
-            """
+            .prepareStatement(
-            insert into $tableName (id, aggregate_id, version, data)
+              """
-            values (?, ?, ?, ?)
+              insert into $tableName (id, aggregate_id, version, data)
-            """.trimIndent(),
+              values (?, ?, ?, ?)
-          ).use {
+              on conflict (id) do nothing
-            it.setObject(1, event.eventId.id)
+              """.trimIndent(),
-            it.setObject(2, event.aggregateId.id)
+            ).use {
-            it.setInt(3, event.version)
+              it.setObject(1, event.eventId.id)
-            it.setObject(4, PGJsonb(objectToString(event)))
+              it.setObject(2, event.aggregateId.id)
-            it.executeUpdate()
+              it.setInt(3, event.version)
-          }
+              it.setObject(4, PGJsonb(objectToString(event)))
+              it.executeUpdate()
+            }
+        }
+      } catch (e: PSQLException) {
+        if (e.serverErrorMessage?.constraint == "game_event_stream_aggregate_id_version_key") {
+          logger.warn { "duplicate version" }
+          throw VersionConflictException(event)
+        } else {
+          throw e
+        }
       }
-      logger.info { "Event published" }
+      logger.info { "Event appended" }
     }
   }
 

src/main/kotlin/eventDemo/libs/eventSource/projection/Projection.kt

@@ -1,8 +0,0 @@
-package eventDemo.libs.eventSource.projection
-
-import eventDemo.libs.eventSource.AggregateId
-
-interface Projection<ID : AggregateId> {
-  val aggregateId: ID
-  val lastEventVersion: Int
-}

src/main/kotlin/eventDemo/libs/eventSource/projection/ProjectionRepository.kt

@@ -1,34 +0,0 @@
-package eventDemo.libs.eventSource.projection
-
-import eventDemo.libs.eventSource.AggregateId
-import eventDemo.libs.eventSource.Event
-
-interface ProjectionRepository<E : Event<ID>, P : Projection<ID>, ID : AggregateId> {
-  /**
-   * Update projection with the event.
-   */
-  fun apply(event: E): P
-
-  /**
-   * Update projection with the event, and save it.
-   */
-  fun applyAndSave(event: E): P
-
-  /**
-   * Save the projection.
-   */
-  fun save(projection: P)
-
-  /**
-   * Build the list of all [Projections][Projection]
-   */
-  fun getList(
-    limit: Int = 100,
-    offset: Int = 0,
-  ): List<P>
-
-  /**
-   * Build the last version of the [Projection] from the cache.
-   */
-  fun get(aggregateId: ID): P
-}

src/main/kotlin/eventDemo/libs/eventSource/projection/ProjectionRepositoryAbs.kt

@@ -1,63 +0,0 @@
-package eventDemo.libs.eventSource.projection
-
-import eventDemo.libs.eventSource.AggregateId
-import eventDemo.libs.eventSource.Event
-import io.github.oshai.kotlinlogging.KotlinLogging
-import io.github.oshai.kotlinlogging.withLoggingContext
-
-/**
- * Repository abstraction to declare common process
- */
-abstract class ProjectionRepositoryAbs<E : Event<ID>, P : Projection<ID>, ID : AggregateId>(
-  private val applyToProjection: P.(event: E) -> P,
-) : ProjectionRepository<E, P, ID> {
-  private val logger = KotlinLogging.logger {}
-
-  /**
-   * Update projection with the event.
-   *
-   * 1. get the last projection
-   * 2. apply the new event to the projection
-   */
-  override fun apply(event: E): P =
-    get(event.aggregateId).applyToProjectionSecure(event)
-
-  /**
-   * Update projection with the event, and save it.
-   *
-   * 1. get the last projection
-   * 2. apply the new event to projection
-   * 3. save it
-   */
-  override fun applyAndSave(event: E): P =
-    apply(event)
-      .also {
-        withLoggingContext("projection" to it.toString(), "event" to event.toString()) {
-          save(it)
-        }
-      }
-
-  /**
-   * Wrap the [applyToProjection] lambda to avoid duplicate apply of the same event.
-   */
-  protected val applyToProjectionSecure: P.(event: E) -> P = { event ->
-    withLoggingContext("event" to event.toString(), "projection" to this.toString()) {
-      if (canBeApply(event)) {
-        applyToProjection(event)
-      } else if (event.version <= lastEventVersion) {
-        "Event is already in the Projection, skip apply.".let {
-          logger.warn { it }
-          error(it)
-        }
-      } else {
-        "The version of the event must follow directly after the version of the projection.".let {
-          logger.error { it }
-          error(it)
-        }
-      }
-    }
-  }
-
-  private fun P.canBeApply(event: E): Boolean =
-    event.version == lastEventVersion + 1
-}

src/main/kotlin/eventDemo/libs/eventSource/projection/ProjectionRepositoryInMemory.kt

@@ -1,46 +0,0 @@
-package eventDemo.libs.eventSource.projection
-
-import eventDemo.libs.eventSource.AggregateId
-import eventDemo.libs.eventSource.Event
-import java.util.concurrent.ConcurrentHashMap
-
-class ProjectionRepositoryInMemory<E : Event<ID>, P : Projection<ID>, ID : AggregateId>(
-  private val initialStateBuilder: (aggregateId: ID) -> P,
-  applyToProjection: P.(event: E) -> P,
-) : ProjectionRepositoryAbs<E, P, ID>(applyToProjection),
-  ProjectionRepository<E, P, ID> {
-  private val projections: ConcurrentHashMap<ID, P> = ConcurrentHashMap()
-
-  /**
-   * Build the list of all [Projections][Projection]
-   */
-  override fun getList(
-    limit: Int,
-    offset: Int,
-  ): List<P> =
-    projections
-      .values
-      .drop(offset)
-      .take(limit)
-
-  /**
-   * Get the [Projection].
-   */
-  override fun get(aggregateId: ID): P =
-    projections[aggregateId]
-      ?: initialStateBuilder(aggregateId)
-
-  /**
-   * Save the projection.
-   */
-  override fun save(projection: P) {
-    projections.compute(projection.aggregateId) { id: ID, proj: P? ->
-      val currentProjection = proj ?: initialStateBuilder(projection.aggregateId)
-      if (currentProjection.lastEventVersion < projection.lastEventVersion) {
-        projection
-      } else {
-        currentProjection
-      }
-    }
-  }
-}

src/main/kotlin/eventDemo/libs/eventSource/projection/ProjectionRepositoryInRedis.kt

@@ -1,80 +0,0 @@
-package eventDemo.libs.eventSource.projection
-
-import eventDemo.libs.eventSource.AggregateId
-import eventDemo.libs.eventSource.Event
-import io.github.oshai.kotlinlogging.KotlinLogging
-import redis.clients.jedis.UnifiedJedis
-import redis.clients.jedis.params.ScanParams
-import java.util.concurrent.locks.ReentrantLock
-import kotlin.concurrent.withLock
-import kotlin.reflect.KClass
-
-class ProjectionRepositoryInRedis<E : Event<ID>, P : Projection<ID>, ID : AggregateId>(
-  private val jedis: UnifiedJedis,
-  private val initialStateBuilder: (aggregateId: ID) -> P,
-  private val projectionClass: KClass<P>,
-  private val projectionToJson: (P) -> String,
-  private val jsonToProjection: (String) -> P,
-  applyToProjection: P.(event: E) -> P,
-) : ProjectionRepositoryAbs<E, P, ID>(applyToProjection),
-  ProjectionRepository<E, P, ID> {
-  val logger = KotlinLogging.logger { }
-  private val lock = ReentrantLock()
-
-  /**
-   * Get the list of all [Projections][Projection]
-   */
-  override fun getList(
-    limit: Int,
-    offset: Int,
-  ): List<P> =
-    jedis
-      .hscan(
-        projectionClass.redisHKey,
-        offset.toString(),
-        ScanParams()
-          .match("*")
-          .count(limit),
-      ).result
-      .mapNotNull {
-        jsonToProjection(it.value)
-      }
-
-  /**
-   * Get the [Projection].
-   */
-  override fun get(aggregateId: ID): P =
-    jedis
-      .hget(
-        projectionClass.redisHKey,
-        aggregateId.id.toString(),
-      ).let {
-        if (it == null || it == "nil") {
-          initialStateBuilder(aggregateId)
-        } else {
-          jsonToProjection(it)
-        }
-      }
-
-  override fun save(projection: P) {
-    lock.withLock {
-      if (get(projection.aggregateId).lastEventVersion < projection.lastEventVersion) {
-        jedis.hset(
-          projection.redisHKey,
-          projection.aggregateId.id.toString(),
-          projectionToJson(projection),
-        )
-        logger.info { "Projection saved" }
-      } else {
-        logger.error { "Projection save SKIP (an early version exists)" }
-        error("Projection save SKIP (an early version exists)")
-      }
-    }
-  }
-}
-
-private val <P : Projection<*>> KClass<P>.redisHKey: String get() =
-  "projection:$simpleName"
-
-private val <P : Projection<*>> P.redisHKey: String get() =
-  this::class.redisHKey

src/test/kotlin/eventDemo/NewUser.kt

@@ -1,21 +0,0 @@
-package eventDemo
-
-import eventDemo.contexts.auth.domain.User
-import eventDemo.sharedKernel.UserId
-import kotlin.reflect.KProperty
-
-class NewUser {
-  operator fun getValue(
-    thisRef: Any?,
-    property: KProperty<*>,
-  ): User =
-    newUser(property.name)
-}
-
-fun newUser(username: String): User =
-  User(
-    id = UserId(),
-    username = username,
-    password = "changeit",
-    version = 1,
-  )

src/test/kotlin/eventDemo/contexts/game/application/GameSimulationTest.kt

@@ -1,10 +1,5 @@
 package eventDemo.contexts.game.application
 
-import eventDemo.GameWithCommandsInChannels.createGameWithCommandsInChannels
-import eventDemo.GameWithCommandsInChannels.joinTheGame
-import eventDemo.GameWithCommandsInChannels.playCard
-import eventDemo.GameWithCommandsInChannels.readyToPlay
-import eventDemo.NewUser
 import eventDemo.Tag
 import eventDemo.contexts.game.application.channels.GameChannelsSubscriber
 import eventDemo.contexts.game.application.command.models.GameCommand
@@ -21,12 +16,16 @@ import eventDemo.contexts.game.domain.game.Card
 import eventDemo.contexts.game.domain.game.GameId
 import eventDemo.contexts.game.domain.game.gameState.Game
 import eventDemo.contexts.game.domain.game.gameState.GameStarted
-import eventDemo.testKoinApplicationWithConfig
+import eventDemo.testHelpers.CreateGameWithCommandsInChannelsHelpers.createGameWithCommandsInChannels
+import eventDemo.testHelpers.CreateGameWithCommandsInChannelsHelpers.joinTheGame
+import eventDemo.testHelpers.CreateGameWithCommandsInChannelsHelpers.playCard
+import eventDemo.testHelpers.CreateGameWithCommandsInChannelsHelpers.readyToPlay
+import eventDemo.testHelpers.createNewUser
+import eventDemo.testHelpers.testKoinApplicationWithConfig
 import io.kotest.assertions.nondeterministic.eventually
 import io.kotest.assertions.nondeterministic.until
 import io.kotest.core.spec.style.FunSpec
 import io.kotest.matchers.equals.shouldBeEqual
-import io.kotest.matchers.shouldNotBe
 import kotlinx.coroutines.DelicateCoroutinesApi
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.GlobalScope
@@ -47,8 +46,8 @@ class GameSimulationTest :
       withTimeout(10.seconds) {
         disableShuffleDeck()
         val gameId = GameId()
-        val user1 by NewUser()
+        val user1 = createNewUser("user1")
-        val user2 by NewUser()
+        val user2 = createNewUser("user2")
 
         val channelCommand1 = Channel<GameCommand>(Channel.BUFFERED)
         val channelCommand2 = Channel<GameCommand>(Channel.BUFFERED)

src/test/kotlin/eventDemo/contexts/game/application/eventStore/GameEventStoreRepositoryTest.kt

@@ -2,14 +2,16 @@ package eventDemo.contexts.game.application.eventStore
 
 import ch.qos.logback.classic.Level
 import com.rabbitmq.client.impl.ForgivingExceptionHandler
-import eventDemo.GameWithCommands
-import eventDemo.GameWithCommands.joinTheGame
-import eventDemo.NewUser
 import eventDemo.Tag
+import eventDemo.contexts.auth.application.eventStores.UserRepository
 import eventDemo.contexts.game.application.eventStores.GameEventStoreRepository
+import eventDemo.contexts.game.application.eventStores.GameRepository
 import eventDemo.contexts.game.domain.game.GameId
-import eventDemo.testKoinApplicationWithConfig
+import eventDemo.testHelpers.CreateGameWithCommandsHelpers
-import eventDemo.withLogLevel
+import eventDemo.testHelpers.CreateGameWithCommandsHelpers.joinTheGame
+import eventDemo.testHelpers.createNewUser
+import eventDemo.testHelpers.testKoinApplicationWithConfig
+import eventDemo.testHelpers.withLogLevel
 import io.kotest.assertions.nondeterministic.eventually
 import io.kotest.assertions.nondeterministic.eventuallyConfig
 import io.kotest.core.spec.style.FunSpec
@@ -29,12 +31,15 @@ class GameEventStoreRepositoryTest :
   FunSpec({
     tags(Tag.Postgresql)
 
-    val user1 by NewUser()
+    val user1 = createNewUser("user1")
-    val user2 by NewUser()
+    val user2 = createNewUser("user2")
 
     test("GameRepository should build return the game after dispatch commands") {
       testKoinApplicationWithConfig {
-        GameWithCommands.createGameWithCommands { getPlayer ->
+        get<UserRepository>().run {
+          save(user1)
+        }
+        CreateGameWithCommandsHelpers.createGameWithCommands {
           user1.joinTheGame()
           getPlayer(user1).userId shouldBeEqual user1.id
         }
@@ -45,11 +50,14 @@ class GameEventStoreRepositoryTest :
       withLogLevel(
         ForgivingExceptionHandler::class.java.name to Level.OFF,
       ) {
-        val gameId = GameId()
         testKoinApplicationWithConfig {
           val repo = get<GameEventStoreRepository>()
+          get<UserRepository>().run {
+            save(user1)
+            save(user2)
+          }
 
-          GameWithCommands.createGameWithCommands {
+          CreateGameWithCommandsHelpers.createGameWithCommands {
             user1.joinTheGame()
             assertNotNull(repo.get(gameId)).run {
               players.isNotEmpty() shouldBeEqual true
@@ -72,34 +80,39 @@ class GameEventStoreRepositoryTest :
         Logger.ROOT_LOGGER_NAME to Level.ERROR,
         ForgivingExceptionHandler::class.java.name to Level.OFF,
       ) {
-        val aggregateId = GameId()
+        var aggregateIds: MutableList<GameId> = mutableListOf()
         testKoinApplicationWithConfig {
-          val repo = get<GameEventStoreRepository>()
+          val repo = get<GameRepository>()
 
           val gameName = "testGame${UUID.randomUUID()}"
-          (1..10)
+          (1..2)
-            .map { r ->
+            .map { treadN ->
               GlobalScope
                 .launch {
-                  GameWithCommands.createGameWithCommands(gameName) {
+                  CreateGameWithCommandsHelpers.createGameWithCommands("testGame $treadN") {
-                    repeat(20) {
+                    repeat(3) { userN ->
-                      val userX by NewUser()
+                      val userX = createNewUser("userX $treadN:$userN")
+                      get<UserRepository>().save(userX)
                       userX.joinTheGame()
                     }
+                    aggregateIds.add(gameId)
                   }
                 }
             }.joinAll()
 
           eventually(
             eventuallyConfig {
-              duration = 60.seconds
+              duration = 5.seconds
-              interval = 2.seconds
+              interval = 1.seconds
               includeFirst = false
             },
           ) {
-            assertNotNull(repo.get(aggregateId)).run {
+            aggregateIds shouldHaveSize 2
-              version shouldBeEqual 200
+            aggregateIds.forEach {
-              players shouldHaveSize 200
+              assertNotNull(repo.get(it)).run {
+                version shouldBeEqual 4
+                players shouldHaveSize 3
+              }
             }
           }
         }

src/test/kotlin/eventDemo/contexts/game/application/notification/EventToNotificationSubscriberTest.kt

@@ -1,5 +1,7 @@
 package eventDemo.contexts.game.application.notification
 
+import eventDemo.contexts.auth.application.eventStores.UserEventStoreRepository
+import eventDemo.contexts.auth.infrastructure.persistence.eventStore.UserEventStoreInMemory
 import eventDemo.contexts.game.application.command.handlers.GameCommandHandlerDispatcher
 import eventDemo.contexts.game.application.command.handlers.JoinTheGameHandler
 import eventDemo.contexts.game.application.command.handlers.PlayCardHandler
@@ -8,11 +10,12 @@ import eventDemo.contexts.game.application.command.handlers.TakeCartFromDrawPile
 import eventDemo.contexts.game.application.command.models.JoinTheGameCommand
 import eventDemo.contexts.game.application.eventStores.GameEventStoreRepository
 import eventDemo.contexts.game.application.notification.models.Notification
-import eventDemo.contexts.game.application.notification.models.PlayerAsJoinTheGameNotification
+import eventDemo.contexts.game.application.notification.models.WelcomeToTheGameNotification
 import eventDemo.contexts.game.infrastructure.persistence.eventBus.GameEventBusInMemory
 import eventDemo.contexts.game.infrastructure.persistence.eventStore.GameEventStoreInMemory
 import eventDemo.sharedKernel.UserId
 import io.kotest.core.spec.style.FunSpec
+import io.kotest.matchers.collections.shouldContain
 import io.kotest.matchers.equals.shouldEqual
 import kotlinx.coroutines.GlobalScope
 import kotlinx.coroutines.channels.Channel
@@ -24,23 +27,24 @@ class EventToNotificationSubscriberTest :
 
     test("When event when send to the bus, a notification should be published") {
       val bus = GameEventBusInMemory()
-      val repository = GameEventStoreRepository(GameEventStoreInMemory())
+      val gameRepository = GameEventStoreRepository(GameEventStoreInMemory())
+      val userRepository = UserEventStoreRepository(UserEventStoreInMemory())
       val subscriber =
         EventToNotificationSubscriber(
           bus,
-          repository,
+          gameRepository,
         )
 
       val commentDispatcher =
         GameCommandHandlerDispatcher(
-          PlayCardHandler(repository, bus),
+          PlayCardHandler(gameRepository, bus),
-          ReadyToPlayHandler(repository, bus),
+          ReadyToPlayHandler(gameRepository, bus),
-          JoinTheGameHandler(repository, bus),
+          JoinTheGameHandler(gameRepository, bus, userRepository),
-          TakeCartFromDrawPileHandler(repository, bus),
+          TakeCartFromDrawPileHandler(gameRepository, bus),
         )
       val notificationChannel = Channel<Notification>(Channel.BUFFERED)
 
-      val game = repository.create()
+      val game = gameRepository.create()
 
       val user1 = UserId()
       val user2 = UserId()
@@ -64,8 +68,8 @@ class EventToNotificationSubscriberTest :
 
       player1Notifications.size shouldEqual 2
       player1Notifications.first().let { notification ->
-        assertInstanceOf<PlayerAsJoinTheGameNotification>(notification)
+        assertInstanceOf<WelcomeToTheGameNotification>(notification)
-        notification.player.userId shouldEqual user1
+        notification.players.map { it.userId } shouldContain user1
       }
     }
   })

src/test/kotlin/eventDemo/contexts/game/domain/game/PlayerHandTest.kt

@@ -1,8 +1,8 @@
 package eventDemo.contexts.game.domain.game
 
-import eventDemo.act
+import eventDemo.testHelpers.act
-import eventDemo.arrange
+import eventDemo.testHelpers.arrange
-import eventDemo.assert
+import eventDemo.testHelpers.assert
 import io.kotest.core.spec.style.FunSpec
 import io.kotest.matchers.ints.shouldBeExactly
 import org.junit.jupiter.api.assertInstanceOf

src/test/kotlin/eventDemo/contexts/game/domain/game/gameState/GameStartedTest.kt

@@ -1,7 +1,5 @@
 package eventDemo.contexts.game.domain.game.gameState
 
-import eventDemo.act
-import eventDemo.assert
 import eventDemo.contexts.game.domain.events.CardIsPlayedEvent
 import eventDemo.contexts.game.domain.events.DrawFilledWithDiscardEvent
 import eventDemo.contexts.game.domain.events.PlayerHaveDrawCardEvent
@@ -14,6 +12,8 @@ import eventDemo.contexts.game.domain.game.Player
 import eventDemo.contexts.game.domain.game.PlayerHand
 import eventDemo.contexts.game.domain.game.PlayerList
 import eventDemo.sharedKernel.UserId
+import eventDemo.testHelpers.act
+import eventDemo.testHelpers.assert
 import io.kotest.core.spec.style.FunSpec
 import io.kotest.datatest.withData
 import io.kotest.matchers.shouldBe

src/test/kotlin/eventDemo/contexts/game/intrastructure/AuthHelper.kt

@@ -2,14 +2,9 @@ package eventDemo.contexts.game.intrastructure
 
 import eventDemo.contexts.auth.domain.User
 import eventDemo.contexts.auth.infrastructure.configure.makeJwt
-import eventDemo.contexts.game.domain.game.Player
 import io.ktor.client.request.HttpRequestBuilder
 import io.ktor.client.request.header
 
 internal fun HttpRequestBuilder.withAuth(user: User) {
   header("Authorization", "Bearer ${user.makeJwt("secret")}")
 }
-
-internal fun HttpRequestBuilder.withAuth(player: Player) {
-  withAuth(User(player.userId, player.name, "changeit", 1))
-}

src/test/kotlin/eventDemo/contexts/game/intrastructure/persistence/connectors/PostgresqlTest.kt

@@ -1,7 +1,7 @@
 package eventDemo.contexts.game.intrastructure.persistence.connectors
 
 import eventDemo.Tag
-import eventDemo.testKoinApplicationWithConfig
+import eventDemo.testHelpers.testKoinApplicationWithConfig
 import io.kotest.core.spec.style.FunSpec
 import io.kotest.matchers.equals.shouldBeEqual
 import javax.sql.DataSource

src/test/kotlin/eventDemo/contexts/game/intrastructure/persistence/connectors/RabbitMQTest.kt

@@ -6,8 +6,8 @@ import com.rabbitmq.client.ConnectionFactory
 import com.rabbitmq.client.DefaultConsumer
 import com.rabbitmq.client.Envelope
 import eventDemo.Tag
-import eventDemo.spyPing
+import eventDemo.testHelpers.spyPing
-import eventDemo.testKoinApplicationWithConfig
+import eventDemo.testHelpers.testKoinApplicationWithConfig
 import io.kotest.core.spec.style.FunSpec
 import io.kotest.matchers.string.shouldStartWith
 import java.util.UUID

src/test/kotlin/eventDemo/contexts/game/intrastructure/persistence/connectors/RedisTest.kt

@@ -1,7 +1,7 @@
 package eventDemo.contexts.game.intrastructure.persistence.connectors
 
 import eventDemo.Tag
-import eventDemo.testKoinApplicationWithConfig
+import eventDemo.testHelpers.testKoinApplicationWithConfig
 import io.kotest.core.spec.style.FunSpec
 import io.kotest.matchers.equals.shouldBeEqual
 import redis.clients.jedis.UnifiedJedis

src/test/kotlin/eventDemo/contexts/game/intrastructure/persistence/eventBus/GameEventBusInRabbitMQTest.kt

@@ -8,8 +8,8 @@ import eventDemo.contexts.game.domain.game.Player
 import eventDemo.contexts.game.infrastructure.persistence.eventBus.GameEventBusInMemory
 import eventDemo.contexts.game.infrastructure.persistence.eventBus.GameEventBusInRabbinMQ
 import eventDemo.sharedKernel.UserId
-import eventDemo.spyPing
+import eventDemo.testHelpers.spyPing
-import eventDemo.testKoinApplicationWithConfig
+import eventDemo.testHelpers.testKoinApplicationWithConfig
 import io.kotest.core.spec.style.FunSpec
 import io.kotest.datatest.withData
 import io.kotest.matchers.equals.shouldBeEqual

src/test/kotlin/eventDemo/contexts/game/intrastructure/rest/GameListRouteTest.kt

@@ -1,18 +1,17 @@
 package eventDemo.contexts.game.intrastructure.rest
 
-import eventDemo.GameWithCommands
+import eventDemo.contexts.auth.application.eventStores.UserRepository
-import eventDemo.GameWithCommands.joinTheGame
-import eventDemo.GameWithCommands.readyToPlay
-import eventDemo.NewUser
-import eventDemo.contexts.game.domain.game.Player
 import eventDemo.contexts.game.infrastructure.persistence.projections.models.GameList
 import eventDemo.contexts.game.intrastructure.httpClient
 import eventDemo.contexts.game.intrastructure.withAuth
-import eventDemo.domain.event.projection.GameListRepository
+import eventDemo.testHelpers.CreateGameWithCommandsHelpers
-import eventDemo.sharedKernel.UserId
+import eventDemo.testHelpers.CreateGameWithCommandsHelpers.joinTheGame
-import eventDemo.testApplicationWithConfig
+import eventDemo.testHelpers.CreateGameWithCommandsHelpers.readyToPlay
+import eventDemo.testHelpers.createNewUser
+import eventDemo.testHelpers.testApplicationWithConfig
 import io.github.oshai.kotlinlogging.KotlinLogging
 import io.kotest.assertions.nondeterministic.eventually
+import io.kotest.assertions.nondeterministic.eventuallyConfig
 import io.kotest.core.spec.style.FunSpec
 import io.kotest.matchers.collections.shouldContain
 import io.kotest.matchers.collections.shouldHaveSize
@@ -24,6 +23,7 @@ import io.ktor.client.statement.bodyAsText
 import io.ktor.http.ContentType
 import io.ktor.http.HttpStatusCode
 import kotlin.test.assertEquals
+import kotlin.test.assertNotNull
 import kotlin.test.assertTrue
 import kotlin.time.Duration.Companion.seconds
 
@@ -32,13 +32,14 @@ val logger = KotlinLogging.logger {}
 class GameListRouteTest :
   FunSpec({
     test("/games with no game started") {
-
+      val user1 = createNewUser("user1")
-      testApplicationWithConfig {
+      testApplicationWithConfig({
-        val player1 = Player(name = "Nikola", UserId())
+        get<UserRepository>().save(user1)
+      }) {
         logger.info { "Starting player1" }
         httpClient()
           .get("/games") {
-            withAuth(player1)
+            withAuth(user1)
             accept(ContentType.Application.Json)
           }.apply {
             assertEquals(HttpStatusCode.OK, status, message = bodyAsText())
@@ -49,25 +50,36 @@ class GameListRouteTest :
     }
 
     test("/games return a game with status OPENING") {
-      val user1 by NewUser()
+      val user1 = createNewUser("user1")
       testApplicationWithConfig({
-        GameWithCommands.createGameWithCommands {
+        get<UserRepository>().save(user1)
+        CreateGameWithCommandsHelpers.createGameWithCommands {
           user1.joinTheGame()
         }
       }) {
         // Wait until the projection is created
-        eventually(3.seconds) {
+        eventually(
+          eventuallyConfig {
+            initialDelay = 1.seconds
+            interval = 1.seconds
+            duration = 3.seconds
+          },
+        ) {
           httpClient()
             .get("/games") {
               withAuth(user1)
               accept(ContentType.Application.Json)
             }.apply {
               assertEquals(HttpStatusCode.OK, status, message = bodyAsText())
-              call.body<List<GameList>>().first().let {
+              call.body<List<GameList>>().let {
-                it.status shouldBeEqual GameList.Status.OPENING
+                assertNotNull(it)
-                it.players shouldHaveSize 1
+                assertTrue { it.isNotEmpty() }
-                it.players.map { it.userId } shouldContain user1.id
+                it.firstOrNull()?.run {
-                it.winners shouldHaveSize 0
+                  status shouldBeEqual GameList.Status.OPENING
+                  players shouldHaveSize 1
+                  players.map { it.userId } shouldContain user1.id
+                  winners shouldHaveSize 0
+                }
               }
             }
         }
@@ -75,17 +87,27 @@ class GameListRouteTest :
     }
 
     test("/games return a game with status IS_STARTED") {
-      val user1 by NewUser()
+      val user1 = createNewUser("user1")
-      val user2 by NewUser()
+      val user2 = createNewUser("user2")
       testApplicationWithConfig({
-        GameWithCommands.createGameWithCommands { getPlayer ->
+        CreateGameWithCommandsHelpers.createGameWithCommands {
+          get<UserRepository>().run {
+            save(user1)
+            save(user2)
+          }
           user1.joinTheGame()
           user2.joinTheGame()
           getPlayer(user1).readyToPlay()
           getPlayer(user2).readyToPlay()
         }
       }) {
-        eventually(3.seconds) {
+        eventually(
+          eventuallyConfig {
+            initialDelay = 1.seconds
+            interval = 1.seconds
+            duration = 3.seconds
+          },
+        ) {
           httpClient()
             .get("/games") {
               withAuth(user1)

src/test/kotlin/eventDemo/libs/bus/BusTest.kt

@@ -1,7 +1,7 @@
 package eventDemo.libs.bus
 
 import com.rabbitmq.client.ConnectionFactory
-import eventDemo.spyPing
+import eventDemo.testHelpers.spyPing
 import io.kotest.core.spec.style.FunSpec
 import io.kotest.datatest.withData
 import io.kotest.matchers.string.shouldStartWith

src/test/kotlin/eventDemo/libs/command/CommandUnicityCheckerTest.kt

@@ -1,12 +1,8 @@
 package eventDemo.libs.command
 
-import eventDemo.spyPing
+import eventDemo.testHelpers.spyPing
-import io.kotest.assertions.nondeterministic.eventually
 import io.kotest.core.spec.style.FunSpec
-import io.mockk.spyk
-import io.mockk.verify
 import org.junit.jupiter.api.assertThrows
-import kotlin.time.Duration
 import kotlin.time.Duration.Companion.seconds
 
 class CommandUnicityCheckerTest :

src/test/kotlin/eventDemo/libs/eventSource/EventStreamTest.kt

@@ -5,8 +5,7 @@ import eventDemo.libs.eventSource.eventStore.EventStream
 import eventDemo.libs.eventSource.eventStore.EventStreamInMemory
 import eventDemo.libs.eventSource.eventStore.EventStreamInPostgresql
 import eventDemo.libs.eventSource.eventStore.EventStreamPublishException
-import eventDemo.testKoinApplicationWithConfig
+import eventDemo.testHelpers.testKoinApplicationWithConfig
-import io.kotest.common.KotestInternal
 import io.kotest.core.spec.style.FunSpec
 import io.kotest.datatest.withData
 import io.kotest.matchers.collections.shouldHaveSize
@@ -28,9 +27,9 @@ class EventStreamTest :
 
     fun EventStream<EventXTest, IdTest>.with3Events(block: EventStream<EventXTest, IdTest>.(id: IdTest) -> Unit) =
       also {
-        publish(EventXTest(aggregateId = aggregateId, version = 1, num = 1))
+        append(EventXTest(aggregateId = aggregateId, version = 1, num = 1))
-        publish(EventXTest(aggregateId = aggregateId, version = 2, num = 2))
+        append(EventXTest(aggregateId = aggregateId, version = 2, num = 2))
-        publish(EventXTest(aggregateId = aggregateId, version = 3, num = 3))
+        append(EventXTest(aggregateId = aggregateId, version = 3, num = 3))
         block(aggregateId)
       }
 
@@ -103,7 +102,7 @@ class EventStreamTest :
     context("publish should be throw error when publish another aggregate event") {
       testKoinApplicationWithConfig {
         withData(eventStreams()) {
-          assertThrows<EventStreamPublishException> { it.publish(EventXTest(aggregateId = IdTest(), version = 1, num = 1)) }
+          assertThrows<EventStreamPublishException> { it.append(EventXTest(aggregateId = IdTest(), version = 1, num = 1)) }
         }
       }
     }
@@ -115,7 +114,7 @@ class EventStreamTest :
             .map { i1 ->
               GlobalScope.launch {
                 (1..10).forEach { i2 ->
-                  stream.publish(
+                  stream.append(
                     EventXTest(
                       aggregateId = stream.aggregateId,
                       version = (i1 * 10) + i2,

src/test/kotlin/eventDemo/libs/eventSource/projection/ProjectionRepositoryTest.kt

@@ -1,223 +0,0 @@
-package eventDemo.libs.eventSource.projection
-
-import eventDemo.cleanProjections
-import eventDemo.libs.eventSource.AggregateId
-import eventDemo.libs.eventSource.Event
-import eventDemo.libs.eventSource.EventId
-import eventDemo.libs.eventSource.eventStore.EventStore
-import eventDemo.libs.eventSource.eventStore.EventStoreInMemory
-import eventDemo.libs.serializer.UUIDSerializer
-import io.kotest.assertions.nondeterministic.continually
-import io.kotest.core.spec.style.FunSpec
-import io.kotest.datatest.withData
-import io.kotest.engine.names.WithDataTestName
-import io.kotest.matchers.equals.shouldBeEqual
-import kotlinx.coroutines.DelicateCoroutinesApi
-import kotlinx.coroutines.GlobalScope
-import kotlinx.coroutines.joinAll
-import kotlinx.coroutines.launch
-import kotlinx.coroutines.runBlocking
-import kotlinx.datetime.Clock
-import kotlinx.datetime.Instant
-import kotlinx.serialization.Serializable
-import kotlinx.serialization.json.Json
-import redis.clients.jedis.JedisPooled
-import java.util.UUID
-import java.util.concurrent.locks.ReentrantLock
-import kotlin.concurrent.withLock
-import kotlin.test.assertNotNull
-import kotlin.time.Duration.Companion.seconds
-
-@OptIn(DelicateCoroutinesApi::class)
-class ProjectionRepositoryTest :
-  FunSpec({
-    data class TestData(
-      val store: EventStore<TestEvents, IdTest>,
-      val repository: ProjectionRepository<TestEvents, ProjectionTest, IdTest>,
-    ) : WithDataTestName {
-      override fun dataTestName(): String =
-        "${repository::class.simpleName} with ${store::class.simpleName}"
-    }
-
-    val eventStores =
-      listOf(
-        { EventStoreInMemory<TestEvents, IdTest>() },
-      )
-    val projectionRepo =
-      listOf(
-        ::getRepoInMemoryTest,
-        ::getRepoInRedisTest,
-      )
-
-    val list =
-      eventStores.flatMap { store ->
-        projectionRepo.map { repo ->
-          TestData(store(), repo())
-        }
-      }
-
-    context("when call applyAndSave, the projection should be built and save to the repository") {
-      withData(list) { (eventStore, repo) ->
-        val aggregateId = IdTest()
-
-        val eventOther = Event2Test(value2 = "valOther", version = 1, aggregateId = IdTest())
-//        eventStore.publish(eventOther)
-        val p = repo.applyAndSave(eventOther)
-        println(p)
-        assertNotNull(repo.get(eventOther.aggregateId)).also {
-          assertNotNull(it.value) shouldBeEqual "valOther"
-        }
-
-        val event1 = Event1Test(value1 = "val1", version = 1, aggregateId = aggregateId)
-//        eventStore.publish(event1)
-        repo.applyAndSave(event1)
-        assertNotNull(repo.get(event1.aggregateId)).also {
-          assertNotNull(it.value) shouldBeEqual "val1"
-        }
-
-        val event2 = Event2Test(value2 = "val2", version = 2, aggregateId = aggregateId)
-//        eventStore.publish(event2)
-        repo.applyAndSave(event2)
-        assertNotNull(repo.get(event2.aggregateId)).also {
-          assertNotNull(it.value) shouldBeEqual "val1val2"
-        }
-      }
-    }
-
-    context("getList method must be return all inserted events") {
-      withData(list) { (eventStore, repo) ->
-        val aggregateId = IdTest()
-        val otherAggregateId = IdTest()
-
-        val eventOther = Event2Test(value2 = "valOther", version = 1, aggregateId = otherAggregateId)
-        eventStore.publish(eventOther)
-        repo.applyAndSave(eventOther)
-        assertNotNull(repo.get(eventOther.aggregateId)).also {
-          assertNotNull(it.value) shouldBeEqual "valOther"
-        }
-
-        val event1 = Event1Test(value1 = "val1", version = 1, aggregateId = aggregateId)
-        eventStore.publish(event1)
-        repo.applyAndSave(event1)
-
-        val event2 = Event2Test(value2 = "val2", version = 2, aggregateId = aggregateId)
-        eventStore.publish(event2)
-        repo.applyAndSave(event2)
-
-        repo.getList().apply {
-          any { it.aggregateId == otherAggregateId } shouldBeEqual true
-          any { it.aggregateId == aggregateId } shouldBeEqual true
-          any { it.value == "val1val2" } shouldBeEqual true
-          any { it.value == "valOther" } shouldBeEqual true
-          any { it.lastEventVersion == 2 } shouldBeEqual true
-          any { it.lastEventVersion == 1 } shouldBeEqual true
-        }
-      }
-    }
-
-    context("ProjectionRepository should be thread safe") {
-      continually(1.seconds) {
-        withData(list) { (eventStore, repo) ->
-          val aggregateId = IdTest()
-          val lock = ReentrantLock()
-          var v = 0
-          (0..9)
-            .map {
-              GlobalScope.launch {
-                repeat(10) {
-                  lock.withLock {
-                    EventXTest(
-                      num = 1,
-                      version = ++v,
-                      aggregateId = aggregateId,
-                    ).also { repo.applyAndSave(it) }
-                  }
-                }
-              }
-            }.joinAll()
-          assertNotNull(repo.get(aggregateId)).lastEventVersion shouldBeEqual 100
-          assertNotNull(repo.get(aggregateId)).num shouldBeEqual 100
-        }
-      }
-    }
-  })
-
-@JvmInline
-@Serializable
-private value class IdTest(
-  @Serializable(with = UUIDSerializer::class)
-  override val id: UUID = UUID.randomUUID(),
-) : AggregateId
-
-@Serializable
-private data class ProjectionTest(
-  override val aggregateId: IdTest,
-  override val lastEventVersion: Int = 0,
-  var value: String? = null,
-  var num: Int = 0,
-) : Projection<IdTest>
-
-private sealed interface TestEvents : Event<IdTest>
-
-private data class Event1Test(
-  override val eventId: EventId = EventId(),
-  override val aggregateId: IdTest,
-  override val createdAt: Instant = Clock.System.now(),
-  override val version: Int,
-  val value1: String,
-) : TestEvents
-
-private data class Event2Test(
-  override val eventId: EventId = EventId(),
-  override val aggregateId: IdTest,
-  override val createdAt: Instant = Clock.System.now(),
-  override val version: Int,
-  val value2: String,
-) : TestEvents
-
-private data class EventXTest(
-  override val eventId: EventId = EventId(),
-  override val aggregateId: IdTest,
-  override val createdAt: Instant = Clock.System.now(),
-  override val version: Int,
-  val num: Int,
-) : TestEvents
-
-private fun getRepoInMemoryTest(): ProjectionRepository<TestEvents, ProjectionTest, IdTest> =
-  ProjectionRepositoryInMemory(
-    initialStateBuilder = { aggregateId: IdTest -> ProjectionTest(aggregateId) },
-    applyToProjection = apply,
-  )
-
-private fun getRepoInRedisTest(): ProjectionRepository<TestEvents, ProjectionTest, IdTest> {
-  val jedis = JedisPooled("redis://localhost:6379")
-  jedis.cleanProjections()
-  return ProjectionRepositoryInRedis(
-    jedis = jedis,
-    initialStateBuilder = { aggregateId: IdTest -> ProjectionTest(aggregateId) },
-    projectionClass = ProjectionTest::class,
-    projectionToJson = { Json.encodeToString(it) },
-    jsonToProjection = { Json.decodeFromString(it) },
-    applyToProjection = apply,
-  )
-}
-
-private val apply: ProjectionTest.(TestEvents) -> ProjectionTest = { event ->
-  this.let { projection ->
-    when (event) {
-      is Event1Test -> {
-        projection.copy(value = (projection.value.orEmpty()) + event.value1)
-      }
-
-      is Event2Test -> {
-        projection.copy(value = (projection.value.orEmpty()) + event.value2)
-      }
-
-      is EventXTest -> {
-        projection.copy(num = projection.num + event.num)
-      }
-    }.copy(
-      lastEventVersion = event.version,
-    )
-  }
-}

src/test/kotlin/eventDemo/testHelpers/CreateGameWithCommandsHelpers.kt

@@ -1,8 +1,7 @@
-package eventDemo
+package eventDemo.testHelpers
 
 import eventDemo.contexts.auth.domain.User
 import eventDemo.contexts.game.application.command.handlers.GameCommandHandlerDispatcher
-import eventDemo.contexts.game.application.command.models.GameCommand
 import eventDemo.contexts.game.application.command.models.JoinTheGameCommand
 import eventDemo.contexts.game.application.command.models.PlayCardCommand
 import eventDemo.contexts.game.application.command.models.ReadyToPlayCommand
@@ -10,51 +9,55 @@ import eventDemo.contexts.game.application.eventStores.GameRepository
 import eventDemo.contexts.game.domain.game.Card
 import eventDemo.contexts.game.domain.game.GameId
 import eventDemo.contexts.game.domain.game.Player
-import eventDemo.sharedKernel.UserId
-import kotlinx.coroutines.channels.Channel
 import org.koin.core.Koin
 import java.util.UUID
 
-object GameWithCommands {
+object CreateGameWithCommandsHelpers {
+  class Data(
+    private val repo: GameRepository,
+    val gameId: GameId,
+  ) {
+    fun getPlayer(user: User): Player =
+      repo.get(gameId)!!.players.get(user.id)
+  }
+
   context(koin: Koin)
   fun <T> createGameWithCommands(
     gameName: String = "testGame${UUID.randomUUID()}",
-    block: context(GameWithCommands, GameCommandHandlerDispatcher, GameId) ((User) -> Player) -> T,
+    block: context(CreateGameWithCommandsHelpers, GameCommandHandlerDispatcher) Data.() -> T,
   ): T {
     val gameId = GameId(UUID.nameUUIDFromBytes(gameName.encodeToByteArray()))
     val repo = koin.get<GameRepository>()
-    repo.getOrCreate(gameId)
+    repo.create(gameId)
 
     return koin.get<GameCommandHandlerDispatcher>().run {
-      with(gameId) {
+      with(CreateGameWithCommandsHelpers) {
-        with(GameWithCommands) {
+        Data(repo, gameId).block()
-          block { user -> repo.get(gameId)!!.players.get(user.id) }
-        }
       }
     }
   }
 
-  context(dispatcher: GameCommandHandlerDispatcher, gameId: GameId)
+  context(dispatcher: GameCommandHandlerDispatcher, data: Data)
   fun User.joinTheGame(): JoinTheGameCommand =
     JoinTheGameCommand(
       id,
-      JoinTheGameCommand.Payload(gameId),
+      JoinTheGameCommand.Payload(data.gameId),
     ).also { dispatcher.dispatch(it) }
 
-  context(dispatcher: GameCommandHandlerDispatcher, gameId: GameId)
+  context(dispatcher: GameCommandHandlerDispatcher, data: Data)
   fun Player.readyToPlay(): ReadyToPlayCommand =
     ReadyToPlayCommand(
       userId,
-      ReadyToPlayCommand.Payload(gameId, id),
+      ReadyToPlayCommand.Payload(data.gameId, id),
     ).also { dispatcher.dispatch(it) }
 
-  context(dispatcher: GameCommandHandlerDispatcher, gameId: GameId)
+  context(dispatcher: GameCommandHandlerDispatcher, data: Data)
   fun Player.playCard(
     card: Card,
     chosenColor: Card.Color? = null,
   ): PlayCardCommand =
     PlayCardCommand(
       userId,
-      PlayCardCommand.Payload(gameId, id, card, chosenColor),
+      PlayCardCommand.Payload(data.gameId, id, card, chosenColor),
     ).also { dispatcher.dispatch(it) }
 }

src/test/kotlin/eventDemo/testHelpers/CreateGameWithCommandsInChannelsHelpers.kt

@@ -1,4 +1,4 @@
-package eventDemo
+package eventDemo.testHelpers
 
 import eventDemo.contexts.auth.domain.User
 import eventDemo.contexts.game.application.command.models.GameCommand
@@ -9,12 +9,11 @@ import eventDemo.contexts.game.application.eventStores.GameRepository
 import eventDemo.contexts.game.domain.game.Card
 import eventDemo.contexts.game.domain.game.GameId
 import eventDemo.contexts.game.domain.game.Player
-import eventDemo.sharedKernel.UserId
 import kotlinx.coroutines.channels.Channel
 import org.koin.core.Koin
 import java.util.UUID
 
-object GameWithCommandsInChannels {
+object CreateGameWithCommandsInChannelsHelpers {
   context(koin: Koin)
   suspend fun <T> createGameWithCommandsInChannels(
     channelCommand: Channel<GameCommand>,
@@ -27,7 +26,7 @@ object GameWithCommandsInChannels {
 
     return with(gameId) {
       with(channelCommand) {
-        with(GameWithCommandsInChannels) {
+        with(CreateGameWithCommandsInChannelsHelpers) {
           block { user -> repo.get(gameId)!!.players.get(user.id) }
         }
       }

src/test/kotlin/eventDemo/testHelpers/LogHelper.kt

@@ -1,4 +1,4 @@
-package eventDemo
+package eventDemo.testHelpers
 
 import ch.qos.logback.classic.Level
 import ch.qos.logback.classic.Logger

src/test/kotlin/eventDemo/testHelpers/NewUserHelper.kt

@@ -0,0 +1,7 @@
+package eventDemo.testHelpers
+
+import eventDemo.contexts.auth.domain.User
+
+fun createNewUser(name: String): User =
+  User
+    .createNewUser(name, "changeit")

src/test/kotlin/eventDemo/testHelpers/TestAllCardCountHelpers.kt

@@ -1,4 +1,4 @@
-package eventDemo
+package eventDemo.testHelpers
 
 import eventDemo.contexts.game.domain.game.gameState.GameStarted
 

src/test/kotlin/eventDemo/testHelpers/TestApplicationHelpers.kt

@@ -1,4 +1,4 @@
-package eventDemo
+package eventDemo.testHelpers
 
 import com.zaxxer.hikari.HikariDataSource
 import eventDemo.configuration.appKoinModule
@@ -43,11 +43,9 @@ fun testApplicationWithConfig(
       logger.info { "Config App" }
       val koin = getKoin()
       koin.cleanDataTest()
-      runCatching {
+      logger.info { "Starting A" }
-        logger.info { "Starting A" }
+      configBuilder(koin)
-        configBuilder(koin)
+      logger.info { "A finish" }
-        logger.info { "A finish" }
-      }
     }
     logger.info { "Starting B" }
     this@testApplication.block()

src/test/kotlin/eventDemo/testHelpers/TestDataHelper.kt

@@ -1,4 +1,4 @@
-package eventDemo
+package eventDemo.testHelpers
 
 import org.koin.core.Koin
 import redis.clients.jedis.UnifiedJedis

src/test/kotlin/eventDemo/testHelpers/TestHelper.kt

@@ -1,4 +1,4 @@
-package eventDemo
+package eventDemo.testHelpers
 
 import io.kotest.assertions.nondeterministic.eventually
 import io.mockk.spyk