Add Support for Postgres and modernize gradle setup

Author: helightdev

.idea/copilot.data.migration.agent.xml

@@ -1,16 +1,14 @@
-# Version catalog is a central place for you to declare and version dependencies
-# https://docs.gradle.org/current/userguide/platforms.html#sub:version-catalog
-# https://docs.gradle.org/current/userguide/platforms.html#sub::toml-dependencies-format
-
 [versions]
 kotlin = "2.1.20"
 kotlinxDatetime = "0.6.1"
 kotlinxSerialization = "1.7.3"
-kotlinxCoroutines = "1.9.0"
+kotlinxCoroutines = "1.10.2"
 kurrentClient = "1.0.1"
 orgReactiveStream = "1.0.4"
 mongoDriver = "5.5.0"
 testContainers = "1.21.0"
+exposed = "1.0.0-rc-1"
+postgresql = "42.7.3"
 
 [libraries]
 kotlinGradlePlugin = { module = "org.jetbrains.kotlin:kotlin-gradle-plugin", version.ref = "kotlin" }
@@ -24,12 +22,19 @@ mongoDriver = { module = "org.mongodb:mongodb-driver-kotlin-coroutine", version.
 mongoBsonX = { module = "org.mongodb:bson-kotlinx", version.ref = "mongoDriver" }
 testContainersJunit = { module = "org.testcontainers:junit-jupiter", version.ref = "testContainers" }
 testContainers = { module = "org.testcontainers:testcontainers", version.ref = "testContainers" }
+exposedCore = { module = "org.jetbrains.exposed:exposed-core", version.ref = "exposed" }
+exposedDao = { module = "org.jetbrains.exposed:exposed-dao", version.ref = "exposed" }
+exposedJdbc = { module = "org.jetbrains.exposed:exposed-jdbc", version.ref = "exposed" }
+exposedJson = { module = "org.jetbrains.exposed:exposed-json", version.ref = "exposed" }
+exposedKotlinDatetime = { module = "org.jetbrains.exposed:exposed-kotlin-datetime", version.ref = "exposed" }
+postgresql = { module = "org.postgresql:postgresql", version.ref = "postgresql" }
 
 # Libraries can be bundled together for easier import
 [bundles]
 kotlinxEcosystem = ["kotlinxDatetime", "kotlinxSerializationJson", "kotlinxSerializationCbor", "kotlinxCoroutines"]
 kurrent = ["kurrentClient", "orgReactiveStream"]
 mongo = ["mongoDriver", "mongoBsonX"]
+exposed = ["exposedCore", "exposedDao", "exposedJdbc", "exposedJson", "exposedKotlinDatetime"]
 testContainers = ["testContainersJunit", "testContainers"]
 
 [plugins]

.idea/copilot.data.migration.edit.xml

@@ -67,12 +67,14 @@ class FlowBuffer<T> {
  * @param first The first flow to collect
  * @param second The second flow to collect
  * @param activateCallback A callback to be called when both flows are collecting values
+ * @param liveCallback A callback to be called when the flow switches to live mode
  */
 @OptIn(ExperimentalAtomicApi::class)
 fun <T> joinSequentialFlows(
     first: Flow<T>,
     second: Flow<T>,
     activateCallback: () -> Unit = { },
+    liveCallback: () -> Unit = { },
 ): Flow<T> = channelFlow {
     val mutex = Mutex()
     val secondBuffer = ArrayDeque<T>()
@@ -101,10 +103,70 @@ fun <T> joinSequentialFlows(
             send(secondBuffer.removeFirst())
         }
         switchToLive = true
+        liveCallback()
     }
     secondJob.join()
 }
 
+/**
+ * Creates a backfilling flow with a live tail from two flows.
+ * The `live` flow will be buffered until the `catchup` flow is fully collected.
+ * Once the `catchup` flow is complete, the buffered values from the `live` flow will be sent,
+ * followed by any new values from the `live` flow.
+ *
+ * @param catchup The flow to backfill from
+ * @param live The live flow to collect from
+ * @param comparator A comparator used for deduplication of values by age or position
+ * @param activateCallback A callback to be called when both flows are collecting values
+ * @param liveCallback A callback to be called when the flow switches to live mode
+ */
+fun <T> createCatchupFlow(
+    catchup: Flow<T>,
+    live: Flow<T>,
+    comparator: Comparator<T>,
+    activateCallback: () -> Unit = { },
+    liveCallback: () -> Unit = { },
+): Flow<T> = channelFlow {
+    val mutex = Mutex()
+    val liveBuffer = mutableListOf<T>()
+    var lastCatchup: T? = null
+    var switchToLive = false
+
+    val catchupJob = launch {
+        catchup.collect { value ->
+            send(value)
+            lastCatchup = value
+        }
+    }
+    val liveJob = launch {
+        live.collect { value ->
+            mutex.withLock {
+                if (switchToLive) {
+                    send(value)
+                } else {
+                    liveBuffer.add(value)
+                }
+            }
+        }
+    }
+    activateCallback()
+    catchupJob.join()
+    mutex.withLock {
+        if (liveBuffer.isNotEmpty()) {
+            if (lastCatchup != null) {
+                liveBuffer.removeAll { comparator.compare(it, lastCatchup) <= 0 }
+            }
+        }
+        for (value in liveBuffer) {
+            send(value)
+        }
+        liveBuffer.clear()
+        switchToLive = true
+        liveCallback()
+    }
+    liveJob.join()
+}
+
 /**
  * Runs a sequence of tasks where each task is executed in order without interruption due to exceptions.
  * If any task throws an exception, it collects all exceptions and throws a single `UninterruptedChainException` with a list of errors.

.idea/dataSources.xml

@@ -0,0 +1,44 @@
+package dev.helight.krescent.source
+
+import dev.helight.krescent.event.EventMessage
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.flow
+
+class PollingEventSource<T : StoredEventSource>(
+    val delegate: T,
+    val pollingLimit: Int? = null,
+    val pollingDelay: Long = 500L,
+) : StreamingEventSource, StoredEventSource by delegate {
+
+    override suspend fun streamEvents(startToken: StreamingToken<*>?): Flow<Pair<EventMessage, StreamingToken<*>>> =
+        coroutineScope {
+            var currentToken: StreamingToken<*> = startToken ?: delegate.getHeadToken()
+            return@coroutineScope flow {
+                while (true) {
+                    val subFlow = fetchEventsAfter(currentToken, pollingLimit)
+                    var counter = 0
+                    subFlow.collect {
+                        emit(it)
+                        currentToken = it.second
+                        counter++
+                    }
+                    val reachedEnd = counter < (pollingLimit ?: Int.MAX_VALUE)
+                    if (reachedEnd) {
+                        delay(pollingDelay)
+                    }
+                }
+            }
+        }
+
+}
+
+fun <T : StoredEventSource> T.asPollingSource(
+    pollingLimit: Int? = null,
+    pollingDelay: Long = 500L,
+): PollingEventSource<T> = PollingEventSource(
+    delegate = this,
+    pollingLimit = pollingLimit,
+    pollingDelay = pollingDelay,
+)

.idea/gradle.xml

@@ -5,11 +5,10 @@ import kotlinx.coroutines.flow.Flow
 import kotlinx.datetime.Instant
 
 /**
- * Interface for an event source that provides a replayable stream of events.
+ * Interface for an event source that provides access to stored events.
  * Supports token-based positioning for fetching events.
  */
-interface StreamingEventSource {
-
+interface StoredEventSource {
     /**
      * Gets a token for the head of the stream, which is *before* the first event in the stream.
      * Fetching after the head token will return the first event in the stream.
@@ -40,8 +39,18 @@ interface StreamingEventSource {
      * @param limit Maximum number of events to fetch, null for no limit
      * @return A pair of the flow of events and the updated token for the next fetch
      */
-    suspend fun fetchEventsAfter(token: StreamingToken<*>? = null, limit: Int? = null): Flow<Pair<EventMessage, StreamingToken<*>>>
+    suspend fun fetchEventsAfter(
+        token: StreamingToken<*>? = null,
+        limit: Int? = null,
+    ): Flow<Pair<EventMessage, StreamingToken<*>>>
 
+}
+
+/**
+ * Interface for an event source that provides a replayable stream of events.
+ * Supports token-based positioning for fetching events.
+ */
+interface StreamingEventSource : StoredEventSource {
     /**
      * Creates a flow of all events in the stream starting from the specified token.
      * Then it will continue to stream new events as they are published.
@@ -52,7 +61,7 @@ interface StreamingEventSource {
     suspend fun streamEvents(startToken: StreamingToken<*>? = null): Flow<Pair<EventMessage, StreamingToken<*>>>
 }
 
-interface ExtendedQueryableStreamingEventSource : StreamingEventSource {
+interface ExtendedQueryableStoredEventSource : StoredEventSource {
     /**
      * Gets a token pointing before the first event at or after the specified timestamp.
      *
@@ -67,4 +76,5 @@ interface ExtendedQueryableStreamingEventSource : StreamingEventSource {
      */
     suspend fun getTokenForEventId(eventId: String): StreamingToken<*>?
 
-}
+}
+

.idea/sqldialects.xml

@@ -2,10 +2,7 @@ package dev.helight.krescent.source.impl
 
 import dev.helight.krescent.event.EventMessage
 import dev.helight.krescent.joinSequentialFlows
-import dev.helight.krescent.source.EventPublisher
-import dev.helight.krescent.source.ExtendedQueryableStreamingEventSource
-import dev.helight.krescent.source.StreamingToken
-import dev.helight.krescent.source.SubscribingEventSource
+import dev.helight.krescent.source.*
 import kotlinx.coroutines.flow.*
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
@@ -17,7 +14,7 @@ import java.nio.charset.StandardCharsets
 
 class InMemoryEventStore(
     private val events: MutableList<EventMessage> = mutableListOf(),
-) : ExtendedQueryableStreamingEventSource, SubscribingEventSource, EventPublisher {
+) : StreamingEventSource, ExtendedQueryableStoredEventSource, SubscribingEventSource, EventPublisher {
 
     constructor(vararg events: EventMessage) : this(events.toMutableList())