perf(quartz): index-driven fanout for LiveEventStore via FilterIndex<S>

Replace the SharedFlow-based broadcast in LiveEventStore with an
inverted index over filter-bearing subscribers. Each REQ registers
its filters into a per-store FilterIndex<LiveSubscription>; insert()
calls index.candidatesFor(event) and only delivers to candidates
whose Filter.match still passes. Cuts the per-event walk from
O(N_subs * N_filters) to a few hash lookups plus match() over a
small candidate set.

FilterIndex itself lives next to Filter.kt (commonMain, KMP-friendly,
AtomicReference + COW) so other call sites with the same shape
(LocalCache.observables, ObservableEventStore.changes) can reuse it.
Each filter contributes entries on its single most-selective dimension
(ids > authors > tags > tagsAll > kinds > unindexed) to keep buckets
narrow and avoid Set-dedupe work in candidatesFor.

The historical-replay race the previous SharedFlow + onSubscription
handoff closed is preserved by registering BEFORE replay starts and
deduping seen ids until EOSE.

geode/plans/2026-05-07-live-broadcast-fanout-index.md

@@ -1,9 +1,14 @@
 # Live broadcast: indexed filter matching for fanout
 
+> **Status (2026-05-07):** Phase 1 (relay server) and Phase 2 (Amethyst client
+> `LocalCache.observables`) are implemented. Phase 3 (per-projection dispatch
+> from `ObservableEventStore.changes`) is left as future work — see "What's
+> next" below.
+
 ## Problem
 
 Every accepted EVENT runs through `LiveEventStore.newEventStream`
-(`quartz/nip01Core/relay/server/LiveEventStore.kt:43`) — a
+(`quartz/nip01Core/relay/server/LiveEventStore.kt`) — a
 `MutableSharedFlow<Event>` that every active subscription collects.
 Each subscriber's collector then calls:
 
@@ -17,62 +22,104 @@ calls per EVENT — and each `Filter.match` itself walks `kinds`,
 `authors`, tag prefixes, since/until, etc. At 2k EPS ingest that's
 ~30M comparisons/sec.
 
-Two specific cost shapes:
+The same shape recurs in two more places:
 
-1. **Filters that almost never match.** Most subscriptions are scoped
+1. **`LocalCache.observables`** in `amethyst/.../LocalCache.kt` —
-   to a small author list. Today every published EVENT walks every
+   a `ConcurrentHashMap<Observable, Observable>` of feed observers.
-   such subscription to learn that. A `HashMap<HexKey, MutableList<Sub>>`
+   `refreshNewNoteObservers` iterates every observer for every
-   keyed by author would cut this to O(1) average for the dominant case.
+   accepted event; each observer's `new()` runs `filter.match`.
-2. **Pseudo-broadcast filters** (`{kinds: [1]}` with no other
+2. **`EventStoreProjection`** under `quartz/.../cache/projection/` —
-   constraint) match almost everything. There's no avoiding the
+   each projection collects every `StoreChange.Insert` from
-   per-subscriber notification, but at least the index lookup is
+   `ObservableEventStore.changes` and runs its own filter list.
-   cheap.
 
-`LoadBenchmark.fanoutLatency` already measures this — current
+All three follow the pattern "many filter-bearing observers, one
-results are not yet noted in tree, but back-of-envelope says fanout
+incoming event — find which observers match". Today that's a per-event
-becomes the dominant cost above ~2k subscribers.
+walk over N observers; with an inverted index it becomes a few hash
+lookups followed by `Filter.match` only on the (small) candidate set.
 
-## Sketch
+## Solution
 
-A new `LiveBroadcastIndex` inside `LiveEventStore`:
+### `FilterIndex<S>`
+
+`quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/filters/FilterIndex.kt`
+is a generic, KMP-friendly inverted index parameterised by the
+subscriber type. It lives next to `Filter.kt`, not under `relay/server/`,
+because it isn't relay-specific.
+
+API:
 
 ```kotlin
-private val byAuthor = ConcurrentHashMap<HexKey, MutableSet<Subscription>>()
+class FilterIndex<S : Any> {
-private val byKind   = ConcurrentHashMap<Int, MutableSet<Subscription>>()
+    fun register(filter: Filter, subscriber: S)
-private val byTag    = ConcurrentHashMap<TagKey, MutableSet<Subscription>>()
+    fun register(filters: List<Filter>, subscriber: S)
-private val unindexed = CopyOnWriteArraySet<Subscription>()  // subs with no
+    fun registerUnindexed(subscriber: S)            // predicate-only callers
-                                                             // narrowing field
+    fun unregister(subscriber: S)
+    fun candidatesFor(event: Event): Set<S>          // index lookup
+    fun forEach(action: (S) -> Unit)                 // full iteration (delete paths)
+    fun size(): Int
+    fun isEmpty(): Boolean
+}
 ```
 
