Refactor event processing and add bookstore read models

Refactored event handling architecture by restructuring packages, introducing dedicated processor classes, and implementing in-memory checkpoint storage. Added services for building read models, including simulation tools for bookstore event streams and their processing, enabling efficient projection and testing scenarios.

Author: helightdev

.idea/dictionaries/project.xml

@@ -1,23 +1,27 @@
 # krescent
 
-This project uses [Gradle](https://gradle.org/).
-To build and run the application, use the *Gradle* tool window by clicking the Gradle icon in the right-hand toolbar,
-or run it directly from the terminal:
+Krescent is a Kotlin library for working with event-sourced systems in an extendable and easy to use and
+slightly opinionated manner.
 
-* Run `./gradlew run` to build and run the application.
-* Run `./gradlew build` to only build the application.
-* Run `./gradlew check` to run all checks, including tests.
-* Run `./gradlew clean` to clean all build outputs.
+The following event sources and projector targets are supported:
 
-Note the usage of the Gradle Wrapper (`./gradlew`).
-This is the suggested way to use Gradle in production projects.
+- KurrentDB (Event Source)
+- MongoDB (Projection, CheckpointStore)
 
-[Learn more about the Gradle Wrapper](https://docs.gradle.org/current/userguide/gradle_wrapper.html).
+## Installation
 
-[Learn more about Gradle tasks](https://docs.gradle.org/current/userguide/command_line_interface.html#common_tasks).
+Add jitpack as a repository in your `build.gradle.kts` file:
 
-This project follows the suggested multi-module setup and consists of the `app` and `utils` subprojects.
-The shared build logic was extracted to a convention plugin located in `buildSrc`.
+```kotlin
+repositories {
+    maven { url = uri("https://jitpack.io") }
+}
+```
 
-This project uses a version catalog (see `gradle/libs.versions.toml`) to declare and version dependencies
-and both a build cache and a configuration cache (see `gradle.properties`).
+Then, add the dependency to your `build.gradle.kts` file:
+
+```kotlin
+dependencies {
+    implementation("com.github.helightdev.krescent:krescent-core:main-SNAPSHOT")
+}
+```

README.md

@@ -0,0 +1,10 @@
+package dev.helight.krescent
+
+import java.util.function.Consumer
+
+interface HandlerChainParticipant {
+
+    fun accept(visitor: Consumer<HandlerChainParticipant>) {
+        visitor.accept(this)
+    }
+}

krescent-core/src/main/kotlin/dev/helight/krescent/EventMessageStreamProcessor.kt

@@ -0,0 +1,23 @@
+package dev.helight.krescent.checkpoint
+
+import kotlinx.serialization.json.JsonElement
+import kotlinx.serialization.json.JsonObject
+
+class CheckpointBucket(
+    val buffer: MutableMap<String, JsonElement>,
+) {
+
+    operator fun set(key: String, value: JsonElement) {
+        buffer.put(key, value)
+    }
+
+    operator fun get(key: String): JsonElement? {
+        return buffer[key]
+    }
+
+
+    fun buildJsonObject(): JsonElement = JsonObject(buffer)
+
+    override fun toString(): String = "CheckpointBucket{buffer=$buffer}"
+
+}

krescent-core/src/main/kotlin/dev/helight/krescent/HandlerChainParticipant.kt

@@ -0,0 +1,7 @@
+package dev.helight.krescent.checkpoint
+
+interface CheckpointStorage {
+    suspend fun storeCheckpoint(checkpoint: StoredCheckpoint)
+    suspend fun getLatestCheckpoint(namespace: String): StoredCheckpoint?
+    suspend fun clearCheckpoints()
+}

krescent-core/src/main/kotlin/dev/helight/krescent/checkpoint/CheckpointBucket.kt

@@ -1,6 +1,6 @@
-package dev.helight.krescent.checkpoints
+package dev.helight.krescent.checkpoint
 
-import dev.helight.krescent.EventMessage
+import dev.helight.krescent.event.EventMessage
 import java.time.Instant
 import kotlin.concurrent.atomics.AtomicLong
 import kotlin.concurrent.atomics.ExperimentalAtomicApi
@@ -11,12 +11,14 @@ interface CheckpointStrategy {
     suspend fun tick(eventMessage: EventMessage, lastCheckpoint: StoredCheckpoint?): Boolean
 }
 
-class NoCheckpointStrategy : CheckpointStrategy {
+@Suppress("unused")
+object NoCheckpointStrategy : CheckpointStrategy {
     override suspend fun tick(eventMessage: EventMessage, lastCheckpoint: StoredCheckpoint?): Boolean {
         return false
     }
 }
 
+@Suppress("unused")
 @OptIn(ExperimentalAtomicApi::class)
 class FixedEventRateCheckpointStrategy(
     private val checkpoint: Long,
@@ -29,8 +31,28 @@ class FixedEventRateCheckpointStrategy(
     }
 }
 
+@Suppress("unused")
+class ManualCheckpointStrategy() : CheckpointStrategy {
+
+    private var shouldCheckpoint = false
+
+    fun mark() {
+        shouldCheckpoint = true
+    }
+
+    override suspend fun tick(eventMessage: EventMessage, lastCheckpoint: StoredCheckpoint?): Boolean {
+        if (shouldCheckpoint) {
+            shouldCheckpoint = false
+            return true
+        }
+        return false
+    }
+
+}
+
+@Suppress("unused")
 class FixedTimeRateCheckpointStrategy(
-    private val rate: Duration
+    private val rate: Duration,
 ) : CheckpointStrategy {
 
     override suspend fun tick(eventMessage: EventMessage, lastCheckpoint: StoredCheckpoint?): Boolean {
@@ -39,4 +61,11 @@ class FixedTimeRateCheckpointStrategy(
         val duration = currentTime.minusMillis(lastCheckpoint.timestamp.toEpochMilli()).toEpochMilli()
         return duration >= rate.inWholeMilliseconds
     }
+}
+
+@Suppress("unused")
+object AlwaysCheckpointStrategy : CheckpointStrategy {
+    override suspend fun tick(eventMessage: EventMessage, lastCheckpoint: StoredCheckpoint?): Boolean {
+        return true
+    }
 }

krescent-core/src/main/kotlin/dev/helight/krescent/checkpoint/CheckpointStorage.kt

@@ -0,0 +1,22 @@
+package dev.helight.krescent.checkpoint
+
+import dev.helight.krescent.HandlerChainParticipant
+
+interface CheckpointSupport {
+    suspend fun createCheckpoint(bucket: CheckpointBucket)
+    suspend fun restoreCheckpoint(bucket: CheckpointBucket)
+
+    companion object {
+        suspend fun storagePass(participant: HandlerChainParticipant, bucket: CheckpointBucket) {
+            if (participant is CheckpointSupport) {
+                participant.createCheckpoint(bucket)
+            }
+        }
+
+        suspend fun restorePass(participant: HandlerChainParticipant, bucket: CheckpointBucket) {
+            if (participant is CheckpointSupport) {
+                participant.restoreCheckpoint(bucket)
+            }
+        }
+    }
+}

…escent/checkpoints/CheckpointStrategy.kt …rescent/checkpoint/CheckpointStrategy.ktkrescent-core/src/main/kotlin/dev/helight/krescent/checkpoints/CheckpointStrategy.kt renamed to krescent-core/src/main/kotlin/dev/helight/krescent/checkpoint/CheckpointStrategy.kt krescent-core/src/main/kotlin/dev/helight/krescent/checkpoints/CheckpointStrategy.kt renamed to krescent-core/src/main/kotlin/dev/helight/krescent/checkpoint/CheckpointStrategy.kt

@@ -1,21 +1,25 @@
-package dev.helight.krescent.checkpoints
+package dev.helight.krescent.checkpoint
 
-import dev.helight.krescent.EventMessage
-import dev.helight.krescent.EventMessageStreamProcessor
-import dev.helight.krescent.StreamingEventSource
-import dev.helight.krescent.StreamingToken
-import dev.helight.krescent.event.EventSourceConsumer
+import dev.helight.krescent.event.EventMessage
+import dev.helight.krescent.event.EventMessageStreamProcessor
+import dev.helight.krescent.event.SystemStreamHeadEvent
+import dev.helight.krescent.event.SystemStreamRestoredEvent
+import dev.helight.krescent.source.EventSourceConsumer
+import dev.helight.krescent.source.StreamingEventSource
+import dev.helight.krescent.source.StreamingToken
 import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.emptyFlow
 import kotlinx.serialization.json.jsonObject
 import java.time.Instant
 
-class CheckpointingEventSourceConsumer<T: StreamingToken<T>> (
+class CheckpointingEventSourceConsumer<T : StreamingToken<T>>(
     val namespace: String,
-    val revision: Long,
+    val revision: Int,
     val strategy: CheckpointStrategy,
     val source: StreamingEventSource<T>,
     val checkpointStorage: CheckpointStorage,
-    val consumer: EventMessageStreamProcessor
+    val additionalCheckpoints: List<CheckpointSupport>,
+    val consumer: EventMessageStreamProcessor,
 ) : EventSourceConsumer {
 
     override suspend fun stream() {
@@ -34,6 +38,12 @@ class CheckpointingEventSourceConsumer<T: StreamingToken<T>> (
         handlerLoop(lastCheckpoint, flow)
     }
 
+    override suspend fun restore() {
+        val lastCheckpoint = loadLastCheckpoint()
+        handlerLoop(lastCheckpoint, emptyFlow())
+
+    }
+
     private suspend fun loadLastCheckpoint(): StoredCheckpoint? {
         var lastCheckpoint = checkpointStorage.getLatestCheckpoint(namespace)
 
@@ -44,6 +54,9 @@ class CheckpointingEventSourceConsumer<T: StreamingToken<T>> (
 
         if (lastCheckpoint != null) {
             load(lastCheckpoint)
+            consumer.forwardSystemEvent(SystemStreamRestoredEvent())
+        } else {
+            consumer.forwardSystemEvent(SystemStreamHeadEvent())
         }
         return lastCheckpoint
     }
@@ -65,6 +78,7 @@ class CheckpointingEventSourceConsumer<T: StreamingToken<T>> (
     private suspend fun checkpoint(position: T): StoredCheckpoint {
         val bucket = CheckpointBucket(mutableMapOf())
         CheckpointSupport.storagePass(consumer, bucket)
+        additionalCheckpoints.forEach { it.createCheckpoint(bucket) }
         return StoredCheckpoint(
             namespace = namespace,
             revision = revision,
@@ -76,6 +90,7 @@ class CheckpointingEventSourceConsumer<T: StreamingToken<T>> (
 
     private suspend fun load(storedCheckpoint: StoredCheckpoint) {
         val bucket = CheckpointBucket(storedCheckpoint.data.jsonObject.toMutableMap())
+        additionalCheckpoints.forEach { it.restoreCheckpoint(bucket) }
         CheckpointSupport.restorePass(consumer, bucket)
     }
 

krescent-core/src/main/kotlin/dev/helight/krescent/checkpoint/CheckpointSupport.kt

@@ -0,0 +1,16 @@
+package dev.helight.krescent.checkpoint
+
+import dev.helight.krescent.serialization.InstantSerializer
+import kotlinx.serialization.Serializable
+import kotlinx.serialization.json.JsonElement
+import java.time.Instant
+
+@Serializable
+data class StoredCheckpoint(
+    val namespace: String,
+    val revision: Int,
+    val position: String,
+    @Serializable(with = InstantSerializer::class)
+    val timestamp: Instant,
+    val data: JsonElement,
+)