, 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
-
- /**
- * Build the last version of the [Projection] from the cache.
- */
- fun get(aggregateId: ID): P
-}
diff --git a/src/main/kotlin/eventDemo/libs/event/projection/ProjectionRepositoryAbs.kt b/src/main/kotlin/eventDemo/libs/event/projection/ProjectionRepositoryAbs.kt
deleted file mode 100644
index b349a82..0000000
--- a/src/main/kotlin/eventDemo/libs/event/projection/ProjectionRepositoryAbs.kt
+++ /dev/null
@@ -1,63 +0,0 @@
-package eventDemo.libs.event.projection
-
-import eventDemo.libs.event.AggregateId
-import eventDemo.libs.event.Event
-import io.github.oshai.kotlinlogging.KotlinLogging
-import io.github.oshai.kotlinlogging.withLoggingContext
-
-/**
- * Repository abstraction to declare common process
- */
-abstract class ProjectionRepositoryAbs, P : Projection, ID : AggregateId>(
- private val applyToProjection: P.(event: E) -> P,
-) : ProjectionRepository {
- 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
-}
diff --git a/src/main/kotlin/eventDemo/libs/event/projection/ProjectionRepositoryInMemory.kt b/src/main/kotlin/eventDemo/libs/event/projection/ProjectionRepositoryInMemory.kt
deleted file mode 100644
index 3098c48..0000000
--- a/src/main/kotlin/eventDemo/libs/event/projection/ProjectionRepositoryInMemory.kt
+++ /dev/null
@@ -1,46 +0,0 @@
-package eventDemo.libs.event.projection
-
-import eventDemo.libs.event.AggregateId
-import eventDemo.libs.event.Event
-import java.util.concurrent.ConcurrentHashMap
-
-class ProjectionRepositoryInMemory, P : Projection, ID : AggregateId>(
- private val initialStateBuilder: (aggregateId: ID) -> P,
- applyToProjection: P.(event: E) -> P,
-) : ProjectionRepositoryAbs(applyToProjection),
- ProjectionRepository {
- private val projections: ConcurrentHashMap = ConcurrentHashMap()
-
- /**
- * Build the list of all [Projections][Projection]
- */
- override fun getList(
- limit: Int,
- offset: Int,
- ): List =
- 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
- }
- }
- }
-}
diff --git a/src/main/kotlin/eventDemo/libs/event/projection/ProjectionRepositoryInRedis.kt b/src/main/kotlin/eventDemo/libs/event/projection/ProjectionRepositoryInRedis.kt
deleted file mode 100644
index eba9554..0000000
--- a/src/main/kotlin/eventDemo/libs/event/projection/ProjectionRepositoryInRedis.kt
+++ /dev/null
@@ -1,80 +0,0 @@
-package eventDemo.libs.event.projection
-
-import eventDemo.libs.event.AggregateId
-import eventDemo.libs.event.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, P : Projection, ID : AggregateId>(
- private val jedis: UnifiedJedis,
- private val initialStateBuilder: (aggregateId: ID) -> P,
- private val projectionClass: KClass,
- private val projectionToJson: (P) -> String,
- private val jsonToProjection: (String) -> P,
- applyToProjection: P.(event: E) -> P,
-) : ProjectionRepositoryAbs(applyToProjection),
- ProjectionRepository {
- val logger = KotlinLogging.logger { }
- private val lock = ReentrantLock()
-
- /**
- * Get the list of all [Projections][Projection]
- */
- override fun getList(
- limit: Int,
- offset: Int,
- ): List =
- 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
> KClass
.redisHKey: String get() =
- "projection:$simpleName"
-
-private val
> P.redisHKey: String get() =
- this::class.redisHKey
diff --git a/src/main/kotlin/eventDemo/libs/eventSource/Event.kt b/src/main/kotlin/eventDemo/libs/eventSource/Event.kt
new file mode 100644
index 0000000..2989a98
--- /dev/null
+++ b/src/main/kotlin/eventDemo/libs/eventSource/Event.kt
@@ -0,0 +1,32 @@
+package eventDemo.libs.eventSource
+
+import eventDemo.libs.serializer.UUIDSerializer
+import kotlinx.datetime.Instant
+import kotlinx.serialization.Serializable
+import java.util.UUID
+
+/**
+ * Represent an ID for one aggregate, and it used in events
+ * @see Event
+ */
+interface AggregateId {
+ val id: UUID
+}
+
+/**
+ * The basic interface for an Event
+ * @see eventDemo.libs.eventSource.eventStore.EventStream
+ */
+interface Event {
+ val eventId: EventId
+ val aggregateId: ID
+ val createdAt: Instant
+ val version: Int
+}
+
+@JvmInline
+@Serializable
+value class EventId(
+ @Serializable(with = UUIDSerializer::class)
+ val id: UUID = UUID.randomUUID(),
+)
diff --git a/src/main/kotlin/eventDemo/libs/eventSource/eventStore/EventStore.kt b/src/main/kotlin/eventDemo/libs/eventSource/eventStore/EventStore.kt
new file mode 100644
index 0000000..b2e3108
--- /dev/null
+++ b/src/main/kotlin/eventDemo/libs/eventSource/eventStore/EventStore.kt
@@ -0,0 +1,19 @@
+package eventDemo.libs.eventSource.eventStore
+
+import eventDemo.libs.eventSource.AggregateId
+import eventDemo.libs.eventSource.Event
+import io.github.oshai.kotlinlogging.withLoggingContext
+
+interface EventStore, ID : AggregateId> {
+ fun getStream(aggregateId: ID): EventStream
+
+ @Throws(VersionConflictException::class)
+ fun append(event: E) =
+ withLoggingContext("event" to event.toString()) {
+ getStream(event.aggregateId).append(event)
+ }
+
+ @Throws(VersionConflictException::class)
+ fun append(events: Set) =
+ events.forEach { append(it) }
+}
diff --git a/src/main/kotlin/eventDemo/libs/event/EventStoreInMemory.kt b/src/main/kotlin/eventDemo/libs/eventSource/eventStore/EventStoreInMemory.kt
similarity index 75%
rename from src/main/kotlin/eventDemo/libs/event/EventStoreInMemory.kt
rename to src/main/kotlin/eventDemo/libs/eventSource/eventStore/EventStoreInMemory.kt
index 9021a76..ac85b99 100644
--- a/src/main/kotlin/eventDemo/libs/event/EventStoreInMemory.kt
+++ b/src/main/kotlin/eventDemo/libs/eventSource/eventStore/EventStoreInMemory.kt
@@ -1,5 +1,7 @@
-package eventDemo.libs.event
+package eventDemo.libs.eventSource.eventStore
+import eventDemo.libs.eventSource.AggregateId
+import eventDemo.libs.eventSource.Event
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.ConcurrentMap
diff --git a/src/main/kotlin/eventDemo/libs/event/EventStoreInPostgresql.kt b/src/main/kotlin/eventDemo/libs/eventSource/eventStore/EventStoreInPostgresql.kt
similarity index 65%
rename from src/main/kotlin/eventDemo/libs/event/EventStoreInPostgresql.kt
rename to src/main/kotlin/eventDemo/libs/eventSource/eventStore/EventStoreInPostgresql.kt
index b15859e..1dc2882 100644
--- a/src/main/kotlin/eventDemo/libs/event/EventStoreInPostgresql.kt
+++ b/src/main/kotlin/eventDemo/libs/eventSource/eventStore/EventStoreInPostgresql.kt
@@ -1,12 +1,15 @@
-package eventDemo.libs.event
+package eventDemo.libs.eventSource.eventStore
+import eventDemo.libs.eventSource.AggregateId
+import eventDemo.libs.eventSource.Event
import javax.sql.DataSource
class EventStoreInPostgresql, ID : AggregateId>(
private val dataSource: DataSource,
private val objectToString: (E) -> String,
private val stringToObject: (String) -> E,
+ private val tableName: String,
) : EventStore {
override fun getStream(aggregateId: ID): EventStream =
- EventStreamInPostgresql(aggregateId, dataSource, objectToString, stringToObject)
+ EventStreamInPostgresql(aggregateId, dataSource, objectToString, stringToObject, tableName)
}
diff --git a/src/main/kotlin/eventDemo/libs/event/EventStream.kt b/src/main/kotlin/eventDemo/libs/eventSource/eventStore/EventStream.kt
similarity index 65%
rename from src/main/kotlin/eventDemo/libs/event/EventStream.kt
rename to src/main/kotlin/eventDemo/libs/eventSource/eventStore/EventStream.kt
index 5cf7a82..faf9b2e 100644
--- a/src/main/kotlin/eventDemo/libs/event/EventStream.kt
+++ b/src/main/kotlin/eventDemo/libs/eventSource/eventStore/EventStream.kt
@@ -1,6 +1,7 @@
-package eventDemo.libs.event
+package eventDemo.libs.eventSource.eventStore
-import eventDemo.libs.event.projection.Projection
+import eventDemo.libs.eventSource.AggregateId
+import eventDemo.libs.eventSource.Event
import io.github.oshai.kotlinlogging.withLoggingContext
/**
@@ -10,13 +11,14 @@ interface EventStream, 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)
}
}
}
@@ -29,12 +31,12 @@ interface EventStream, ID : AggregateId> {
fun readVersionBetween(version: IntRange): Set
- fun > readVersionBetween(
- projection: P?,
- event: E,
- ): Set =
- 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}")
diff --git a/src/main/kotlin/eventDemo/libs/event/EventStreamInMemory.kt b/src/main/kotlin/eventDemo/libs/eventSource/eventStore/EventStreamInMemory.kt
similarity index 81%
rename from src/main/kotlin/eventDemo/libs/event/EventStreamInMemory.kt
rename to src/main/kotlin/eventDemo/libs/eventSource/eventStore/EventStreamInMemory.kt
index 48607ad..9b8428b 100644
--- a/src/main/kotlin/eventDemo/libs/event/EventStreamInMemory.kt
+++ b/src/main/kotlin/eventDemo/libs/eventSource/eventStore/EventStreamInMemory.kt
@@ -1,5 +1,7 @@
-package eventDemo.libs.event
+package eventDemo.libs.eventSource.eventStore
+import eventDemo.libs.eventSource.AggregateId
+import eventDemo.libs.eventSource.Event
import io.github.oshai.kotlinlogging.KotlinLogging
import java.util.Queue
import java.util.concurrent.ConcurrentLinkedQueue
@@ -15,7 +17,7 @@ class EventStreamInMemory, ID : AggregateId>(
private val logger = KotlinLogging.logger {}
private val events: Queue = 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!",
@@ -34,4 +36,7 @@ class EventStreamInMemory, ID : AggregateId>(
events
.filter { version.contains(it.version) }
.toSet()
+
+ override fun exist(): Boolean =
+ events.isNotEmpty()
}
diff --git a/src/main/kotlin/eventDemo/libs/event/EventStreamInPostgresql.kt b/src/main/kotlin/eventDemo/libs/eventSource/eventStore/EventStreamInPostgresql.kt
similarity index 54%
rename from src/main/kotlin/eventDemo/libs/event/EventStreamInPostgresql.kt
rename to src/main/kotlin/eventDemo/libs/eventSource/eventStore/EventStreamInPostgresql.kt
index fe81fe5..a41f768 100644
--- a/src/main/kotlin/eventDemo/libs/event/EventStreamInPostgresql.kt
+++ b/src/main/kotlin/eventDemo/libs/eventSource/eventStore/EventStreamInPostgresql.kt
@@ -1,7 +1,11 @@
-package eventDemo.libs.event
+package eventDemo.libs.eventSource.eventStore
+import eventDemo.libs.eventSource.AggregateId
+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
/**
@@ -14,31 +18,44 @@ class EventStreamInPostgresql, ID : AggregateId>(
private val dataSource: DataSource,
private val objectToString: (E) -> String,
private val stringToObject: (String) -> E,
+ private val tableName: String,
) : EventStream {
private val logger = KotlinLogging.logger {}
- override fun publish(event: E) {
- if (event.aggregateId != aggregateId) {
- throw EventStreamPublishException(
- "You cannot publish this event in this stream because it has a different aggregateId!",
- )
- }
- dataSource.connection.use { connection ->
- connection
- .prepareStatement(
- """
- insert into event_stream(id, aggregate_id, version, data)
- values (?, ?, ?, ?)
- """.trimIndent(),
- ).use {
- it.setObject(1, event.eventId)
- it.setObject(2, event.aggregateId.id)
- it.setInt(3, event.version)
- it.setObject(4, PGJsonb(objectToString(event)))
- it.executeUpdate()
+ 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!",
+ )
+ }
+ try {
+ dataSource.connection.use { connection ->
+ connection
+ .prepareStatement(
+ """
+ insert into $tableName (id, aggregate_id, version, data)
+ values (?, ?, ?, ?)
+ on conflict (id) do nothing
+ """.trimIndent(),
+ ).use {
+ it.setObject(1, event.eventId.id)
+ it.setObject(2, event.aggregateId.id)
+ 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 appended" }
}
- logger.info { "Event published" }
}
override fun readAll(): Set =
@@ -46,8 +63,8 @@ class EventStreamInPostgresql, ID : AggregateId>(
connection
.prepareStatement(
"""
- select data
- from event_stream
+ select data
+ from $tableName
where aggregate_id = ?
order by version asc
""".trimIndent(),
@@ -66,13 +83,32 @@ class EventStreamInPostgresql, ID : AggregateId>(
}
}
+ override fun exist(): Boolean =
+ dataSource.connection.use { connection ->
+ connection
+ .prepareStatement(
+ """
+ select 1
+ from $tableName
+ where aggregate_id = ?
+ limit 1
+ order by version asc
+ """.trimIndent(),
+ ).use {
+ it.setObject(1, aggregateId.id)
+ it.executeQuery().use { resultSet ->
+ resultSet.next()
+ }
+ }
+ }
+
override fun readVersionBetween(version: IntRange): Set =
dataSource.connection.use { connection ->
connection
.prepareStatement(
"""
select data
- from event_stream
+ from $tableName
where version between ? and ?
and aggregate_id = ?
order by version asc
diff --git a/src/main/kotlin/eventDemo/libs/event/EventStreamPublishException.kt b/src/main/kotlin/eventDemo/libs/eventSource/eventStore/EventStreamPublishException.kt
similarity index 66%
rename from src/main/kotlin/eventDemo/libs/event/EventStreamPublishException.kt
rename to src/main/kotlin/eventDemo/libs/eventSource/eventStore/EventStreamPublishException.kt
index c2013f7..df58501 100644
--- a/src/main/kotlin/eventDemo/libs/event/EventStreamPublishException.kt
+++ b/src/main/kotlin/eventDemo/libs/eventSource/eventStore/EventStreamPublishException.kt
@@ -1,4 +1,4 @@
-package eventDemo.libs.event
+package eventDemo.libs.eventSource.eventStore
class EventStreamPublishException(
override val message: String,
diff --git a/src/main/kotlin/eventDemo/libs/FrameChannelConverter.kt b/src/main/kotlin/eventDemo/libs/helpers/FrameChannelConverter.kt
similarity index 86%
rename from src/main/kotlin/eventDemo/libs/FrameChannelConverter.kt
rename to src/main/kotlin/eventDemo/libs/helpers/FrameChannelConverter.kt
index e51fd3d..902dfcd 100644
--- a/src/main/kotlin/eventDemo/libs/FrameChannelConverter.kt
+++ b/src/main/kotlin/eventDemo/libs/helpers/FrameChannelConverter.kt
@@ -1,4 +1,4 @@
-package eventDemo.libs
+package eventDemo.libs.helpers
import io.github.oshai.kotlinlogging.KotlinLogging
import io.ktor.websocket.Frame
@@ -25,8 +25,9 @@ inline fun CoroutineScope.toObjectChannel(
return produce(capacity = bufferSize) {
frames.consumeEach { frame ->
if (frame is Frame.Text) {
- logger.debug { "Conversion of the Frame: ${frame.readText()}" }
- send(Json.decodeFromString(frame.readText()))
+ val frameText = frame.readText()
+ logger.debug { "Conversion of the Frame: $frameText to ${T::class.simpleName}" }
+ send(Json.decodeFromString(frameText))
} else {
logger.warn { "The frame is not a text frame" }
}
diff --git a/src/main/kotlin/eventDemo/libs/ListToRange.kt b/src/main/kotlin/eventDemo/libs/helpers/ListToRange.kt
similarity index 89%
rename from src/main/kotlin/eventDemo/libs/ListToRange.kt
rename to src/main/kotlin/eventDemo/libs/helpers/ListToRange.kt
index 4db29c3..bf15b76 100644
--- a/src/main/kotlin/eventDemo/libs/ListToRange.kt
+++ b/src/main/kotlin/eventDemo/libs/helpers/ListToRange.kt
@@ -1,4 +1,4 @@
-package eventDemo.libs
+package eventDemo.libs.helpers
fun List.toRanges(): List =
fold(listOf()) { acc, i ->
diff --git a/src/main/kotlin/eventDemo/libs/helpers/ReplaceValue.kt b/src/main/kotlin/eventDemo/libs/helpers/ReplaceValue.kt
new file mode 100644
index 0000000..72dd2ed
--- /dev/null
+++ b/src/main/kotlin/eventDemo/libs/helpers/ReplaceValue.kt
@@ -0,0 +1,25 @@
+package eventDemo.libs.helpers
+
+inline fun Map.withReplacedValue(
+ toReplace: K,
+ transform: (V) -> V,
+): Map =
+ mapValues {
+ if (it.key == toReplace) {
+ transform(it.value)
+ } else {
+ it.value
+ }
+ }
+
+inline fun Set.withReplacedValue(
+ toReplace: V,
+ transform: (V) -> V,
+): Set =
+ map {
+ if (it == toReplace) {
+ transform(it)
+ } else {
+ it
+ }
+ }.toSet()
diff --git a/src/main/kotlin/eventDemo/configuration/serializer/UUIDSerializer.kt b/src/main/kotlin/eventDemo/libs/serializer/UUIDSerializer.kt
similarity index 94%
rename from src/main/kotlin/eventDemo/configuration/serializer/UUIDSerializer.kt
rename to src/main/kotlin/eventDemo/libs/serializer/UUIDSerializer.kt
index 6328f10..92b186d 100644
--- a/src/main/kotlin/eventDemo/configuration/serializer/UUIDSerializer.kt
+++ b/src/main/kotlin/eventDemo/libs/serializer/UUIDSerializer.kt
@@ -1,4 +1,4 @@
-package eventDemo.configuration.serializer
+package eventDemo.libs.serializer
import kotlinx.serialization.KSerializer
import kotlinx.serialization.descriptors.PrimitiveKind
diff --git a/src/main/kotlin/eventDemo/sharedKernel/GetUserIdCredentials.kt b/src/main/kotlin/eventDemo/sharedKernel/GetUserIdCredentials.kt
new file mode 100644
index 0000000..71098b0
--- /dev/null
+++ b/src/main/kotlin/eventDemo/sharedKernel/GetUserIdCredentials.kt
@@ -0,0 +1,12 @@
+package eventDemo.sharedKernel
+
+import io.ktor.server.application.ApplicationCall
+import io.ktor.server.auth.jwt.JWTPrincipal
+import io.ktor.server.auth.principal
+import java.util.UUID
+
+internal val ApplicationCall.currentUserId: UserId
+ get() =
+ principal()!!.run {
+ UserId(UUID.fromString(payload.getClaim("userid").asString()))
+ }
diff --git a/src/main/kotlin/eventDemo/sharedKernel/UserId.kt b/src/main/kotlin/eventDemo/sharedKernel/UserId.kt
new file mode 100644
index 0000000..7c6a20a
--- /dev/null
+++ b/src/main/kotlin/eventDemo/sharedKernel/UserId.kt
@@ -0,0 +1,13 @@
+package eventDemo.sharedKernel
+
+import eventDemo.libs.eventSource.AggregateId
+import eventDemo.libs.serializer.UUIDSerializer
+import kotlinx.serialization.Serializable
+import java.util.UUID
+
+@Serializable
+@JvmInline
+value class UserId(
+ @Serializable(with = UUIDSerializer::class)
+ override val id: UUID = UUID.randomUUID(),
+) : AggregateId
diff --git a/src/main/resources/application.conf b/src/main/resources/application.conf
index 69b6a46..66a0f43 100644
--- a/src/main/resources/application.conf
+++ b/src/main/resources/application.conf
@@ -3,7 +3,7 @@ ktor {
port = 8080
}
application {
- modules = [ eventDemo.configuration.ConfigureKt.configure ]
+ modules = [ eventDemo.configuration.ConfigureKtorKt.configure ]
}
}
diff --git a/src/test/kotlin/eventDemo/adapter/infrastructure/event/GameEventBusInRabbitMQTest.kt b/src/test/kotlin/eventDemo/adapter/infrastructure/event/GameEventBusInRabbitMQTest.kt
deleted file mode 100644
index 32f2246..0000000
--- a/src/test/kotlin/eventDemo/adapter/infrastructure/event/GameEventBusInRabbitMQTest.kt
+++ /dev/null
@@ -1,48 +0,0 @@
-package eventDemo.adapter.infrastructure.event
-
-import com.rabbitmq.client.ConnectionFactory
-import eventDemo.domain.entity.GameId
-import eventDemo.domain.entity.Player
-import eventDemo.domain.event.GameEventBus
-import eventDemo.domain.event.event.NewPlayerEvent
-import eventDemo.testKoinApplicationWithConfig
-import io.kotest.assertions.nondeterministic.eventually
-import io.kotest.core.spec.style.FunSpec
-import io.kotest.datatest.withData
-import io.kotest.matchers.equals.shouldBeEqual
-import io.mockk.mockk
-import io.mockk.spyk
-import io.mockk.verify
-import java.util.function.Function
-import kotlin.time.Duration.Companion.seconds
-
-class GameEventBusInRabbitMQTest :
- FunSpec({
- context("Pub/sub") {
- testKoinApplicationWithConfig {
- val busListToTest: Map =
- mapOf(
- GameEventBusInMemory::class.java.simpleName to GameEventBusInMemory(),
- GameEventBusInRabbinMQ::class.java.simpleName to GameEventBusInRabbinMQ(get()),
- )
-
- withData(busListToTest) { bus ->
- val spy = spyk<() -> Unit>()
- val aggregateId = GameId()
- val player1 = Player(name = "Tesla")
- val player2 = Player(name = "Einstein")
-
- bus.subscribe { obj ->
- spy()
- obj.aggregateId shouldBeEqual aggregateId
- }
- bus.publish(NewPlayerEvent(aggregateId, player1, 1))
- bus.publish(NewPlayerEvent(aggregateId, player2, 2))
-
- eventually(1.seconds) {
- verify(exactly = 2) { spy() }
- }
- }
- }
- }
- })
diff --git a/src/test/kotlin/eventDemo/adapter/presenter/query/AuthHelper.kt b/src/test/kotlin/eventDemo/adapter/presenter/query/AuthHelper.kt
deleted file mode 100644
index 55d8823..0000000
--- a/src/test/kotlin/eventDemo/adapter/presenter/query/AuthHelper.kt
+++ /dev/null
@@ -1,10 +0,0 @@
-package eventDemo.adapter.presenter.query
-
-import eventDemo.domain.entity.Player
-import eventDemo.configuration.ktor.makeJwt
-import io.ktor.client.request.HttpRequestBuilder
-import io.ktor.client.request.header
-
-internal fun HttpRequestBuilder.withAuth(player: Player) {
- header("Authorization", "Bearer ${player.makeJwt("secret")}")
-}
diff --git a/src/test/kotlin/eventDemo/adapter/presenter/query/GameListRouteTest.kt b/src/test/kotlin/eventDemo/adapter/presenter/query/GameListRouteTest.kt
deleted file mode 100644
index 4c50d65..0000000
--- a/src/test/kotlin/eventDemo/adapter/presenter/query/GameListRouteTest.kt
+++ /dev/null
@@ -1,115 +0,0 @@
-package eventDemo.adapter.presenter.query
-
-import eventDemo.domain.entity.GameId
-import eventDemo.domain.entity.Player
-import eventDemo.domain.event.GameEventHandler
-import eventDemo.domain.event.event.GameStartedEvent
-import eventDemo.domain.event.event.NewPlayerEvent
-import eventDemo.domain.event.event.PlayerReadyEvent
-import eventDemo.domain.event.projection.GameList
-import eventDemo.testApplicationWithConfig
-import io.github.oshai.kotlinlogging.KotlinLogging
-import io.kotest.assertions.nondeterministic.eventually
-import io.kotest.core.spec.style.FunSpec
-import io.kotest.matchers.collections.shouldContain
-import io.kotest.matchers.collections.shouldHaveSize
-import io.kotest.matchers.equals.shouldBeEqual
-import io.ktor.client.call.body
-import io.ktor.client.request.accept
-import io.ktor.client.request.get
-import io.ktor.client.statement.bodyAsText
-import io.ktor.http.ContentType
-import io.ktor.http.HttpStatusCode
-import kotlin.test.assertEquals
-import kotlin.test.assertTrue
-import kotlin.time.Duration.Companion.seconds
-
-val logger = KotlinLogging.logger {}
-
-class GameListRouteTest :
- FunSpec({
- test("/games with no game started") {
-
- testApplicationWithConfig {
- val player1 = Player(name = "Nikola")
- logger.info { "Starting player1" }
- httpClient()
- .get("/games") {
- withAuth(player1)
- accept(ContentType.Application.Json)
- }.apply {
- assertEquals(HttpStatusCode.OK, status, message = bodyAsText())
- val list = call.body>()
- assertTrue(list.isEmpty())
- }
- }
- }
-
- test("/games return a game with status OPENING") {
- val gameId = GameId()
- val player1 = Player(name = "Nikola")
- testApplicationWithConfig(
- {
- get()
- .handle(gameId) {
- NewPlayerEvent(gameId, player1, it)
- }
- },
- ) {
- // Wait until the projection is created
- eventually(10.seconds) {
- httpClient()
- .get("/games") {
- withAuth(player1)
- accept(ContentType.Application.Json)
- }.apply {
- assertEquals(HttpStatusCode.OK, status, message = bodyAsText())
- call.body>().first().let {
- it.status shouldBeEqual GameList.Status.OPENING
- it.players shouldHaveSize 1
- it.players shouldContain player1
- it.winners shouldHaveSize 0
- }
- }
- }
- }
- }
-
- test("/games return a game with status IS_STARTED") {
- val gameId = GameId()
- val player1 = Player(name = "Nikola")
- val player2 = Player(name = "Einstein")
- testApplicationWithConfig({
- val eventHandler = get()
- eventHandler.handle(gameId) { NewPlayerEvent(gameId, player1, it) }
- eventHandler.handle(gameId) { NewPlayerEvent(gameId, player2, it) }
- eventHandler.handle(gameId) { PlayerReadyEvent(gameId, player1, it) }
- eventHandler.handle(gameId) { PlayerReadyEvent(gameId, player2, it) }
- eventHandler.handle(gameId) {
- GameStartedEvent.new(
- gameId,
- setOf(player1, player2),
- it,
- shuffleIsDisabled = true,
- )
- }
- }) {
- eventually(3.seconds) {
- httpClient()
- .get("/games") {
- withAuth(player1)
- accept(ContentType.Application.Json)
- }.apply {
- assertEquals(HttpStatusCode.OK, status, message = bodyAsText())
- call.body>().first().let {
- it.status shouldBeEqual GameList.Status.IS_STARTED
- it.players shouldHaveSize 2
- it.players shouldContain player1
- it.players shouldContain player2
- it.winners shouldHaveSize 0
- }
- }
- }
- }
- }
- })
diff --git a/src/test/kotlin/eventDemo/adapter/presenter/query/GameSimulationTest.kt b/src/test/kotlin/eventDemo/adapter/presenter/query/GameSimulationTest.kt
deleted file mode 100644
index 6efdcfe..0000000
--- a/src/test/kotlin/eventDemo/adapter/presenter/query/GameSimulationTest.kt
+++ /dev/null
@@ -1,196 +0,0 @@
-package eventDemo.adapter.presenter.query
-
-import eventDemo.Tag
-import eventDemo.domain.command.GameCommandHandler
-import eventDemo.domain.command.command.GameCommand
-import eventDemo.domain.command.command.IWantToJoinTheGameCommand
-import eventDemo.domain.command.command.IWantToPlayCardCommand
-import eventDemo.domain.command.command.IamReadyToPlayCommand
-import eventDemo.domain.entity.Card
-import eventDemo.domain.entity.GameId
-import eventDemo.domain.entity.Player
-import eventDemo.domain.event.event.disableShuffleDeck
-import eventDemo.domain.event.projection.GameState
-import eventDemo.domain.event.projection.GameStateRepository
-import eventDemo.domain.event.projection.projectionListener.PlayerNotificationListener
-import eventDemo.domain.notification.CommandSuccessNotification
-import eventDemo.domain.notification.ItsTheTurnOfNotification
-import eventDemo.domain.notification.Notification
-import eventDemo.domain.notification.PlayerAsJoinTheGameNotification
-import eventDemo.domain.notification.PlayerAsPlayACardNotification
-import eventDemo.domain.notification.PlayerWasReadyNotification
-import eventDemo.domain.notification.TheGameWasStartedNotification
-import eventDemo.domain.notification.WelcomeToTheGameNotification
-import eventDemo.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 kotlinx.coroutines.DelicateCoroutinesApi
-import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.GlobalScope
-import kotlinx.coroutines.channels.Channel
-import kotlinx.coroutines.channels.trySendBlocking
-import kotlinx.coroutines.joinAll
-import kotlinx.coroutines.launch
-import kotlinx.coroutines.withTimeout
-import kotlin.test.assertNotNull
-import kotlin.test.assertTrue
-import kotlin.time.Duration.Companion.seconds
-
-@DelicateCoroutinesApi
-class GameSimulationTest :
- FunSpec({
- tags(Tag.Postgresql)
-
- test("Simulation of a game") {
- withTimeout(10.seconds) {
- disableShuffleDeck()
- val gameId = GameId()
- val player1 = Player(name = "Nikola")
- val player2 = Player(name = "Einstein")
- val channelCommand1 = Channel(Channel.BUFFERED)
- val channelCommand2 = Channel(Channel.BUFFERED)
- val channelNotification1 = Channel(Channel.BUFFERED)
- val channelNotification2 = Channel(Channel.BUFFERED)
-
- var playedCard1: Card? = null
- var playedCard2: Card? = null
-
- var player1HasJoin = false
-
- testKoinApplicationWithConfig {
- val commandHandler = get()
- val playerNotificationListener = get()
- val gameStateRepository = get()
-
- // Run command handler
- // In the normal process, these handlers is invoque players connect to the websocket
- run {
- GlobalScope.launch(Dispatchers.IO) {
- commandHandler.handleIncomingPlayerCommands(player1, gameId, channelCommand1, channelNotification1)
- }
- GlobalScope.launch(Dispatchers.IO) {
- commandHandler.handleIncomingPlayerCommands(player2, gameId, channelCommand2, channelNotification2)
- }
- }
-
- // Consume etch notification of players, and put theses in a list.
- // Is used later to control when other players can execute the next action
- val player1Notifications = mutableListOf()
- val player2Notifications = mutableListOf()
- run {
- GlobalScope.launch {
- for (notification in channelNotification1) {
- player1Notifications.add(notification)
- }
- }
-
- GlobalScope.launch {
- for (notification in channelNotification2) {
- player2Notifications.add(notification)
- }
- }
- }
-
- // Player 1 actions
- val player1Job =
- launch {
- playerNotificationListener.startListening(player1, gameId) {
- channelNotification1.trySendBlocking(it)
- }
- IWantToJoinTheGameCommand(IWantToJoinTheGameCommand.Payload(gameId, player1)).also { sendCommand ->
- channelCommand1.send(sendCommand)
- player1Notifications.waitNotification { commandId == sendCommand.id }
- }
-
- player1HasJoin = true
-
- player1Notifications.waitNotification