refactor(git): collapse status/PR-update indexes to map().stateIn

Now that observeEvents re-emits the whole matching list each time,
GitStatusIndex and GitPullRequestUpdateIndex no longer need the imperative
launch/collect-into-MutableStateFlow wrapper carried over from the old
newEventBundles version. Each is now a single observeEvents().map { reduce }
.stateIn(scope, Eagerly, null) — dropping startIfNeeded(), the AtomicBoolean
double-start guard, and the MutableStateFlow/asStateFlow pair.

Eagerly (not WhileSubscribed) is required: callers read .value synchronously
(isClosedOrResolved, the feed filters, the home open-count derivations) and
must not see a stale map when nobody is collecting. All startIfNeeded() call
sites removed; stateIn shares one upstream subscription across collectors.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01DpNmN8CvP6HnEsdTGAjVUr

amethyst/src/main/java/com/vitorpamplona/amethyst/model/GitPullRequestUpdateIndex.kt

@@ -27,11 +27,11 @@ import com.vitorpamplona.quartz.nip34Git.pr.GitPullRequestUpdateEvent
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.SupervisorJob
-import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.SharingStarted
 import kotlinx.coroutines.flow.StateFlow
-import kotlinx.coroutines.flow.asStateFlow
-import kotlinx.coroutines.launch
-import java.util.concurrent.atomic.AtomicBoolean
+import kotlinx.coroutines.flow.flowOn
+import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.flow.stateIn
 
 /**
  * Cross-screen index of the most recent NIP-34 pull-request update event
@@ -41,29 +41,21 @@ import java.util.concurrent.atomic.AtomicBoolean
  * than listing updates separately. Like [GitStatusIndex], updates aren't tracked
  * in `Note.replies`, so a per-row cache scan would otherwise be required.
  *
- * The kind-indexed [observeEvents] subscription replaces both the full-cache
- * `onStart` scan and the per-bundle type filtering the old
- * `LocalCache.live.newEventBundles` flow needed: the observable's `init()` seeds
- * the matching set from the index and re-emits the whole list on every new 1619,
- * so we just reduce it to the latest-per-parent map each time. PR updates are
- * rare, so recomputing the full map per emission is cheaper than it scanning
- * every event of every kind.
+ * The kind-indexed [observeEvents] re-emits the whole matching list on every new
+ * 1619 (and seeds it from the cache index via `init()`), so [latestByPullRequest]
+ * is just that list reduced to the latest-per-parent map. Shared [SharingStarted.Eagerly]
+ * — never `WhileSubscribed` — because callers read `.value` synchronously and must
+ * not see a stale map when no one is actively collecting. `null` means "not loaded yet".
  */
 object GitPullRequestUpdateIndex {
     private val scope = CoroutineScope(Dispatchers.IO + SupervisorJob())
-    private val started = AtomicBoolean(false)
 
-    private val mutableLatestByPullRequest = MutableStateFlow<Map<HexKey, GitPullRequestUpdateEvent>?>(null)
-    val latestByPullRequest: StateFlow<Map<HexKey, GitPullRequestUpdateEvent>?> = mutableLatestByPullRequest.asStateFlow()
-
-    fun startIfNeeded() {
-        if (!started.compareAndSet(false, true)) return
-        scope.launch {
-            LocalCache
-                .observeEvents<GitPullRequestUpdateEvent>(Filter(kinds = listOf(GitPullRequestUpdateEvent.KIND)))
-                .collect { events -> mutableLatestByPullRequest.value = latestByParent(events) }
-        }
-    }
+    val latestByPullRequest: StateFlow<Map<HexKey, GitPullRequestUpdateEvent>?> =
+        LocalCache
+            .observeEvents<GitPullRequestUpdateEvent>(Filter(kinds = listOf(GitPullRequestUpdateEvent.KIND)))
+            .map { latestByParent(it) }
+            .flowOn(Dispatchers.IO)
+            .stateIn(scope, SharingStarted.Eagerly, null)
 
     private fun latestByParent(events: List<GitPullRequestUpdateEvent>): Map<HexKey, GitPullRequestUpdateEvent> {
         val latest = HashMap<HexKey, GitPullRequestUpdateEvent>()

amethyst/src/main/java/com/vitorpamplona/amethyst/model/GitStatusIndex.kt

@@ -29,11 +29,11 @@ import com.vitorpamplona.quartz.nip34Git.status.GitStatusEvent
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.SupervisorJob
-import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.SharingStarted
 import kotlinx.coroutines.flow.StateFlow
-import kotlinx.coroutines.flow.asStateFlow
-import kotlinx.coroutines.launch
-import java.util.concurrent.atomic.AtomicBoolean
+import kotlinx.coroutines.flow.flowOn
+import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.flow.stateIn
 
 /**
  * Cross-screen index of the most recent NIP-34 status event (kinds
@@ -42,15 +42,15 @@ import java.util.concurrent.atomic.AtomicBoolean
  * `Note.replies` (see `LocalCache.computeReplyTo`), so the only way to
  * find them otherwise would be a full cache scan per row.
  *
- * The kind-indexed [observeEvents] subscription replaces both the full-cache
- * `onStart` scan and the per-bundle type filtering the old
- * `LocalCache.live.newEventBundles` flow needed: the observable's `init()` seeds
- * the matching set from the index and re-emits the whole list on every new
- * status event, so we just reduce it to the latest-per-target map each time.
+ * The kind-indexed [observeEvents] re-emits the whole matching list on every new
+ * status event (and seeds it from the cache index via `init()`), so [latestByTarget]
+ * is just that list reduced to the latest-per-target map. Shared [SharingStarted.Eagerly]
+ * — never `WhileSubscribed` — because callers (e.g. [isClosedOrResolved] and the feed
+ * filters) read `.value` synchronously and must not see a stale map when no one is
+ * actively collecting. `null` means "not loaded yet".
  */
 object GitStatusIndex {
     private val scope = CoroutineScope(Dispatchers.IO + SupervisorJob())
-    private val started = AtomicBoolean(false)
 
     private val statusKinds =
         listOf(
@@ -60,19 +60,14 @@ object GitStatusIndex {
             GitStatusEvent.KIND_DRAFT,
         )
 
-    private val mutableLatestByTarget = MutableStateFlow<Map<HexKey, GitStatusEvent>?>(null)
-    val latestByTarget: StateFlow<Map<HexKey, GitStatusEvent>?> = mutableLatestByTarget.asStateFlow()
+    val latestByTarget: StateFlow<Map<HexKey, GitStatusEvent>?> =
+        LocalCache
+            .observeEvents<GitStatusEvent>(Filter(kinds = statusKinds))
+            .map { reduceLatestByTarget(it) }
+            .flowOn(Dispatchers.IO)
+            .stateIn(scope, SharingStarted.Eagerly, null)
 
-    fun startIfNeeded() {
-        if (!started.compareAndSet(false, true)) return
-        scope.launch {
-            LocalCache
-                .observeEvents<GitStatusEvent>(Filter(kinds = statusKinds))
-                .collect { events -> mutableLatestByTarget.value = latestByTarget(events) }
-        }
-    }
-
-    private fun latestByTarget(events: List<GitStatusEvent>): Map<HexKey, GitStatusEvent> {
+    private fun reduceLatestByTarget(events: List<GitStatusEvent>): Map<HexKey, GitStatusEvent> {
         val latest = HashMap<HexKey, GitStatusEvent>()
         for (event in events) {
             val target = event.rootEventId() ?: continue

amethyst/src/main/java/com/vitorpamplona/amethyst/ui/note/types/Git.kt

@@ -582,7 +582,6 @@ private fun RenderGitPullRequestEvent(
 ) {
     // A later pull-request update (kind 1619) revises this PR with a newer commit /
     // merge base. Fold the most recent one in so the card reflects the current state.
-    LaunchedEffect(Unit) { GitPullRequestUpdateIndex.startIfNeeded() }
     val updateIndex by GitPullRequestUpdateIndex.latestByPullRequest.collectAsStateWithLifecycle()
     val update = updateIndex?.get(note.idHex)
 

amethyst/src/main/java/com/vitorpamplona/amethyst/ui/note/types/GitStatusActions.kt

@@ -30,7 +30,6 @@ import androidx.compose.material3.MaterialTheme
 import androidx.compose.material3.OutlinedButton
 import androidx.compose.material3.Text
 import androidx.compose.runtime.Composable
-import androidx.compose.runtime.LaunchedEffect
 import androidx.compose.runtime.getValue
 import androidx.compose.runtime.remember
 import androidx.compose.ui.Modifier
@@ -75,7 +74,6 @@ fun GitStatusActions(
     val canModerate = remember(event, note) { canModerate(repoAddress, note, accountViewModel) }
     if (!canModerate) return
 
-    LaunchedEffect(Unit) { GitStatusIndex.startIfNeeded() }
     val index by GitStatusIndex.latestByTarget.collectAsStateWithLifecycle()
     if (index == null) return
     val current = index?.get(note.idHex)

amethyst/src/main/java/com/vitorpamplona/amethyst/ui/note/types/GitStatusPill.kt

@@ -29,7 +29,6 @@ import androidx.compose.foundation.shape.RoundedCornerShape
 import androidx.compose.material3.MaterialTheme
 import androidx.compose.material3.Text
 import androidx.compose.runtime.Composable
-import androidx.compose.runtime.LaunchedEffect
 import androidx.compose.runtime.getValue
 import androidx.compose.ui.Alignment
 import androidx.compose.ui.Modifier
@@ -74,7 +73,6 @@ fun GitStatusPill(
     modifier: Modifier = Modifier,
     defaultIfMissing: StatusKind? = null,
 ) {
-    LaunchedEffect(Unit) { GitStatusIndex.startIfNeeded() }
     val index by GitStatusIndex.latestByTarget.collectAsStateWithLifecycle()
     val map = index ?: return // hide pill until the initial scan completes — avoids a default-then-real flicker
     val kind = map[targetIdHex]?.statusKind() ?: defaultIfMissing ?: return

amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/gitRepo/GitItemListRow.kt

@@ -37,7 +37,6 @@ import androidx.compose.material3.HorizontalDivider
 import androidx.compose.material3.MaterialTheme
 import androidx.compose.material3.Text
 import androidx.compose.runtime.Composable
-import androidx.compose.runtime.LaunchedEffect
 import androidx.compose.runtime.getValue
 import androidx.compose.runtime.remember
 import androidx.compose.ui.Alignment
@@ -244,7 +243,6 @@ private fun LabelChip(label: String) {
 /** Small "Revised" badge shown when a pull request has a later kind-1619 update. */
 @Composable
 private fun GitRevisedChip(prIdHex: String) {
-    LaunchedEffect(Unit) { GitPullRequestUpdateIndex.startIfNeeded() }
     val index by GitPullRequestUpdateIndex.latestByPullRequest.collectAsStateWithLifecycle()
     if (index?.get(prIdHex) == null) return
 

amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/gitRepo/GitRepositoryScreen.kt

@@ -257,9 +257,8 @@ private fun GitRepositoryHome(
         }
 
     // Nav-card badges count only the OPEN issues/PRs. The open/closed split needs the status
-    // index (kinds 1630-1633), which is started here so the home reflects it without visiting
-    // the Issues screen first; the count is then derived directly from the live index.
-    LaunchedEffect(Unit) { GitStatusIndex.startIfNeeded() }
+    // index (kinds 1630-1633); it's an eagerly-shared StateFlow, so the home reflects it without
+    // visiting the Issues screen first. The count is derived directly from the live index.
     val statusMap by GitStatusIndex.latestByTarget.collectAsStateWithLifecycle()
     val openIssueCount =
         remember(openIssueItems, closedIssueItems, statusMap) {

amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/gitRepo/dal/RepositoryIssuesFeedViewModel.kt

@@ -39,7 +39,6 @@ class RepositoryIssuesFeedViewModel(
         // Status events (kinds 1630-1633) don't mutate the issue note, so the additive
         // feed update can't move an item between the Open/Closed buckets on its own.
         // Watch the status index and force a full re-partition whenever it changes.
-        GitStatusIndex.startIfNeeded()
         viewModelScope.launch(Dispatchers.IO) {
             GitStatusIndex.latestByTarget.collect { invalidateData() }
         }

amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/gitRepo/dal/RepositoryPatchesFeedViewModel.kt

@@ -39,7 +39,6 @@ class RepositoryPatchesFeedViewModel(
         // Status events (kinds 1630-1633) don't mutate the patch/PR note, so the additive
         // feed update can't move an item between the Open/Closed buckets on its own.
         // Watch the status index and force a full re-partition whenever it changes.
-        GitStatusIndex.startIfNeeded()
         viewModelScope.launch(Dispatchers.IO) {
             GitStatusIndex.latestByTarget.collect { invalidateData() }
         }