@@ -1,6 +1,7 @@
1package soy.iko.opencode.data.repo
1package soy.iko.opencode.data.repo
3import soy.iko.opencode.data.model.BusEvent
3import soy.iko.opencode.data.model.BusEvent
4+import soy.iko.opencode.data.model.MessageInfo
4import soy.iko.opencode.data.model.MessagePartRemoved
5import soy.iko.opencode.data.model.MessagePartRemoved
5import soy.iko.opencode.data.model.MessagePartUpdated
6import soy.iko.opencode.data.model.MessagePartUpdated
6import soy.iko.opencode.data.model.MessageRemoved
7import soy.iko.opencode.data.model.MessageRemoved
@@ -17,11 +18,13 @@ import soy.iko.opencode.data.network.OpencodeApiClient
17import soy.iko.opencode.util.runCatchingCancellable
18import soy.iko.opencode.util.runCatchingCancellable
18import soy.iko.opencode.util.safeExceptionSummary
19import soy.iko.opencode.util.safeExceptionSummary
19import android.util.Log
20import android.util.Log
21+import kotlinx.coroutines.Dispatchers
20import kotlinx.coroutines.channels.Channel
22import kotlinx.coroutines.channels.Channel
21import kotlinx.coroutines.flow.Flow
23import kotlinx.coroutines.flow.Flow
22import kotlinx.coroutines.flow.channelFlow
24import kotlinx.coroutines.flow.channelFlow
23import kotlinx.coroutines.flow.consumeAsFlow
25import kotlinx.coroutines.flow.consumeAsFlow
24import kotlinx.coroutines.flow.conflate
26import kotlinx.coroutines.flow.conflate
27+import kotlinx.coroutines.flow.flowOn
25import kotlinx.coroutines.launch
28import kotlinx.coroutines.launch
26import kotlinx.coroutines.sync.Mutex
29import kotlinx.coroutines.sync.Mutex
27import kotlinx.coroutines.sync.withLock
30import kotlinx.coroutines.sync.withLock
@@ -159,7 +162,13 @@ open class SessionRepository(
159 val snapshot = lock.withLock { store.snapshot() }
162 val snapshot = lock.withLock { store.snapshot() }
160 send(snapshot)
163 send(snapshot)
165+ // Run the whole reducer pipeline — event reduction, list rebuilds, and the O(N)
166+ // snapshot copy — on Dispatchers.Default rather than the collector's context. The
167+ // sole consumer collects via stateIn(viewModelScope), i.e. Main.immediate, so without
168+ // this every streamed token's reduce()/snapshot() would execute on the UI thread
169+ // (hundreds/sec on a fast run). The seed REST calls here only suspend, so moving them
170+ // off Main is harmless. JSON decoding already runs off-Main upstream in EventStreamClient.
162− }.conflate()
171+ }.flowOn(Dispatchers.Default).conflate()
164 companion object {
173 companion object {
165 /** Convenience: is this event a run-completion signal for [sessionId]?
174 /** Convenience: is this event a run-completion signal for [sessionId]?
@@ -178,8 +187,34 @@ open class SessionRepository(
178 * Exposed as `internal` so the reducer logic can be unit-tested directly.
187 * Exposed as `internal` so the reducer logic can be unit-tested directly.
180internal class MessageStore {
189internal class MessageStore {
190+ // messageId -> mutable holder (info + insertion-ordered parts); the map itself is
181− // messageId -> (info + parts), insertion-ordered.
191+ // insertion-ordered.
182− private val messages = LinkedHashMap<String, MessageWithParts>()
192+ private val messages = LinkedHashMap<String, Holder>()
195+ * Mutable per-message holder. Parts live in an insertion-ordered map keyed by part id, so a
196+ * streamed part update is an O(1) put instead of an O(P) list rebuild. The immutable
197+ * [MessageWithParts] view is materialized lazily and cached until the next mutation — so a
198+ * burst of streamed token updates rebuilds a message's part list at most once per published
199+ * [snapshot], not once per token. The cached view is also reference-stable across snapshots
200+ * while unchanged, so Compose skips recomposing bubbles that didn't change.
202+ private class Holder(var info: MessageInfo, val parts: LinkedHashMap<String, Part>) {
203+ private var cached: MessageWithParts? = null
204+ fun view(): MessageWithParts =
205+ cached ?: MessageWithParts(info, parts.values.toList()).also { cached = it }
206+ /** Adopt an existing immutable instance as the cache — avoids an immediate rebuild and
207+ * preserves its identity (e.g. when seeding an authoritative REST snapshot). */
208+ fun adopt(m: MessageWithParts) { cached = m }
209+ fun invalidate() { cached = null }
212+ /** Build a [Holder] from an immutable message, reusing [m] itself as the cached view. */
213+ private fun holderOf(m: MessageWithParts): Holder {
214+ val parts = LinkedHashMap<String, Part>(m.parts.size)
215+ for (p in m.parts) parts[p.id] = p
216+ return Holder(m.info, parts).also { it.adopt(m) }
184 /** Part ids upserted via [reduce] since the last reseed pivot ([beginReseed]) or seed.
219 /** Part ids upserted via [reduce] since the last reseed pivot ([beginReseed]) or seed.
185 * On a reconnect re-seed these are the parts that streamed in *during* the REST fetch,
220 * On a reconnect re-seed these are the parts that streamed in *during* the REST fetch,
@@ -211,7 +246,7 @@ internal class MessageStore {
211 /** Maximum number of messages to keep in memory; oldest are evicted when exceeded. */
246 /** Maximum number of messages to keep in memory; oldest are evicted when exceeded. */
212 internal val maxMessages = NetworkConfig.maxInMemoryMessages
247 internal val maxMessages = NetworkConfig.maxInMemoryMessages
214− fun snapshot(): List<MessageWithParts> = messages.values.toList()
249+ fun snapshot(): List<MessageWithParts> = messages.values.map { it.view() }
216 fun seed(initial: List<MessageWithParts>, prune: Boolean = false): Boolean {
251 fun seed(initial: List<MessageWithParts>, prune: Boolean = false): Boolean {
217 var changed = false
252 var changed = false
@@ -230,7 +265,7 @@ internal class MessageStore {
230 for (m in initial) {
265 for (m in initial) {
231 val existing = messages[m.info.id]
266 val existing = messages[m.info.id]
232 if (existing == null) {
267 if (existing == null) {
233− messages[m.info.id] = m
268+ messages[m.info.id] = holderOf(m)
234 changed = true
269 changed = true
236 // A part streamed in between subscribe and this initial load may already
271 // A part streamed in between subscribe and this initial load may already
@@ -238,7 +273,8 @@ internal class MessageStore {
238 // overwriting so that newer streamed part isn't discarded: take the
273 // overwriting so that newer streamed part isn't discarded: take the
239 // snapshot's part order, swap in the streamed version where ids overlap,
274 // snapshot's part order, swap in the streamed version where ids overlap,
240 // append any streamed-only parts, and adopt the authoritative REST info.
275 // append any streamed-only parts, and adopt the authoritative REST info.
276+ // existing.parts is already keyed by part id (insertion-ordered).
241− val streamedById = existing.parts.associateBy { it.id }
277+ val streamedById = existing.parts
242 val snapshotIds = m.parts.mapTo(mutableSetOf()) { it.id }
278 val snapshotIds = m.parts.mapTo(mutableSetOf()) { it.id }
243 // Prefer the in-memory version of an overlapping part ONLY if it streamed
279 // Prefer the in-memory version of an overlapping part ONLY if it streamed
244 // in since the reseed pivot (i.e. during this fetch), so it's newer than
280 // in since the reseed pivot (i.e. during this fetch), so it's newer than
@@ -247,7 +283,7 @@ internal class MessageStore {
247 val ordered = m.parts.map { p ->
283 val ordered = m.parts.map { p ->
248 if (p.id in streamedSincePivot) streamedById[p.id] ?: p else p
284 if (p.id in streamedSincePivot) streamedById[p.id] ?: p else p
249 }.toMutableList()
285 }.toMutableList()
250− for (p in existing.parts) {
286+ for (p in existing.parts.values) {
251 if (p.id !in snapshotIds) ordered.add(p)
287 if (p.id !in snapshotIds) ordered.add(p)
253 // Adopt the authoritative REST info unless the in-memory info was itself
289 // Adopt the authoritative REST info unless the in-memory info was itself
@@ -256,8 +292,8 @@ internal class MessageStore {
256 // in-memory info that went stale during a disconnect.
292 // in-memory info that went stale during a disconnect.
257 val info = if (existing.info is UnknownMessage || m.info.id !in messageInfoUpdatedSincePivot) m.info else existing.info
293 val info = if (existing.info is UnknownMessage || m.info.id !in messageInfoUpdatedSincePivot) m.info else existing.info
258 val merged = MessageWithParts(info = info, parts = ordered)
294 val merged = MessageWithParts(info = info, parts = ordered)
259− if (merged != existing) {
295+ if (merged != existing.view()) {
260− messages[m.info.id] = merged
296+ messages[m.info.id] = holderOf(merged)
261 changed = true
297 changed = true
@@ -284,7 +320,13 @@ internal class MessageStore {
284 private fun reorderByTime(): Boolean {
320 private fun reorderByTime(): Boolean {
285 val entries = messages.entries.map { it.key to it.value }
321 val entries = messages.entries.map { it.key to it.value }
286 val sorted = entries.sortedBy { it.second.info.time?.created ?: Long.MAX_VALUE }
322 val sorted = entries.sortedBy { it.second.info.time?.created ?: Long.MAX_VALUE }
287− if (sorted.map { it.first } == entries.map { it.first }) return false
323+ // Detect a reordering by comparing keys positionally, instead of allocating two more
324+ // N-sized key lists just to compare them.
326+ for (i in sorted.indices) {
327+ if (sorted[i].first != entries[i].first) { changed = true; break }
329+ if (!changed) return false
288 messages.clear()
330 messages.clear()
289 for ((k, v) in sorted) messages[k] = v
331 for ((k, v) in sorted) messages[k] = v
290 return true
332 return true
@@ -293,30 +335,7 @@ internal class MessageStore {
293 /** Returns true if the state changed (and a new snapshot should be published). */
335 /** Returns true if the state changed (and a new snapshot should be published). */
294 fun reduce(sessionId: String, event: BusEvent): Boolean {
336 fun reduce(sessionId: String, event: BusEvent): Boolean {
295 return when (event) {
337 return when (event) {
296− is MessageUpdated -> {
338+ is MessageUpdated -> handleMessageUpdated(sessionId, event)
297− val info = event.properties.info
298− if (info.sessionID != sessionId) {
301− // An UnknownMessage with an empty id (e.g. from an unrecognized
302− // server role with no id field) would collide with other such
303− // messages in the map. Generate a unique synthetic key so each
304− // unknown message gets its own entry instead of overwriting others.
305− val key = if (info.id.isEmpty() && info is UnknownMessage) {
306− "unknown-${unknownCounter.incrementAndGet()}"
310− val existing = messages[key]
311− messages[key] = existing?.copy(info = info) ?: MessageWithParts(info)
312− // Record that this message's info streamed in live, so a re-seed after a
313− // reconnect knows it's newer than the REST snapshot and keeps it (see
314− // seed()/beginReseed()).
315− messageInfoUpdatedSincePivot.add(key)
316− if (existing == null) evictOldMessages()
321 is MessagePartUpdated -> {
340 is MessagePartUpdated -> {
322 val part = event.properties.part
341 val part = event.properties.part
@@ -343,6 +362,34 @@ internal class MessageStore {
365+ private fun handleMessageUpdated(sessionId: String, event: MessageUpdated): Boolean {
366+ val info = event.properties.info
367+ if (info.sessionID != sessionId) return false
368+ // An UnknownMessage with an empty id (e.g. from an unrecognized server role with no id
369+ // field) would collide with other such messages in the map. Generate a unique synthetic
370+ // key so each unknown message gets its own entry instead of overwriting others.
371+ val key = if (info.id.isEmpty() && info is UnknownMessage) {
372+ "unknown-${unknownCounter.incrementAndGet()}"
376+ val existing = messages[key]
377+ // A server re-sending byte-identical info is a no-op: skip it so we don't publish a
378+ // redundant snapshot (mirrors upsertPart's identical-part guard).
379+ if (existing != null && existing.info == info) return false
380+ if (existing == null) {
381+ messages[key] = Holder(info, LinkedHashMap())
383+ existing.info = info
384+ existing.invalidate()
386+ // Record that this message's info streamed in live, so a re-seed after a reconnect knows
387+ // it's newer than the REST snapshot and keeps it (see seed()/beginReseed()).
388+ messageInfoUpdatedSincePivot.add(key)
389+ if (existing == null) evictOldMessages()
346 /** Returns false if [eventSession] is non-null and doesn't match [sessionId]. */
393 /** Returns false if [eventSession] is non-null and doesn't match [sessionId]. */
347 private fun matchesSession(eventSession: String?, sessionId: String): Boolean =
394 private fun matchesSession(eventSession: String?, sessionId: String): Boolean =
348 eventSession == null || eventSession == sessionId
395 eventSession == null || eventSession == sessionId
@@ -359,25 +406,27 @@ internal class MessageStore {
359 if (!matchesSession(event.properties.sessionID, sessionId)) return false
406 if (!matchesSession(event.properties.sessionID, sessionId)) return false
360 val removed = messages.remove(id) ?: return false
407 val removed = messages.remove(id) ?: return false
361 messageInfoUpdatedSincePivot.remove(id)
408 messageInfoUpdatedSincePivot.remove(id)
362− removed.parts.forEach { streamedSincePivot.remove(it.id) }
409+ removed.parts.keys.forEach { streamedSincePivot.remove(it) }
363 return true
410 return true
366 private fun upsertPart(messageId: String, part: Part): Boolean {
413 private fun upsertPart(messageId: String, part: Part): Boolean {
367 val current = messages[messageId]
414 val current = messages[messageId]
368− val parts = current?.parts.orEmpty()
415+ if (current != null) {
369− val idx = parts.indexOfFirst { it.id == part.id }
416+ // O(1) map lookup + put, vs the former O(P) indexOfFirst scan and full-list copy
417+ // on every streamed token. A LinkedHashMap re-put keeps the part's original
418+ // position (replace-in-place); a new id is appended — matching the old list order.
370− if (idx >= 0 && parts[idx] == part) return false
419+ if (current.parts[part.id] == part) return false
371− val newParts = if (idx >= 0) {
420+ current.parts[part.id] = part
372− parts.toMutableList().also { it[idx] = part }
421+ current.invalidate()
374− parts + part
423+ val parts = LinkedHashMap<String, Part>()
375− }
424+ parts[part.id] = part
376− messages[messageId] = current?.copy(parts = newParts)
425+ messages[messageId] = Holder(
377− ?: MessageWithParts(
378 info = UnknownMessage(id = messageId, sessionID = part.sessionID ?: ""),
426 info = UnknownMessage(id = messageId, sessionID = part.sessionID ?: ""),
379− parts = newParts,
427+ parts = parts,
381 // Record that this part streamed in, so a re-seed after a reconnect knows it's
430 // Record that this part streamed in, so a re-seed after a reconnect knows it's
382 // newer than the REST snapshot and keeps it (see seed()/beginReseed()).
431 // newer than the REST snapshot and keeps it (see seed()/beginReseed()).
383 streamedSincePivot.add(part.id)
432 streamedSincePivot.add(part.id)
@@ -387,9 +436,8 @@ internal class MessageStore {
388 private fun removePart(messageId: String, partId: String): Boolean {
437 private fun removePart(messageId: String, partId: String): Boolean {
389 val current = messages[messageId] ?: return false
438 val current = messages[messageId] ?: return false
390− val newParts = current.parts.filterNot { it.id == partId }
439+ if (current.parts.remove(partId) == null) return false
391− if (newParts.size == current.parts.size) return false
440+ current.invalidate()
392− messages[messageId] = current.copy(parts = newParts)
393 return true
441 return true
@@ -404,7 +452,7 @@ internal class MessageStore {
404 // unbounded over a long-running session that never reconnects (they're only
452 // unbounded over a long-running session that never reconnects (they're only
405 // otherwise cleared on reseed/seed).
453 // otherwise cleared on reseed/seed).
406 messageInfoUpdatedSincePivot.remove(oldestKey)
454 messageInfoUpdatedSincePivot.remove(oldestKey)
407− removed?.parts?.forEach { streamedSincePivot.remove(it.id) }
455+ removed?.parts?.keys?.forEach { streamedSincePivot.remove(it) }
408 evicted = true
456 evicted = true
410 return evicted
458 return evicted