kami/correx
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(server-02): promote EventStore to global-sequence store with subscribeAll

Browse Source 
StoredEvent gains sessionSequence: Long (per-session monotonic); existing
sequence field becomes the global monotonic cursor across the entire store.
SqliteEventStore allocates both counters inside the same BEGIN IMMEDIATE
transaction and emits to a global MutableSharedFlow after commit.
InMemoryEventStore mirrors this with an AtomicLong global counter and a
suspend-on-overflow SharedFlow. EventStore interface gains subscribeAll()
and lastGlobalSequence(). Contract tests updated and extended to cover
global-sequence and cross-session subscribeAll invariants.
This commit is contained in:
kami
2026-05-24 22:23:43 +04:00
parent 0cfb784187
commit 70420083ac
8 changed files with 184 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files
infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/InMemoryEventStore.kt
+38 -23
View File
@@ -5,59 +5,71 @@ import com.correx.core.events.events.StoredEvent
import com.correx.core.events.stores.EventStore
import com.correx.core.events.types.EventId
import com.correx.core.events.types.SessionId
import kotlinx.coroutines.channels.BufferOverflow
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.asSharedFlow
import java.util.concurrent.*
import java.util.concurrent.atomic.*
class InMemoryEventStore : EventStore {
private val streams = ConcurrentHashMap<SessionId, MutableList<StoredEvent>>()
private val sequences = ConcurrentHashMap<SessionId, AtomicLong>()
private val globalSeq = AtomicLong(0)
private val seenEventIds = ConcurrentHashMap.newKeySet<EventId>()
private val subscriptions = ConcurrentHashMap<SessionId, MutableSharedFlow<StoredEvent>>()
private val globalFlow = MutableSharedFlow<StoredEvent>(
replay = 0,
extraBufferCapacity = 1024,
onBufferOverflow = BufferOverflow.SUSPEND,
)
override suspend fun append(event: NewEvent): StoredEvent {
val stream = streams.computeIfAbsent(event.metadata.sessionId) { mutableListOf() }
val stored: StoredEvent
synchronized(stream) {
if (!seenEventIds.add(event.metadata.eventId)) {
error("duplicate event_id: ${event.metadata.eventId}")
}
return doAppend(event, stream)
stored = doAppend(event, stream)
}
subscriptions[event.metadata.sessionId]?.tryEmit(stored)
globalFlow.emit(stored)
return stored
}
override suspend fun appendAll(events: List<NewEvent>): List<StoredEvent> {
if (events.isEmpty()) return emptyList()
val sessionId = events.first().metadata.sessionId
val stream = streams.computeIfAbsent(sessionId) { mutableListOf() }
val stored: List<StoredEvent>
synchronized(stream) {
return events.map { doAppend(it, stream) }
stored = events.map { doAppend(it, stream) }
}
val flow = subscriptions[sessionId]
if (flow != null) stored.forEach { flow.tryEmit(it) }
stored.forEach { globalFlow.emit(it) }
return stored
}
override fun read(sessionId: SessionId): List<StoredEvent> {
return streams[sessionId]
?.sortedBy { it.sequence }
?: emptyList()
}
override fun read(sessionId: SessionId): List<StoredEvent> =
streams[sessionId]?.sortedBy { it.sessionSequence } ?: emptyList()
override fun readFrom(sessionId: SessionId, fromSequence: Long): List<StoredEvent> {
return streams[sessionId]
?.filter { it.sequence > fromSequence }
override fun readFrom(sessionId: SessionId, fromSequence: Long): List<StoredEvent> =
streams[sessionId]
?.filter { it.sessionSequence > fromSequence }
?.toList()
?: emptyList()
}
override fun lastSequence(sessionId: SessionId): Long? {
return sequences[sessionId]?.get()
}
override fun lastSequence(sessionId: SessionId): Long? = sequences[sessionId]?.get()
override fun subscribe(sessionId: SessionId): Flow<StoredEvent> =
subscriptions.computeIfAbsent(sessionId) { MutableSharedFlow(replay = 0, extraBufferCapacity = 64) }
override fun subscribeAll(): Flow<StoredEvent> = globalFlow.asSharedFlow()
override suspend fun lastGlobalSequence(): Long = globalSeq.get()
override fun allEvents(): Sequence<StoredEvent> =
streams.values
.flatMap { stream -> synchronized(stream) { stream.toList() } }
@@ -66,17 +78,20 @@ class InMemoryEventStore : EventStore {
override fun allSessionIds(): Set<SessionId> = sequences.keys
private fun doAppend(event: NewEvent, stream: MutableList<StoredEvent>): StoredEvent {
val seq = sequences.computeIfAbsent(event.metadata.sessionId) { AtomicLong(0) }
val sessionSeq = sequences.computeIfAbsent(event.metadata.sessionId) { AtomicLong(0) }
.incrementAndGet()
val globalSeqVal = globalSeq.incrementAndGet()
val stored = StoredEvent(metadata = event.metadata, sequence = seq, payload = event.payload)
// safety: enforce ordering invariant
if (stream.isNotEmpty() && stream.last().sequence >= stored.sequence) {
val stored = StoredEvent(
metadata = event.metadata,
sequence = globalSeqVal,
sessionSequence = sessionSeq,
payload = event.payload,
)
if (stream.isNotEmpty() && stream.last().sessionSequence >= stored.sessionSequence) {
error("sequence violation for session ${event.metadata.sessionId}")
}
stream.add(stored)
subscriptions[event.metadata.sessionId]?.tryEmit(stored)
return stored
}
}
infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/SqliteEventStore.kt
+63 -38
View File
@@ -13,8 +13,10 @@ import com.correx.core.events.types.EventId
import com.correx.core.events.types.SessionId
import com.correx.infrastructure.persistence.util.JDBCHelper.transaction
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.channels.BufferOverflow
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.asSharedFlow
import kotlinx.coroutines.withContext
import kotlinx.datetime.Instant
import java.sql.Connection
@@ -27,6 +29,11 @@ class SqliteEventStore(
private val artifactStore: ArtifactStore,
) : EventStore {
private val subscriptions = ConcurrentHashMap<SessionId, MutableSharedFlow<StoredEvent>>()
private val globalFlow = MutableSharedFlow<StoredEvent>(
replay = 0,
extraBufferCapacity = 1024,
onBufferOverflow = BufferOverflow.SUSPEND,
)
init {
connection.createStatement().use { stmt ->
@@ -36,6 +43,7 @@ class SqliteEventStore(
event_id TEXT PRIMARY KEY,
session_id TEXT NOT NULL,
sequence INTEGER NOT NULL,
session_sequence INTEGER NOT NULL,
timestamp TEXT NOT NULL,
schema_version INTEGER NOT NULL,
causation_id TEXT,
@@ -44,6 +52,12 @@ class SqliteEventStore(
);
""".trimIndent(),
)
stmt.execute(
"CREATE UNIQUE INDEX IF NOT EXISTS events_sequence_uq ON events(sequence);"
)
stmt.execute(
"CREATE UNIQUE INDEX IF NOT EXISTS events_session_seq_uq ON events(session_id, session_sequence);"
)
}
}
@@ -54,15 +68,14 @@ class SqliteEventStore(
stored = connection.transaction {
val existing = findByEventId(event.metadata.eventId)
check(existing == null) { "duplicate event_id: ${event.metadata.eventId}" }
val seq = nextSequence(event.metadata.sessionId)
val globalSeqVal = nextGlobalSequence()
val sessionSeqVal = nextSessionSequence(event.metadata.sessionId)
val s = StoredEvent(
metadata = event.metadata,
sequence = seq,
sequence = globalSeqVal,
sessionSequence = sessionSeqVal,
payload = event.payload,
)
insert(s)
s
}
@@ -70,14 +83,13 @@ class SqliteEventStore(
}
val result = checkNotNull(stored)
subscriptions[event.metadata.sessionId]?.tryEmit(result)
globalFlow.emit(result)
return result
}
override suspend fun appendAll(events: List<NewEvent>): List<StoredEvent> {
if (events.isEmpty()) return emptyList()
val sessionId = events.first().metadata.sessionId
var stored: List<StoredEvent> = emptyList()
artifactStore.flushBefore {
withContext(Dispatchers.IO) {
@@ -85,15 +97,14 @@ class SqliteEventStore(
events.map { event ->
val existing = findByEventId(event.metadata.eventId)
check(existing == null) { "duplicate event_id: ${event.metadata.eventId}" }
val seq = nextSequence(sessionId)
val globalSeqVal = nextGlobalSequence()
val sessionSeqVal = nextSessionSequence(sessionId)
val s = StoredEvent(
metadata = event.metadata,
sequence = seq,
sequence = globalSeqVal,
sessionSequence = sessionSeqVal,
payload = event.payload,
)
insert(s)
s
}
@@ -102,6 +113,7 @@ class SqliteEventStore(
}
val flow = subscriptions[sessionId]
if (flow != null) stored.forEach { flow.tryEmit(it) }
stored.forEach { globalFlow.emit(it) }
return stored
}
@@ -110,11 +122,10 @@ class SqliteEventStore(
"""
SELECT * FROM events
WHERE session_id = ?
ORDER BY sequence ASC
ORDER BY session_sequence ASC
""".trimIndent(),
).use { ps ->
ps.setString(1, sessionId.value)
ps.executeQuery().use { rs ->
buildList {
while (rs.next()) {
@@ -129,13 +140,12 @@ class SqliteEventStore(
"""
SELECT * FROM events
WHERE session_id = ?
AND sequence > ?
ORDER BY sequence ASC
AND session_sequence > ?
ORDER BY session_sequence ASC
""".trimIndent(),
).use { ps ->
ps.setString(1, sessionId.value)
ps.setLong(2, fromSequence)
ps.executeQuery().use { rs ->
buildList {
while (rs.next()) {
@@ -147,14 +157,9 @@ class SqliteEventStore(
override fun lastSequence(sessionId: SessionId): Long? =
connection.prepareStatement(
"""
SELECT MAX(sequence)
FROM events
WHERE session_id = ?
""".trimIndent(),
"SELECT MAX(session_sequence) FROM events WHERE session_id = ?"
).use { ps ->
ps.setString(1, sessionId.value)
ps.executeQuery().use { rs ->
if (rs.next()) rs.getLong(1).takeIf { !rs.wasNull() }
else null
@@ -164,6 +169,20 @@ class SqliteEventStore(
override fun subscribe(sessionId: SessionId): Flow<StoredEvent> =
subscriptions.computeIfAbsent(sessionId) { MutableSharedFlow(replay = 0, extraBufferCapacity = 64) }
override fun subscribeAll(): Flow<StoredEvent> = globalFlow.asSharedFlow()
override suspend fun lastGlobalSequence(): Long =
withContext(Dispatchers.IO) {
connection.prepareStatement(
"SELECT COALESCE(MAX(sequence), 0) FROM events"
).use { ps ->
ps.executeQuery().use { rs ->
rs.next()
rs.getLong(1)
}
}
}
override fun allEvents(): Sequence<StoredEvent> =
connection.prepareStatement(
"""
@@ -191,13 +210,19 @@ class SqliteEventStore(
// ---------- helpers ----------
private fun nextSequence(sessionId: SessionId): Long =
private fun nextGlobalSequence(): Long =
connection.prepareStatement(
"""
SELECT COALESCE(MAX(sequence), 0)
FROM events
WHERE session_id = ?
""".trimIndent(),
"SELECT COALESCE(MAX(sequence), 0) FROM events"
).use { ps ->
ps.executeQuery().use { rs ->
rs.next()
rs.getLong(1) + 1
}
}
private fun nextSessionSequence(sessionId: SessionId): Long =
connection.prepareStatement(
"SELECT COALESCE(MAX(session_sequence), 0) FROM events WHERE session_id = ?"
).use { ps ->
ps.setString(1, sessionId.value)
ps.executeQuery().use { rs ->
@@ -214,23 +239,24 @@ class SqliteEventStore(
event_id,
session_id,
sequence,
session_sequence,
timestamp,
schema_version,
causation_id,
correlation_id,
payload
) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""".trimIndent(),
).use { ps ->
ps.setString(1, event.metadata.eventId.value)
ps.setString(2, event.metadata.sessionId.value)
ps.setLong(3, event.sequence)
ps.setString(4, event.metadata.timestamp.toString())
ps.setInt(5, event.metadata.schemaVersion)
ps.setString(6, event.metadata.causationId?.value)
ps.setString(7, event.metadata.correlationId?.value)
ps.setString(8, jsonSerializer.serialize(event.payload))
ps.setLong(4, event.sessionSequence)
ps.setString(5, event.metadata.timestamp.toString())
ps.setInt(6, event.metadata.schemaVersion)
ps.setString(7, event.metadata.causationId?.value)
ps.setString(8, event.metadata.correlationId?.value)
ps.setString(9, jsonSerializer.serialize(event.payload))
ps.executeUpdate()
}
}
@@ -243,12 +269,10 @@ class SqliteEventStore(
""".trimIndent(),
).use { ps ->
ps.setString(1, eventId.value)
ps.executeQuery().use { rs ->
if (rs.next()) rs.toStoredEvent(jsonSerializer) else null
}
}
}
private fun ResultSet.toStoredEvent(jsonSerializer: JsonEventSerializer): StoredEvent =
@@ -262,5 +286,6 @@ private fun ResultSet.toStoredEvent(jsonSerializer: JsonEventSerializer): Stored
correlationId = getString("correlation_id")?.let { CorrelationId(it) },
),
sequence = getLong("sequence"),
sessionSequence = getLong("session_sequence"),
payload = jsonSerializer.deserialize(getString("payload")),
)