-Each `RelaySession.handleReq` registers its `Subscription` (a tuple of
+State is held in a single `AtomicReference<State<S>>` (BanStore-style)
-filters + the existing `EventMessage` send callback) into whichever
+so reads in the hot path are wait-free. Writes copy-on-write; that's
-buckets each filter narrows on. A filter with `kinds=[1] and
+fine because writes are subscription-rate (rare) while reads are
-authors=[a,b]` registers into `byKind[1]` AND `byAuthor[a]`,
+event-rate (frequent).
-`byAuthor[b]` — broadcast unions the resulting candidate sets.
 
-On EVENT arrival:
+Indexing strategy: each filter contributes entries to **one**
+dimension — the most selective indexable field. Picking one dimension
+instead of all of them avoids over-counting subscribers in
+`candidatesFor` and minimises bucket churn:
 
-1. Build the candidate set: union of `byAuthor[event.pubkey]`,
+1. `ids` (most selective; an id matches one event).
-   `byKind[event.kind]`, every `byTag[(letter, value)]` for the
+2. `authors`.
-   event's single-letter tags, plus `unindexed`.
+3. The first single-letter tag in `tags` (then `tagsAll`).
-2. Run the existing `Filter.match` on each candidate to handle
+4. `kinds`.
-   negative constraints (`since`, `until`, `limit` already-reached,
+5. None of the above → registered into the unindexed pool.
-   composite predicates).
-3. Send.
 
-Expected: **>10× speedup** on fanout for realistic subscriptions.
+Multi-filter registrations OR the per-filter selections together,
-Worst case (all filters in `unindexed`) degrades to current behaviour.
+which mirrors `filters.any { it.match(...) }`. Negative constraints
+(`since` / `until` / `tagsAll` / saturated `limit`) stay in
+`Filter.match` — the index produces a super-set, callers post-filter.
 
-## Where it lives
+### Phase 1: `LiveEventStore`
 
-`quartz/nip01Core/relay/server/LiveBroadcastIndex.kt` — protocol-level,
+`LiveEventStore.kt` no longer uses a `MutableSharedFlow`. Instead:
-reusable by any relay embed. `RelaySession.handleReq` registers/
-unregisters; `LiveEventStore.insert` calls
-`index.candidatesFor(event)`.
 
