perf(talkie): fold idea board once, keep live via subscribeAll (#55)

IdeaReader folded eventStore.allEvents() on every activeIdeas()/capturedOf()
call — a full deserializing log scan per CHAT context build. Now folds once at
construction and stays current incrementally via subscribeAll(); the fold is
idempotent so seed/live overlap needs no dedupe.
This commit is contained in:
2026-07-12 13:12:35 +04:00
parent 1fd878eb9b
commit da7fe61b77
@@ -1,46 +1,61 @@
package com.correx.core.talkie package com.correx.core.talkie
import com.correx.core.events.events.EventPayload
import com.correx.core.events.events.IdeaCapturedEvent import com.correx.core.events.events.IdeaCapturedEvent
import com.correx.core.events.events.IdeaDiscardedEvent import com.correx.core.events.events.IdeaDiscardedEvent
import com.correx.core.events.events.IdeaPromotedEvent import com.correx.core.events.events.IdeaPromotedEvent
import com.correx.core.events.stores.EventStore import com.correx.core.events.stores.EventStore
import com.correx.core.events.types.SessionId import com.correx.core.events.types.SessionId
import com.correx.core.talkie.model.Idea import com.correx.core.talkie.model.Idea
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.launch
import java.util.concurrent.ConcurrentHashMap
/** /**
* Rebuilds the operator's idea board from the event log. Cross-session by design: it folds over * Holds the operator's idea board in memory. Cross-session by design: it folds over the whole event
* [EventStore.allEvents] rather than a single session's replay, so an idea captured in one session * log (not a single session's replay), so an idea captured in one session shows on the board (and
* shows on the board (and feeds the router) in every session. A captured idea is dropped once a * feeds the router) in every session. A captured idea is dropped once a matching [IdeaDiscardedEvent]
* matching [IdeaDiscardedEvent] (or [IdeaPromotedEvent]) tombstones it (invariant #1 — the capture * (or [IdeaPromotedEvent]) tombstones it (invariant #1 — the capture stays in the log).
* stays in the log). *
* The board is folded ONCE at construction from [EventStore.allEvents] (a full, eager table load) and
* then kept current incrementally via [EventStore.subscribeAll]. This replaces the previous
* fold-the-entire-log-per-call design, which ran a full deserializing scan on every CHAT context
* build. The fold is idempotent (map put / set add), so the seed⇄live overlap needs no dedupe.
*/ */
class IdeaReader(private val eventStore: EventStore) { class IdeaReader(
eventStore: EventStore,
scope: CoroutineScope = CoroutineScope(Dispatchers.Default + SupervisorJob()),
) {
// id -> its capture event (source for text/sessionId); tombstoned = discarded/promoted ids.
private val captures = ConcurrentHashMap<String, IdeaCapturedEvent>()
private val tombstoned: MutableSet<String> = ConcurrentHashMap.newKeySet()
/** Active ideas (captured, not later discarded or promoted) across all sessions, newest first. */ init {
fun activeIdeas(): List<Idea> { eventStore.allEvents().forEach { apply(it.payload) }
val tombstoned = mutableSetOf<String>() scope.launch { eventStore.subscribeAll().collect { apply(it.payload) } }
val captured = mutableListOf<Idea>() }
eventStore.allEvents().forEach { stored ->
when (val payload = stored.payload) { private fun apply(payload: EventPayload) {
when (payload) {
is IdeaDiscardedEvent -> tombstoned += payload.ideaId is IdeaDiscardedEvent -> tombstoned += payload.ideaId
is IdeaPromotedEvent -> tombstoned += payload.ideaId is IdeaPromotedEvent -> tombstoned += payload.ideaId
is IdeaCapturedEvent -> captured += Idea(payload.ideaId, payload.text, payload.timestampMs) is IdeaCapturedEvent -> captures[payload.ideaId] = payload
else -> Unit else -> Unit
} }
} }
return captured.filterNot { it.id in tombstoned }.sortedByDescending { it.capturedAtMs }
} /** Active ideas (captured, not later discarded or promoted) across all sessions, newest first. */
fun activeIdeas(): List<Idea> =
captures.values
.filterNot { it.ideaId in tombstoned }
.map { Idea(it.ideaId, it.text, it.timestampMs) }
.sortedByDescending { it.capturedAtMs }
/** The session that captured [ideaId], so a tombstone lands alongside its capture. */ /** The session that captured [ideaId], so a tombstone lands alongside its capture. */
fun sessionOf(ideaId: String): SessionId? = fun sessionOf(ideaId: String): SessionId? = captures[ideaId]?.sessionId
capturedOf(ideaId)?.sessionId
/** The captured text of [ideaId] (so a promotion can write it into the project profile), or null. */ /** The captured text of [ideaId] (so a promotion can write it into the project profile), or null. */
fun textOf(ideaId: String): String? = fun textOf(ideaId: String): String? = captures[ideaId]?.text
capturedOf(ideaId)?.text
private fun capturedOf(ideaId: String): IdeaCapturedEvent? =
eventStore.allEvents()
.mapNotNull { it.payload as? IdeaCapturedEvent }
.firstOrNull { it.ideaId == ideaId }
} }