General overhaul of the new postgres integration to not be streaming directly anymore and added projection support

Author: helightdev

krescent-core/src/main/kotlin/dev/helight/krescent/source/PollingEventSource.kt

@@ -0,0 +1,46 @@
+package dev.helight.krescent.exposed
+
+import dev.helight.krescent.checkpoint.CheckpointStorage
+import dev.helight.krescent.checkpoint.StoredCheckpoint
+import org.jetbrains.exposed.sql.*
+import org.jetbrains.exposed.sql.SqlExpressionBuilder.eq
+
+class ExposedCheckpointStorage(
+    val database: Database,
+    val table: KrescentCheckpointTable = KrescentCheckpointTable(),
+) : CheckpointStorage {
+
+    suspend fun ensureCreated() {
+        table.create(database)
+    }
+
+    override suspend fun storeCheckpoint(checkpoint: StoredCheckpoint): Unit = jdbcSuspendTransaction(database) {
+        table.upsert(keys = arrayOf(table.namespace)) {
+            it[namespace] = checkpoint.namespace
+            it[position] = checkpoint.position
+            it[timestamp] = checkpoint.timestamp
+            it[version] = checkpoint.version
+            it[data] = checkpoint.data
+        }
+    }
+
+    override suspend fun getLatestCheckpoint(namespace: String): StoredCheckpoint? = jdbcSuspendTransaction(database) {
+        table.selectAll().where { table.namespace eq namespace }.firstOrNull()?.let {
+            StoredCheckpoint(
+                namespace = it[table.namespace],
+                position = it[table.position],
+                version = it[table.version],
+                timestamp = it[table.timestamp],
+                data = it[table.data]
+            )
+        }
+    }
+
+    override suspend fun deleteCheckpoint(namespace: String): Unit = jdbcSuspendTransaction(database) {
+        table.deleteWhere(1) { table.namespace eq namespace }
+    }
+
+    override suspend fun clearCheckpoints(): Unit = jdbcSuspendTransaction(database) {
+        table.deleteAll()
+    }
+}

krescent-exposed/src/main/kotlin/dev/helight/krescent/exposed/ExposedCheckpointStorage.kt

