Add payload filtering to ExposedEventSource and introduce StreamPayloadFilter class.

Also converter json to jsonb field and make it indexed

Author: helightdev

.idea/gradle.xml

@@ -7,9 +7,9 @@ import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.channelFlow
 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 org.jetbrains.exposed.sql.json.contains
 import kotlin.math.min
 import kotlin.time.ExperimentalTime
 
@@ -20,6 +20,7 @@ class ExposedEventSource(
     val table: KrescentEventLogTable = KrescentEventLogTable(),
     val streamIdMatcher: StreamIdMatcher = StreamIdMatcher.EQ,
     val eventFilter: StreamEventFilter? = null,
+    val payloadFilter: StreamPayloadFilter? = null,
     val batchSize: Int = 500,
 ) : StoredEventSource {
     override suspend fun getHeadToken(): StreamingToken<*> {
@@ -45,10 +46,8 @@ class ExposedEventSource(
             }
         }?.let { andWhere { it } }
 
-        when (eventFilter) {
-            null -> null
-            else -> table.type inList eventFilter.eventNames
-        }?.let { andWhere { it } }
+        if (eventFilter != null) andWhere { table.type inList eventFilter.eventNames }
+        if (payloadFilter != null) andWhere { table.data.contains(payloadFilter.toQuery()) }
 
         return this
     }

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

@@ -9,6 +9,7 @@ import org.jetbrains.exposed.sql.Database
 import org.jetbrains.exposed.sql.SchemaUtils
 import org.jetbrains.exposed.sql.Table
 import org.jetbrains.exposed.sql.json.json
+import org.jetbrains.exposed.sql.json.jsonb
 import org.jetbrains.exposed.sql.kotlin.datetime.KotlinInstantColumnType
 import org.jetbrains.exposed.sql.upsert
 import kotlin.time.ExperimentalTime
@@ -20,11 +21,11 @@ class KrescentEventLogTable(tableName: String = "krescent") : LongIdTable(tableN
     val type = text("type").index()
     val timestamp = registerColumn("timestamp", KotlinInstantColumnType()).index()
 
-    val data = json("data", {
+    val data = jsonb("data", {
         Json.encodeToString(it)
     }, {
         Json.decodeFromString(JsonElement.serializer(), it)
-    })
+    }).index()
 
     suspend fun create(database: Database) {
         jdbcSuspendTransaction(database) { SchemaUtils.create(this@KrescentEventLogTable) }

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

@@ -27,5 +27,5 @@ class StreamEventFilter(
             return StreamEventFilter(includes)
         }
     }
+}
 
-}

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

@@ -0,0 +1,17 @@
+package dev.helight.krescent.exposed
+
+import kotlinx.serialization.json.Json
+import kotlinx.serialization.json.JsonElement
+import kotlinx.serialization.json.JsonPrimitive
+
+class StreamPayloadFilter(
+    val values: Map<String, JsonElement>,
+) {
+    fun toQuery(): String = Json.encodeToString(values)
+
+    companion object {
+        fun field(key: String, value: String) = StreamPayloadFilter(mapOf(key to JsonPrimitive(value)))
+        fun field(key: String, value: Number) = StreamPayloadFilter(mapOf(key to JsonPrimitive(value)))
+        fun field(key: String, value: Boolean) = StreamPayloadFilter(mapOf(key to JsonPrimitive(value)))
+    }
+}

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

@@ -127,6 +127,21 @@ class PostgresStreamingSource : StreamingEventSourceContract {
         assertEquals(3, listReceived.count())
     }
 
+    @Test
+    fun `Simple event payload filter for string`() = execWithTable { db, table ->
+        val publisher = ExposedEventPublisher(db, "event-stream", table)
+        publisher.publish(exampleCatalog.create(EventA(1)))
+        publisher.publish(exampleCatalog.create(EventB(2)))
+        publisher.publish(exampleCatalog.create(EventC(2)))
+        publisher.publish(exampleCatalog.create(EventC(3)))
+
+        val payloadFilterReceived = ExposedEventSource(
+            db, "event-stream", table,
+            payloadFilter = StreamPayloadFilter.field("num", 2)
+        ).fetchEventsAfter().toList()
+        assertEquals(2, payloadFilterReceived.count())
+    }
+
     val exampleCatalog = buildEventCatalog(1) {
         event<EventA>("main.a")
         event<EventB>("main.b")