-## How to verify
+- One `FilterIndex<LiveSubscription>` shared across all REQs.
+- `insert()` calls `index.candidatesFor(event)` then runs
+  `Filter.match` on each candidate. Synchronous delivery —
+  callers (the relay's `RelaySession`) keep the `deliver` callback
+  cheap (queue-to-outbound).
+- `query()` registers the subscription into the index *before* the
+  historical replay starts (closes the same race the previous
+  `onSubscription` handoff closed), runs the replay, signals EOSE,
+  then `awaitCancellation()`. Live events arrive via index dispatch
+  during the suspend; `finally { index.unregister(sub) }` cleans up.
+- Dedupe set during the historical phase is held in an
+  `AtomicReference<HashSet<String>?>` so the live-dispatch coroutine
+  sees the post-EOSE handoff promptly.
 
-Add `geode.perf.LoadBenchmark.fanoutScaling`:
+### Phase 2: `LocalCache.observables`
+
+`amethyst/.../LocalCache.kt` swapped its
+`ConcurrentHashMap<Observable, Observable>` for a
+`FilterIndex<Observable>`. `observeNotes` / `observeEvents` /
+`observeNewEvents(filter: Filter)` now `register(filter, observer)`;
+the predicate-only `observeNewEvents(predicate)` overload uses
+`registerUnindexed` because the index can't introspect an opaque
+predicate. Dispatch:
+
+- `refreshNewNoteObservers` iterates `observables.candidatesFor(event)`
+  instead of every observer.
+- `refreshDeletedNoteObservers` still uses `observables.forEach { ... }`
+  — the index doesn't help on the delete path because every observer
+  might hold the deleted note in its result set, and there's no
+  event-shape to consult.
+
+### How to verify
+
+`geode.perf.LoadBenchmark.fanoutScaling` (to be added):
 
 - N connections, each subscribes to `{authors: [pk_i], kinds: [1]}`.
 - Publish 10k EVENTs from a producer connection; each event matches
@@ -82,19 +129,47 @@ Add `geode.perf.LoadBenchmark.fanoutScaling`:
 Without the index, p99 grows roughly linearly with N. With the
 index, p99 should be flat up to a much higher N.
 
+For the LocalCache side, a similar benchmark would publish events
+matching one of M observers and measure dispatch cost as M grows.
+
 ## Risks
 
 - **Subscription churn**: re-subscribing on every page (the way some
   client features work) means many index insert/remove operations.
-  `ConcurrentHashMap` value-set operations need to be lock-free or
+  COW on a single `AtomicReference` makes each write a full inner-map
-  finely locked; benchmark this path explicitly.
+  copy; benchmark this path on a busy account to confirm the constants
+  stay reasonable.
 - **Tag explosion**: an EVENT with many `e`/`p` tags hits many tag
-  buckets. Cap candidate-set union work or short-circuit when the
+  buckets. The single-dimension-per-filter selection caps how many
-  union saturates.
+  buckets contribute candidates per event — registering on the
+  *most* selective dimension means tag-keyed filters typically pick
+  one specific tag value, so an event's tag walk only finds filters
+  registered under that exact `(letter, value)`.
 - **Memory**: the index is a per-bucket set of subscription handles.
-  At 5k subs × average 3 narrowing fields, ~15k entries — negligible.
+  At 5k subs × average 1 dimension × a handful of values per filter,
-- **Correctness fence**: the index must see new subscriptions before
+  ~10k–20k entries — negligible.
-  the next EVENT broadcast. Today `RelaySession.handleReq` writes its
+- **Correctness fence**: `register` happens before historical replay
-  `Job` into a `LargeCache` then launches the collector. Order of
+  on the relay side, and inside the `callbackFlow`'s `register` /
-  operations needs to be revisited so the index is updated atomically
+  `awaitClose { unregister }` pair on the client side. Events
-  with the collector being ready.
+  arriving mid-historical are deduped via `seenIds` (relay) or are a
+  non-issue because the client `observe*` flow seeds via `init()`
+  before any new event can fire.
+- **Filters with no narrowing field** (e.g. `{since: X}`) fall into
+  the unindexed pool and behave like today — every event reaches
+  them. That's the worst case; it's not worse than the pre-index
+  baseline.
+- **AddressableEvent / replaceable v2 path**: an observer holding v1
+  whose filter doesn't match v2 won't be in `candidatesFor(v2)`.
+  Today such an observer wouldn't update its membership either
+  (`filter.match(v2) == false` short-circuits before the
+  re-emit branch). Pre-existing behaviour preserved.
+
+## What's next (Phase 3)
+
+`ObservableEventStore.changes` is still a `SharedFlow<StoreChange>`
+that every projection collects. To use the index there, the dispatcher
+between `_changes.emit` and the per-projection collectors would consult
+a `FilterIndex<EventStoreProjection<*>>`-style index and only deliver
+to interested projections. Doable, but the SharedFlow contract is
+public; replacing it is a larger refactor than Phase 1/2 and the ROI
+is lower (per-projection apply is already small). Treat as a follow-up.

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/filters/FilterIndex.kt

@@ -0,0 +1,285 @@
+/*
+ * Copyright (c) 2025 Vitor Pamplona
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
+ * Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
+ * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+package com.vitorpamplona.quartz.nip01Core.relay.filters
+
+import com.vitorpamplona.quartz.nip01Core.core.Event
+import com.vitorpamplona.quartz.nip01Core.core.HexKey
+import kotlin.concurrent.atomics.AtomicReference
+import kotlin.concurrent.atomics.ExperimentalAtomicApi
+
+/**
+ * Inverted index over a population of [Filter]-bearing subscribers.
+ * Given an [Event], returns the (much smaller) set of subscribers
+ * whose filters could match — callers still run [Filter.match] on
+ * each candidate to enforce negative constraints (`since` / `until`
+ * / `tagsAll` / etc.) but skip the per-event walk over subscribers
+ * that share no narrowing field with the event.
+ *
+ * Used by the relay server's `LiveEventStore` for fanout from one
+ * inserted event to many `REQ` subscriptions, and by the client-side
+ * `LocalCache.observables` registry for the same shape inside the
+ * app (one accepted event, many feed observers).
+ *
+ * ## Indexing strategy
+ *
+ * Each registered filter contributes entries to **one** dimension —
+ * the most selective indexable field. Picking one dimension instead
+ * of all of them avoids over-counting subscribers in [candidatesFor]
+ * (no `Set` dedup work) and minimises bucket churn on register /
+ * unregister:
+ *
+ * 1. `ids` (most selective; an id matches one event).
+ * 2. `authors`.
+ * 3. The first single-letter tag in `tags` (then `tagsAll`).
+ * 4. `kinds`.
+ * 5. None of the above → registered into [unindexedKey].
+ *
+ * Multi-filter registrations OR the per-filter selections together
+ * (a subscriber matches if *any* of its filters matches the event,
+ * which mirrors `filters.any { it.match(...) }`).
+ *
+ * ## Concurrency
+ *
+ * State is held in a single [AtomicReference] and mutated via
+ * copy-on-write CAS loops, mirroring the
+ * `nip86RelayManagement.server.BanStore` pattern. Reads in
+ * [candidatesFor] and [forEach] are wait-free single-load atomic.
+ * Writes (subscription register / unregister) copy the inner maps
+ * — fine for this workload because writes are subscription-rate
+ * (rare) while reads are event-rate (frequent).
+ *
+ * ## What the index does NOT cover
+ *
+ *  - Negative filter constraints (`since`, `until`, `tagsAll`,
+ *    `limit`-already-saturated). The candidate set is a
+ *    super-set; callers must still run [Filter.match] on each
+ *    candidate.
+ *  - Subscribers driven by an arbitrary `(Event) -> Boolean`
+ *    predicate without an underlying [Filter]. Use
+ *    [registerUnindexed] for those — they're returned for every
+ *    event.
+ *  - Membership-driven re-evaluation paths (e.g. an addressable
+ *    `v2` that no longer matches a filter but the observer
+ *    already holds `v1`). Those callers must consult their own
+ *    membership state in addition to [candidatesFor].
+ */
+@OptIn(ExperimentalAtomicApi::class)
+class FilterIndex<S : Any> {
+    /**
+     * Bucket key. Five concrete shapes plus a sentinel for filters
+     * with no indexable narrowing field.
+     */
+    private sealed interface BucketKey
+
+    private data class IdKey(
+        val id: HexKey,
+    ) : BucketKey
+
+    private data class AuthorKey(
+        val author: HexKey,
+    ) : BucketKey
+
+    private data class TagKey(
+        val letter: String,
+        val value: String,
+    ) : BucketKey
+
+    private data class KindKey(
+        val kind: Int,
+    ) : BucketKey
+
+    private object Unindexed : BucketKey
+
+    /**
+     * Single immutable snapshot. [buckets] maps a key to the set of
+     * subscribers registered under it; [assignments] is the reverse
+     * map used by [unregister] to find a subscriber's keys without
+     * scanning every bucket.
+     */
+    private data class State<S>(
+        val buckets: Map<BucketKey, Set<S>> = emptyMap(),
+        val assignments: Map<S, Set<BucketKey>> = emptyMap(),
+    )
+
+    private val state: AtomicReference<State<S>> = AtomicReference(State())
+
+    /** Number of distinct subscribers currently registered. */
+    fun size(): Int = state.load().assignments.size
+
+    fun isEmpty(): Boolean = state.load().assignments.isEmpty()
+
+    /**
+     * Register [subscriber] under the bucket(s) selected for [filter].
+     * If [filter] has no indexable field the subscriber is added to
+     * the unindexed pool and is returned for every event.
+     */
+    fun register(
+        filter: Filter,
+        subscriber: S,
+    ) {
+        val keys = selectKeys(filter).ifEmpty { listOf(Unindexed) }
+        addAssignments(subscriber, keys)
+    }
+
+    /**
+     * Register [subscriber] for a list of filters (OR semantics).
+     * Each filter's most-selective dimension contributes its keys;
+     * any filter with no indexable field adds the subscriber to the
+     * unindexed pool, which dominates dispatch (the subscriber
+     * matches every event).
+     */
+    fun register(
+        filters: List<Filter>,
+        subscriber: S,
+    ) {
+        if (filters.isEmpty()) {
+            addAssignments(subscriber, listOf(Unindexed))
+            return
+        }
+        val keys = mutableListOf<BucketKey>()
+        for (f in filters) {
+            val perFilter = selectKeys(f)
+            if (perFilter.isEmpty()) {
+                keys.add(Unindexed)
+            } else {
+                keys.addAll(perFilter)
+            }
+        }
+        addAssignments(subscriber, keys)
+    }
+
+    /**
+     * Register [subscriber] in the unindexed pool. Use this for
+     * subscribers driven by an opaque predicate where the index
+     * can't infer a narrowing field.
+     */
+    fun registerUnindexed(subscriber: S) = addAssignments(subscriber, listOf(Unindexed))
+
+    /**
+     * Remove [subscriber] from every bucket it was registered in.
+     * No-op if the subscriber isn't currently registered.
+     */
+    fun unregister(subscriber: S) {
+        while (true) {
+            val current = state.load()
+            val keys = current.assignments[subscriber] ?: return
+            val newBuckets = current.buckets.toMutableMap()
+            for (key in keys) {
+                val cur = newBuckets[key] ?: continue
+                val next = cur - subscriber
+                if (next.isEmpty()) {
+                    newBuckets.remove(key)
+                } else {
+                    newBuckets[key] = next
+                }
+            }
+            val newAssignments = current.assignments - subscriber
+            if (state.compareAndSet(current, State(newBuckets, newAssignments))) return
+        }
+    }
+
+    /**
+     * Subscribers whose filters might match [event]. The result is a
+     * super-set: callers must still run `filter.match(event)` on each
+     * candidate to handle negative constraints.
+     *
+     * Iteration order is insertion-stable per call but otherwise
+     * unspecified.
+     */
+    fun candidatesFor(event: Event): Set<S> {
+        val s = state.load()
+        if (s.buckets.isEmpty()) return emptySet()
+        val result = LinkedHashSet<S>()
+        s.buckets[Unindexed]?.let { result.addAll(it) }
+        s.buckets[IdKey(event.id)]?.let { result.addAll(it) }
+        s.buckets[AuthorKey(event.pubKey)]?.let { result.addAll(it) }
+        s.buckets[KindKey(event.kind)]?.let { result.addAll(it) }
+        for (tag in event.tags) {
+            if (tag.size >= 2 && tag[0].length == 1) {
+                s.buckets[TagKey(tag[0], tag[1])]?.let { result.addAll(it) }
+            }
+        }
+        return result
+    }
+
+    /**
+     * Visit every registered subscriber. Used by callers that need
+     * to broadcast something the index can't help with (e.g.
+     * `LocalCache.refreshDeletedNoteObservers` — the deletion path
+     * has no event-shape to consult, every observer must see it).
+     */
+    fun forEach(action: (S) -> Unit) {
+        for (sub in state.load().assignments.keys) action(sub)
+    }
+
+    private fun addAssignments(
+        subscriber: S,
+        keys: List<BucketKey>,
+    ) {
+        if (keys.isEmpty()) return
+        val keySet = keys.toSet()
+        while (true) {
+            val current = state.load()
+            val newBuckets = current.buckets.toMutableMap()
+            for (key in keySet) {
+                val cur = newBuckets[key] ?: emptySet()
+                if (subscriber in cur) continue
+                newBuckets[key] = cur + subscriber
+            }
+            val existing = current.assignments[subscriber]
+            val merged = if (existing == null) keySet else existing + keySet
+            val newAssignments = current.assignments + (subscriber to merged)
+            if (state.compareAndSet(current, State(newBuckets, newAssignments))) return
+        }
+    }
+
+    /**
+     * Pick the most-selective indexable dimension for [filter] and
+     * expand it into one [BucketKey] per value. Returns an empty
+     * list if no field is indexable — caller maps that to [Unindexed].
+     */
+    private fun selectKeys(filter: Filter): List<BucketKey> {
+        if (!filter.ids.isNullOrEmpty()) {
+            return filter.ids.map { IdKey(it) }
+        }
+        if (!filter.authors.isNullOrEmpty()) {
+            return filter.authors.map { AuthorKey(it) }
+        }
+        if (!filter.tags.isNullOrEmpty()) {
+            val first =
+                filter.tags.entries.firstOrNull {
+                    it.key.length == 1 && it.value.isNotEmpty()
+                }
+            if (first != null) return first.value.map { TagKey(first.key, it) }
+        }
+        if (!filter.tagsAll.isNullOrEmpty()) {
+            val first =
+                filter.tagsAll.entries.firstOrNull {
+                    it.key.length == 1 && it.value.isNotEmpty()
+                }
+            if (first != null) return first.value.map { TagKey(first.key, it) }
+        }
+        if (!filter.kinds.isNullOrEmpty()) {
+            return filter.kinds.map { KindKey(it) }
+        }
+        return emptyList()
+    }
+}

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/LiveEventStore.kt

@@ -22,10 +22,11 @@ package com.vitorpamplona.quartz.nip01Core.relay.server
 
 import com.vitorpamplona.quartz.nip01Core.core.Event
 import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
+import com.vitorpamplona.quartz.nip01Core.relay.filters.FilterIndex
 import com.vitorpamplona.quartz.nip01Core.store.IEventStore
-import kotlinx.coroutines.channels.BufferOverflow
+import kotlinx.coroutines.awaitCancellation
-import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlin.concurrent.atomics.AtomicReference
-import kotlinx.coroutines.flow.onSubscription
+import kotlin.concurrent.atomics.ExperimentalAtomicApi
 
 /**
  * A reactive event store that combines historical data retrieval with live event streaming.
@@ -35,21 +36,41 @@ import kotlinx.coroutines.flow.onSubscription
  * End of Stored Events (EOSE), and then continues to stream matching new events as they
  * are inserted.
  *
+ * Live fanout from `insert` to interested subscribers is index-driven via [FilterIndex]:
+ * each [query] registers its filters; [insert] looks up the candidate set and only delivers
+ * to those whose filters actually match. This avoids the quadratic
+ * O(N_subscribers × N_filters_per_sub) per-event walk that a naïve broadcast would do.
+ *
  * @property store The underlying persistent storage for events.
  */
+@OptIn(ExperimentalAtomicApi::class)
 class LiveEventStore(
     private val store: IEventStore,
 ) {
-    private val newEventStream =
+    private val index = FilterIndex<LiveSubscription>()
-        MutableSharedFlow<Event>(
+
-            replay = 0,
+    /**
-            extraBufferCapacity = 100, // Optional: adjust for backpressure
+     * One live REQ subscription. Carries the filters (for the
-            onBufferOverflow = BufferOverflow.DROP_LATEST, // Default behavior
+     * post-index `match` re-check needed for negative constraints
+     * like `since` / `until` / `tagsAll`) and the delivery callback
+     * the index dispatches into. Identity-keyed inside [FilterIndex].
+     */
+    private class LiveSubscription(
+        val filters: List<Filter>,
+        val deliver: (Event) -> Unit,
     )
 
     suspend fun insert(event: Event) {
         store.insert(event)
-        newEventStream.tryEmit(event)
+        // Live fanout. The index returns a super-set; `match` enforces
+        // negative constraints. Synchronous delivery — callers are
+        // expected to keep `deliver` cheap (typically a `tryEmit` to
+        // a per-connection outbound queue).
+        for (sub in index.candidatesFor(event)) {
+            if (sub.filters.any { it.match(event) }) {
+                sub.deliver(event)
+            }
+        }
     }
 
     suspend fun query(
@@ -57,41 +78,50 @@ class LiveEventStore(
         onEach: (Event) -> Unit,
         onEose: () -> Unit,
     ) {
-        // Order matters: register the live collector BEFORE replaying
+        // During the historical replay, mark ids the store has
-        // stored events and signalling EOSE. Otherwise an event emitted
+        // emitted so the live path can dedupe. The index registers
-        // between EOSE and `collect` is lost because [newEventStream] has
+        // *before* the replay starts (otherwise an event accepted
-        // replay=0. The race is only occasionally visible for kinds the
+        // mid-replay would slip past the live path entirely — same
-        // store persists (insert latency masks it) but fires reliably for
+        // race the previous SharedFlow-based implementation closed
-        // ephemeral kinds (20000-29999) where insert is a no-op — and
+        // with `onSubscription`). Anything the store and the live
-        // ephemeral events MUST still reach matching live subscribers per
+        // path both observe gets dropped here on the live side.
-        // NIP-01.
         //
-        // Side effect of registering the collector first: an event
+        // Held in an AtomicReference because the live-dispatch
-        // inserted *during* `store.query` will be both replayed by the
+        // coroutine (which calls `deliver` from `insert`) needs to
-        // store AND emitted to the live stream. We dedupe by tracking
+        // see the post-EOSE handoff promptly. Once cleared to null,
-        // ids seen during the historical replay and skipping them on
+        // the dedupe check short-circuits and every live event is
-        // the live path. The set is dropped after EOSE so live-only
+        // forwarded. The set itself is mutated only from the
-        // events don't accumulate memory.
+        // historical-replay closure below, which runs on the same
-        var inHistoricalPhase = true
+        // coroutine that owns `query` — no cross-thread mutation.
-        var seenIds: HashSet<String>? = HashSet()
+        val seenIds = AtomicReference<HashSet<String>?>(HashSet())
-        val historicalOnEach: (Event) -> Unit = { event ->
+
-            seenIds?.add(event.id)
+        val sub =
+            LiveSubscription(
+                filters = filters,
+                deliver = { event ->
+                    val seen = seenIds.load()
+                    if (seen != null && seen.contains(event.id)) return@LiveSubscription
+                    onEach(event)
+                },
+            )
+
+        index.register(filters, sub)
+        try {
+            store.query<Event>(filters) { event ->
+                seenIds.load()?.add(event.id)
                 onEach(event)
             }
-        newEventStream
-            .onSubscription {
-                store.query(filters, historicalOnEach)
             onEose()
-                // Free the dedupe set once we've crossed EOSE: from
+            // Drop the dedupe set so the live path stops paying for
-                // here on the live stream is the only source of
+            // it. From this point the index drives delivery and
-                // events, so duplicates aren't possible.
+            // duplicates are no longer possible.
-                inHistoricalPhase = false
+            seenIds.store(null)
-                seenIds = null
+            // Suspend until the caller's coroutine is cancelled
-            }.collect { newEvent ->
+            // (e.g. NIP-01 CLOSE or connection drop). The `finally`
-                if (inHistoricalPhase && seenIds?.contains(newEvent.id) == true) return@collect
+            // unregisters from the index.
-                if (filters.any { it.match(newEvent) }) {
+            awaitCancellation()
-                    onEach(newEvent)
+        } finally {
-                }
+            index.unregister(sub)
         }
     }
 

quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/filters/FilterIndexTest.kt

@@ -0,0 +1,252 @@
+/*
+ * Copyright (c) 2025 Vitor Pamplona
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
+ * Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
+ * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+package com.vitorpamplona.quartz.nip01Core.relay.filters
+
+import com.vitorpamplona.quartz.nip01Core.core.Event
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertFalse
+import kotlin.test.assertTrue
+
+class FilterIndexTest {
+    private val authorA = "a".repeat(64)
+    private val authorB = "b".repeat(64)
+    private val authorC = "c".repeat(64)
+
+    private val pTag1 = "1".repeat(64)
+    private val pTag2 = "2".repeat(64)
+
+    private fun event(
+        id: String = "e".repeat(64),
+        pubkey: String = authorA,
+        kind: Int = 1,
+        tags: Array<Array<String>> = emptyArray(),
+        createdAt: Long = 1_700_000_000,
+    ) = Event(
+        id = id,
+        pubKey = pubkey,
+        createdAt = createdAt,
+        kind = kind,
+        tags = tags,
+        content = "",
+        sig = "",
+    )
+
+    /** Distinct identity wrapper so tests can hold a stable handle. */
+    private data class Sub(
+        val name: String,
+    )
+
+    @Test
+    fun emptyIndexReturnsNoCandidates() {
+        val index = FilterIndex<Sub>()
+        assertTrue(index.isEmpty())
+        assertTrue(index.candidatesFor(event()).isEmpty())
+    }
+
+    @Test
+    fun authorFilterMatchesByAuthor() {
+        val index = FilterIndex<Sub>()
+        val s = Sub("s")
+        index.register(Filter(authors = listOf(authorA)), s)
+
+        assertTrue(s in index.candidatesFor(event(pubkey = authorA)))
+        assertFalse(s in index.candidatesFor(event(pubkey = authorB)))
+    }
+
+    @Test
+    fun multipleAuthorsAllRoutedToSameSubscriber() {
+        val index = FilterIndex<Sub>()
+        val s = Sub("s")
+        index.register(Filter(authors = listOf(authorA, authorB)), s)
+
+        assertTrue(s in index.candidatesFor(event(pubkey = authorA)))
+        assertTrue(s in index.candidatesFor(event(pubkey = authorB)))
+        assertFalse(s in index.candidatesFor(event(pubkey = authorC)))
+    }
+
+    @Test
+    fun kindFilterMatchesByKind() {
+        val index = FilterIndex<Sub>()
+        val s = Sub("s")
+        index.register(Filter(kinds = listOf(1, 7)), s)
+
+        assertTrue(s in index.candidatesFor(event(kind = 1)))
+        assertTrue(s in index.candidatesFor(event(kind = 7)))
+        assertFalse(s in index.candidatesFor(event(kind = 30023)))
+    }
+
+    @Test
+    fun authorWinsOverKindWhenBothPresent() {
+        // Filter has authors AND kinds — the more-selective dimension
+        // (authors) is used. An event with the right kind but a
+        // different author must NOT appear in candidates, otherwise
+        // the index didn't actually narrow.
+        val index = FilterIndex<Sub>()
+        val s = Sub("s")
+        index.register(Filter(authors = listOf(authorA), kinds = listOf(1)), s)
+
+        assertTrue(s in index.candidatesFor(event(pubkey = authorA, kind = 1)))
+        // Wrong author, right kind — index excludes correctly.
+        assertFalse(s in index.candidatesFor(event(pubkey = authorB, kind = 1)))
+        // Right author, wrong kind — index includes; Filter.match
+        // would post-reject. Exposed candidate is acceptable.
+        assertTrue(s in index.candidatesFor(event(pubkey = authorA, kind = 7)))
+    }
+
+    @Test
+    fun idFilterMostSelective() {
+        val index = FilterIndex<Sub>()
+        val s = Sub("s")
+        val targetId = "9".repeat(64)
+        index.register(Filter(ids = listOf(targetId), kinds = listOf(1)), s)
+
+        assertTrue(s in index.candidatesFor(event(id = targetId)))
+        assertFalse(s in index.candidatesFor(event(id = "8".repeat(64))))
+    }
+
+    @Test
+    fun tagFilterMatchesEventsCarryingTheTag() {
+        val index = FilterIndex<Sub>()
+        val s = Sub("s")
+        index.register(Filter(tags = mapOf("p" to listOf(pTag1))), s)
+
+        val matching = event(tags = arrayOf(arrayOf("p", pTag1)))
+        val nonMatching = event(tags = arrayOf(arrayOf("p", pTag2)))
+
+        assertTrue(s in index.candidatesFor(matching))
+        assertFalse(s in index.candidatesFor(nonMatching))
+    }
+
+    @Test
+    fun unindexedFilterMatchesEverything() {
+        // A filter with no narrowing field (e.g. just `since`) lives
+        // in the unindexed pool. Every event must include it.
+        val index = FilterIndex<Sub>()
+        val s = Sub("s")
+        index.register(Filter(since = 1L), s)
+
+        assertTrue(s in index.candidatesFor(event(pubkey = authorA, kind = 1)))
+        assertTrue(s in index.candidatesFor(event(pubkey = authorB, kind = 30023)))
+    }
+
+    @Test
+    fun registerUnindexedExplicit() {
+        val index = FilterIndex<Sub>()
+        val s = Sub("s")
+        index.registerUnindexed(s)
+
+        assertTrue(s in index.candidatesFor(event(pubkey = authorA)))
+        assertTrue(s in index.candidatesFor(event(pubkey = authorB)))
+    }
+
+    @Test
+    fun unregisterRemovesFromAllBuckets() {
+        val index = FilterIndex<Sub>()
+        val s = Sub("s")
+        index.register(Filter(authors = listOf(authorA, authorB)), s)
+        assertEquals(1, index.size())
+
+        index.unregister(s)
+        assertEquals(0, index.size())
+        assertFalse(s in index.candidatesFor(event(pubkey = authorA)))
+        assertFalse(s in index.candidatesFor(event(pubkey = authorB)))
+    }
+
+    @Test
+    fun unregisterOfUnknownIsNoOp() {
+        val index = FilterIndex<Sub>()
+        val s = Sub("s")
+        index.unregister(s) // should not throw
+        assertEquals(0, index.size())
+    }
+
+    @Test
+    fun multiFilterRegistrationOrsAllSelections() {
+        // Subscriber wants events from authorA OR kind 30023.
+        val index = FilterIndex<Sub>()
+        val s = Sub("s")
+        index.register(
+            filters =
+                listOf(
+                    Filter(authors = listOf(authorA)),
+                    Filter(kinds = listOf(30023)),
+                ),
+            subscriber = s,
+        )
+
+        assertTrue(s in index.candidatesFor(event(pubkey = authorA, kind = 1)))
+        assertTrue(s in index.candidatesFor(event(pubkey = authorB, kind = 30023)))
+        assertFalse(s in index.candidatesFor(event(pubkey = authorB, kind = 1)))
+    }
+
+    @Test
+    fun multipleSubscribersUnionInCandidates() {
+        val index = FilterIndex<Sub>()
+        val s1 = Sub("s1")
+        val s2 = Sub("s2")
+        val s3 = Sub("s3")
+
+        index.register(Filter(authors = listOf(authorA)), s1)
+        index.register(Filter(authors = listOf(authorB)), s2)
+        index.register(Filter(kinds = listOf(1)), s3)
+
+        val cands = index.candidatesFor(event(pubkey = authorA, kind = 1))
+        // s1 hits via author, s3 via kind, s2 must be excluded.
+        assertTrue(s1 in cands)
+        assertTrue(s3 in cands)
+        assertFalse(s2 in cands)
+    }
+
+    @Test
+    fun forEachVisitsEverySubscriberOnce() {
+        val index = FilterIndex<Sub>()
+        val s1 = Sub("s1")
+        val s2 = Sub("s2")
+        val s3 = Sub("s3")
+        index.register(Filter(authors = listOf(authorA, authorB)), s1)
+        index.register(Filter(kinds = listOf(1)), s2)
+        index.registerUnindexed(s3)
+
+        val visited = mutableListOf<Sub>()
+        index.forEach { visited.add(it) }
+
+        assertEquals(3, visited.size)
+        assertEquals(setOf(s1, s2, s3), visited.toSet())
+    }
+
+    @Test
+    fun registerThenUnregisterLeavesNoStaleBuckets() {
+        // Churn test — repeatedly add and remove a subscriber and
+        // verify the index ends up empty (no leaked bucket entries).
+        val index = FilterIndex<Sub>()
+        val s = Sub("s")
+        repeat(100) {
+            index.register(
+                Filter(authors = listOf(authorA), kinds = listOf(1, 7), tags = mapOf("p" to listOf(pTag1))),
+                s,
+            )
+            index.unregister(s)
+        }
+        assertTrue(index.isEmpty())
+        assertTrue(index.candidatesFor(event(pubkey = authorA)).isEmpty())
+    }
+}