@@ -8,9 +8,9 @@ import java.util.*
 import kotlin.time.ExperimentalTime
 
 class ExposedEventPublisher(
-    val table: KrescentTable,
     val database: Database,
     val streamId: String,
+    val table: KrescentEventsTable = KrescentEventsTable(),
 ) : EventPublisher {
     @OptIn(ExperimentalTime::class)
     override suspend fun publish(event: EventMessage) {

krescent-exposed/src/main/kotlin/dev/helight/krescent/exposed/ExposedEventPublisher.kt

@@ -1,63 +1,64 @@
 package dev.helight.krescent.exposed
 
 import dev.helight.krescent.event.EventMessage
-import dev.helight.krescent.source.StreamingEventSource
+import dev.helight.krescent.source.StoredEventSource
 import dev.helight.krescent.source.StreamingToken
-import kotlinx.coroutines.coroutineScope
-import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.channelFlow
-import kotlinx.coroutines.flow.emptyFlow
-import kotlinx.coroutines.flow.flow
-import org.jetbrains.exposed.sql.Database
-import org.jetbrains.exposed.sql.Query
-import org.jetbrains.exposed.sql.ResultRow
-import org.jetbrains.exposed.sql.SortOrder
+import org.jetbrains.exposed.sql.*
+import org.jetbrains.exposed.sql.SqlExpressionBuilder.eq
+import org.jetbrains.exposed.sql.SqlExpressionBuilder.inList
+import org.jetbrains.exposed.sql.SqlExpressionBuilder.like
+import org.jetbrains.exposed.sql.SqlExpressionBuilder.regexp
 import kotlin.math.min
 import kotlin.time.ExperimentalTime
 
 @OptIn(ExperimentalTime::class)
-class StreamingExposedEventSource(
-    val table: KrescentTable,
+class ExposedEventSource(
     val database: Database,
+    val table: KrescentEventsTable = KrescentEventsTable(),
     val streamId: String? = null,
     val streamIdMatcher: StreamIdMatcher = StreamIdMatcher.EQ,
+    val eventFilter: StreamEventFilter? = null,
     val batchSize: Int = 20,
-    val pollingDelay: Long = 500L,
-) : StreamingEventSource {
+) : StoredEventSource {
     override suspend fun getHeadToken(): StreamingToken<*> {
         return ExposedStreamingToken.HeadToken()
     }
 
-    override suspend fun getTailToken(): StreamingToken<*> {
-        return ExposedStreamingToken.TailToken()
-    }
+    override suspend fun getTailToken(): StreamingToken<*> = peakEnd()
 
     override suspend fun deserializeToken(encoded: String): StreamingToken<*> {
         return when (encoded) {
             "HEAD" -> ExposedStreamingToken.HeadToken()
-            "TAIL" -> ExposedStreamingToken.TailToken()
             else -> ExposedStreamingToken.PositionToken(encoded.toLong())
         }
     }
 
-    private fun Query.withStreamIdFilter(): Query {
-        return when (streamId) {
-            null -> this
+    private fun Query.withFilterClause(): Query {
+        when (streamId) {
+            null -> null
             else -> when (streamIdMatcher) {
-                StreamIdMatcher.EQ -> this.where { table.streamId eq streamId }
-                StreamIdMatcher.LIKE -> this.where { table.streamId like streamId }
-                StreamIdMatcher.REGEX -> this.where { table.streamId regexp streamId }
+                StreamIdMatcher.EQ -> table.streamId eq streamId
+                StreamIdMatcher.LIKE -> table.streamId like streamId
+                StreamIdMatcher.REGEX -> table.streamId regexp streamId
             }
-        }
+        }?.let { andWhere { it } }
+
+        when (eventFilter) {
+            null -> null
+            else -> table.type inList eventFilter.eventNames
+        }?.let { andWhere { it } }
+
+        return this
     }
 
     private suspend fun peakEnd(): ExposedStreamingToken {
         return jdbcSuspendTransaction(database) {
             val last = table
                 .select(table.id)
                 .orderBy(table.id, SortOrder.DESC)
-                .withStreamIdFilter()
+                .withFilterClause()
                 .limit(1)
                 .firstOrNull()
             when (last) {
@@ -85,17 +86,25 @@ class StreamingExposedEventSource(
         }
         val list = jdbcSuspendTransaction(database) {
             token.begin(table)
-                .withStreamIdFilter()
+                .withFilterClause()
                 .limit(actualBatchSize)
                 .map(::mapRowToPair)
                 .toList()
         }
-        val endToken = list.lastOrNull()?.second ?: peakEnd()
-        return BatchResult(
-            events = list,
-            endToken = endToken,
-            reachedEnd = list.size < actualBatchSize || endToken is ExposedStreamingToken.TailToken,
-        )
+        val endToken = list.lastOrNull()?.second
+        return if (endToken == null) {
+            BatchResult(
+                events = list,
+                endToken = peakEnd(),
+                reachedEnd = true
+            )
+        } else {
+            BatchResult(
+                events = list,
+                endToken = endToken,
+                reachedEnd = list.size < actualBatchSize
+            )
+        }
     }
 
     override suspend fun fetchEventsAfter(
@@ -107,7 +116,6 @@ class StreamingExposedEventSource(
             is ExposedStreamingToken -> token
             else -> throw IllegalArgumentException("Token must be of type ExposedStreamingToken")
         }
-        if (parsedToken is ExposedStreamingToken.TailToken) return emptyFlow()
         var currentToken = parsedToken
         var remaining = limit?.toLong() ?: Long.MAX_VALUE
         return channelFlow {
@@ -123,29 +131,6 @@ class StreamingExposedEventSource(
         }
     }
 
-    override suspend fun streamEvents(startToken: StreamingToken<*>?): Flow<Pair<EventMessage, StreamingToken<*>>> =
-        coroutineScope {
-            val parsedToken = when (startToken) {
-                null -> ExposedStreamingToken.HeadToken()
-                is ExposedStreamingToken.TailToken -> peakEnd()
-                is ExposedStreamingToken -> startToken
-                else -> throw IllegalArgumentException("Token must be of type ExposedStreamingToken")
-            }
-            var currentToken: ExposedStreamingToken = parsedToken
-            return@coroutineScope flow {
-                while (true) {
-                    val batch = fetchBatch(currentToken)
-                    for (event in batch.events) {
-                        emit(event)
-                    }
-                    currentToken = batch.endToken
-                    if (batch.reachedEnd) {
-                        delay(pollingDelay)
-                    }
-                }
-            }
-        }
-
     private data class BatchResult(
         val events: List<Pair<EventMessage, StreamingToken<*>>>,
         val endToken: ExposedStreamingToken,

…t/exposed/StreamingExposedEventSource.kt …t/krescent/exposed/ExposedEventSource.ktkrescent-exposed/src/main/kotlin/dev/helight/krescent/exposed/StreamingExposedEventSource.kt renamed to krescent-exposed/src/main/kotlin/dev/helight/krescent/exposed/ExposedEventSource.kt krescent-exposed/src/main/kotlin/dev/helight/krescent/exposed/StreamingExposedEventSource.kt renamed to krescent-exposed/src/main/kotlin/dev/helight/krescent/exposed/ExposedEventSource.kt

@@ -2,6 +2,7 @@ package dev.helight.krescent.exposed
 
 import dev.helight.krescent.source.StreamingToken
 import org.jetbrains.exposed.sql.Query
+import org.jetbrains.exposed.sql.andWhere
 import org.jetbrains.exposed.sql.selectAll
 
 sealed class ExposedStreamingToken : StreamingToken<ExposedStreamingToken> {
@@ -10,7 +11,6 @@ sealed class ExposedStreamingToken : StreamingToken<ExposedStreamingToken> {
         // Doesn't follow the kurrentdb token format, since our interpretation of head and tail is different
         return when (this) {
             is HeadToken -> "HEAD"
-            is TailToken -> "TAIL"
             is PositionToken -> pos.toString()
         }
     }
@@ -19,34 +19,26 @@ sealed class ExposedStreamingToken : StreamingToken<ExposedStreamingToken> {
         return when (this) {
             is HeadToken -> when (other) {
                 is HeadToken -> 0
-                is TailToken -> -1
                 is PositionToken -> -1
             }
 
-            is TailToken -> when (other) {
-                is HeadToken -> 1
-                is TailToken -> 0
-                is PositionToken -> 1
-            }
-
             is PositionToken -> when (other) {
                 is HeadToken -> 1
-                is TailToken -> -1
                 is PositionToken -> pos.compareTo(other.pos)
             }
         }
     }
 
     class HeadToken : ExposedStreamingToken()
-    class TailToken : ExposedStreamingToken()
     data class PositionToken(val pos: Long) : ExposedStreamingToken()
 
-    fun begin(table: KrescentTable): Query {
+    fun begin(table: KrescentEventsTable): Query {
         val query = table.selectAll()
         return when (this) {
             is HeadToken -> query.orderBy(table.id)
-            is TailToken -> throw IllegalArgumentException("Cannot apply TailToken to a query")
-            is PositionToken -> query.orderBy(table.id).where { table.id greater pos }
+            is PositionToken -> {
+                query.orderBy(table.id).andWhere { table.id greater pos }
+            }
         }
     }
 }