Merge pull request #2772 from vitorpamplona/claude/review-fanout-index-e1Uqq

Add FilterIndex for inverted-index fanout dispatch

amethyst/src/main/java/com/vitorpamplona/amethyst/DebugUtils.kt

@@ -139,7 +139,7 @@ fun debugState(context: Context) {
     Log.d(
         STATE_DUMP_TAG,
         "Observables: " +
-            LocalCache.observables.size,
+            LocalCache.observables.size(),
     )
 
     Log.d(

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

@@ -88,6 +88,7 @@ import com.vitorpamplona.quartz.nip01Core.metadata.MetadataEvent
 import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
 import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.EventCmd
 import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
+import com.vitorpamplona.quartz.nip01Core.relay.filters.FilterIndex
 import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
 import com.vitorpamplona.quartz.nip01Core.tags.aTag.ATag
 import com.vitorpamplona.quartz.nip01Core.tags.aTag.taggedAddresses
@@ -261,7 +262,6 @@ import java.io.File
 import java.io.FileOutputStream
 import java.io.IOException
 import java.util.SortedSet
-import java.util.concurrent.ConcurrentHashMap
 
 interface ILocalCache {
     fun markAsSeen(
@@ -290,7 +290,15 @@ object LocalCache : ILocalCache, ICacheProvider {
 
     val deletionIndex = DeletionIndex()
 
-    val observables = ConcurrentHashMap<Observable, Observable>(10)
+    /**
+     * Inverted index over the active [Observable]s. New events fan
+     * out only to observers whose filter actually narrows on a field
+     * the event carries (author, kind, single-letter tag), instead
+     * of waking every observer per event. Predicate-only observers
+     * (no underlying [Filter]) live in the unindexed pool and still
+     * see every event.
+     */
+    val observables = FilterIndex<Observable>()
 
     fun Filter.match(note: Note): Boolean {
         val event = note.event
@@ -361,10 +369,10 @@ object LocalCache : ILocalCache, ICacheProvider {
 
             newFilter.init()
 
-            observables[newFilter] = newFilter
+            observables.register(filter, newFilter)
 
             awaitClose {
-                observables.remove(newFilter)
+                observables.unregister(newFilter)
             }
         }.buffer(kotlinx.coroutines.channels.Channel.CONFLATED)
 
@@ -377,10 +385,10 @@ object LocalCache : ILocalCache, ICacheProvider {
 
             cachedFilter.init()
 
-            observables.put(cachedFilter, cachedFilter)
+            observables.register(filter, cachedFilter)
 
             awaitClose {
-                observables.remove(cachedFilter)
+                observables.unregister(cachedFilter)
             }
         }.buffer(kotlinx.coroutines.channels.Channel.CONFLATED)
 
@@ -402,14 +410,28 @@ object LocalCache : ILocalCache, ICacheProvider {
                     trySend(it)
                 }
 
-            observables.put(newFilter, newFilter)
+            // Unindexed: predicate is opaque, the index can't narrow.
+            // Caller is delivered every event and runs the predicate.
+            observables.registerUnindexed(newFilter)
 
             awaitClose {
-                observables.remove(newFilter)
+                observables.unregister(newFilter)
             }
         }
 
-    fun <T : Event> observeNewEvents(filter: Filter): Flow<T> = observeNewEvents(filter::match)
+    fun <T : Event> observeNewEvents(filter: Filter): Flow<T> =
+        callbackFlow {
+            val newFilter =
+                NewEventMatchingFilter<T>(filter::match) {
+                    trySend(it)
+                }
+
+            observables.register(filter, newFilter)
+
+            awaitClose {
+                observables.unregister(newFilter)
+            }
+        }
 
     @Suppress("UNCHECKED_CAST")
     fun <T : Event> observeLatestEvent(filter: Filter) = observeEvents<T>(filter).map { it.firstOrNull() }
@@ -2514,22 +2536,21 @@ object LocalCache : ILocalCache, ICacheProvider {
     private fun refreshNewNoteObservers(newNote: Note) {
         val event = newNote.event as Event
 
-        val observableBiConsumer =
-            java.util.function.BiConsumer<Observable, Observable> { _, u ->
-                u.new(event, newNote)
-            }
-
-        observables.forEach(observableBiConsumer)
+        // Index-driven fanout: only observers whose filter narrows
+        // on a field this event carries (or that registered as
+        // unindexed) get woken up. The match check inside each
+        // observer's `new()` still enforces negative constraints.
+        for (observer in observables.candidatesFor(event)) {
+            observer.new(event, newNote)
+        }
         live.newNote(newNote)
     }
 
     private fun refreshDeletedNoteObservers(newNote: Note) {
-        val observableBiConsumer =
-            java.util.function.BiConsumer<Observable, Observable> { _, u ->
-                u.remove(newNote)
-            }
-
-        observables.forEach(observableBiConsumer)
+        // Deletes don't have a filterable shape — every observer
+        // might hold this note in its result set, so iterate them
+        // all. The index doesn't help here.
+        observables.forEach { it.remove(newNote) }
         live.removedNote(newNote)
     }
 

geode/build.gradle.kts

@@ -42,6 +42,8 @@ tasks.withType<Test>().configureEach {
     // perf.LoadBenchmark tests opt in. Off by default — load tests
     // are noisy and slow.
     systemProperty("runLoadBenchmark", System.getProperty("runLoadBenchmark") ?: "false")
+    System.getProperty("fanoutScalingEvents")?.let { systemProperty("fanoutScalingEvents", it) }
+    System.getProperty("fanoutScalingSubs")?.let { systemProperty("fanoutScalingSubs", it) }
     // Show println output from test JVM so the benchmark numbers are
     // actually visible without grepping the report XML.
     testLogging {

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,84 +22,173 @@ 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
-   to a small author list. Today every published EVENT walks every
-   such subscription to learn that. A `HashMap<HexKey, MutableList<Sub>>`
-   keyed by author would cut this to O(1) average for the dominant case.
-2. **Pseudo-broadcast filters** (`{kinds: [1]}` with no other
-   constraint) match almost everything. There's no avoiding the
-   per-subscriber notification, but at least the index lookup is
-   cheap.
+1. **`LocalCache.observables`** in `amethyst/.../LocalCache.kt` —
+   a `ConcurrentHashMap<Observable, Observable>` of feed observers.
+   `refreshNewNoteObservers` iterates every observer for every
+   accepted event; each observer's `new()` runs `filter.match`.
+2. **`EventStoreProjection`** under `quartz/.../cache/projection/` —
+   each projection collects every `StoreChange.Insert` from
+   `ObservableEventStore.changes` and runs its own filter list.
 
-`LoadBenchmark.fanoutLatency` already measures this — current
-results are not yet noted in tree, but back-of-envelope says fanout
-becomes the dominant cost above ~2k subscribers.
+All three follow the pattern "many filter-bearing observers, one
+incoming event — find which observers match". Today that's a per-event
+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>>()
-private val byKind   = ConcurrentHashMap<Int, MutableSet<Subscription>>()
-private val byTag    = ConcurrentHashMap<TagKey, MutableSet<Subscription>>()
-private val unindexed = CopyOnWriteArraySet<Subscription>()  // subs with no
-                                                             // narrowing field
+class FilterIndex<S : Any> {
+    fun register(filter: Filter, subscriber: S)
+    fun register(filters: List<Filter>, subscriber: S)
+    fun registerUnindexed(subscriber: S)            // predicate-only callers
+    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
-filters + the existing `EventMessage` send callback) into whichever
-buckets each filter narrows on. A filter with `kinds=[1] and
-authors=[a,b]` registers into `byKind[1]` AND `byAuthor[a]`,
-`byAuthor[b]` — broadcast unions the resulting candidate sets.
+State is held in a single `AtomicReference<State<S>>` (BanStore-style)
+so reads in the hot path are wait-free. Writes copy-on-write; that's
+fine because writes are subscription-rate (rare) while reads are
+event-rate (frequent).
 
-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]`,
-   `byKind[event.kind]`, every `byTag[(letter, value)]` for the
-   event's single-letter tags, plus `unindexed`.
-2. Run the existing `Filter.match` on each candidate to handle
-   negative constraints (`since`, `until`, `limit` already-reached,
-   composite predicates).
-3. Send.
+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 the unindexed pool.
 
-Expected: **>10× speedup** on fanout for realistic subscriptions.
-Worst case (all filters in `unindexed`) degrades to current behaviour.
+Multi-filter registrations OR the per-filter selections together,
+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,
-reusable by any relay embed. `RelaySession.handleReq` registers/
-unregisters; `LiveEventStore.insert` calls
-`index.candidatesFor(event)`.
+`LiveEventStore.kt` no longer uses a `MutableSharedFlow`. Instead:
 
-## 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` (added):
 
 - N connections, each subscribes to `{authors: [pk_i], kinds: [1]}`.
-- Publish 10k EVENTs from a producer connection; each event matches
+- Publish events round-robin across the N pubkeys; each event matches
   exactly one subscriber.
 - Measure end-to-end latency p50/p99 for N ∈ {100, 1000, 5000}.
 
-Without the index, p99 grows roughly linearly with N. With the
-index, p99 should be flat up to a much higher N.
+The benchmark adapts the per-N event count downward to stay below
+geode's `WebSocketSessionPump.MAX_OUTGOING_BUFFER` (8192 frames per
+session). The single-WS test subClient can't drain (subs + events)
+frames at full firehose rate above ~5k subs in a shared test JVM —
+that's a test-infra ceiling, not a relay one. Override with
+`-DfanoutScalingEvents=N` to push past the default.
+
+Measured numbers from a development laptop (Linux, JDK 21, full
+sweep):
+
+| N subs | events | mean (ms) | p50 (ms) | p99 (ms) | p999 (ms) |
+|-------:|-------:|----------:|---------:|---------:|----------:|
+|    100 | 2 000  | 1.63      | 1.35     | 5.71     | 20.49     |
+|  1 000 | 2 000  | 1.12      | 1.05     | 3.05     |  9.40     |
+|  5 000 | 1 000  | 1.07      | 1.01     | 2.96     |  7.38     |
+
+p50 stays at ~1 ms across all three N values — the index is producing
+the predicted O(1)-per-event scaling. The minor p99 spread comes from
+GC pauses and OkHttp scheduler jitter on the test client, not from
+linear per-event work in the relay.
+
+For the LocalCache side, a similar benchmark would publish events
+matching one of M observers and measure dispatch cost as M grows.
+Not yet added — Phase 2 candidate for a follow-up.
 
 ## 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
-  finely locked; benchmark this path explicitly.
+  COW on a single `AtomicReference` makes each write a full inner-map
+  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
-  union saturates.
+  buckets. The single-dimension-per-filter selection caps how many
+  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.
-- **Correctness fence**: the index must see new subscriptions before
-  the next EVENT broadcast. Today `RelaySession.handleReq` writes its
-  `Job` into a `LargeCache` then launches the collector. Order of
-  operations needs to be revisited so the index is updated atomically
-  with the collector being ready.
+  At 5k subs × average 1 dimension × a handful of values per filter,
+  ~10k–20k entries — negligible.
+- **Correctness fence**: `register` happens before historical replay
+  on the relay side, and inside the `callbackFlow`'s `register` /
+  `awaitClose { unregister }` pair on the client side. Events
+  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.

geode/src/test/kotlin/com/vitorpamplona/geode/perf/LoadBenchmark.kt

@@ -22,6 +22,7 @@ package com.vitorpamplona.geode.perf
 
 import com.vitorpamplona.geode.LocalRelayServer
 import com.vitorpamplona.geode.Relay
+import com.vitorpamplona.quartz.nip01Core.core.toHexKey
 import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
 import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient
 import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.publishAndConfirm
@@ -40,6 +41,7 @@ import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.withTimeout
 import okhttp3.OkHttpClient
 import java.util.concurrent.atomic.AtomicLong
+import java.util.concurrent.atomic.AtomicLongArray
 import kotlin.test.Test
 import kotlin.time.measureTime
 
@@ -496,6 +498,168 @@ class LoadBenchmark {
             }
         }
 
+    /**
+     * Selective fanout: each event matches exactly ONE of N
+     * subscribers. Stresses the inverted-index path in
+     * `FilterIndex` / `LiveEventStore`: without an index, the relay
+     * walks all N subscribers per event and runs `Filter.match` to
+     * find the one true match; with the index, an author-keyed
+     * lookup returns the single subscriber directly.
+     *
+     * Different from [fanoutLatency], which tests broadcast fanout
+     * (one event reaches every subscriber). Here per-event work is
+     * lookup-bound rather than delivery-bound, so latency should be
+     * roughly **flat** in N once the index is wired up — that's the
+     * shape change the index is meant to deliver.
+     *
+     * Configurable via system properties:
+     *  - `fanoutScalingSubs`: comma-separated list of N values
+     *    (default `100,1000,5000`).
+     *  - `fanoutScalingEvents`: total events per N (default scales
+     *    inversely with N to keep test wall time bounded — at 5000
+     *    subs the OkHttp WebSocket dispatcher on the test client
+     *    starts serializing inbound EVENT frames at high throughput,
+     *    which inflates measured latency without telling us anything
+     *    about the index itself).
+     */
+    @Test
+    fun fanoutScaling() =
+        benchmark("fanout scaling") {
+            val subSweep =
+                System
+                    .getProperty("fanoutScalingSubs")
+                    ?.split(",")
+                    ?.mapNotNull { it.trim().toIntOrNull() }
+                    ?.takeIf { it.isNotEmpty() }
+                    ?: listOf(100, 1_000, 5_000)
+            for (subs in subSweep) {
+                runBenchmarkServer { server, http ->
+                    val scope = CoroutineScope(Dispatchers.Default + SupervisorJob())
+                    val subClient = NostrClient(BasicOkHttpWebSocket.Builder { _ -> http }, scope)
+                    val pubClient = NostrClient(BasicOkHttpWebSocket.Builder { _ -> http }, scope)
+                    try {
+                        val relayUrl = server.url.normalizeRelayUrl()
+
+                        // One keypair per subscriber. Each subscriber
+                        // filters on its own author so the relay can
+                        // route an event to exactly one sub via the
+                        // FilterIndex's byAuthor bucket.
+                        val keys = List(subs) { KeyPair() }
+                        val pubkeys = keys.map { it.pubKey.toHexKey() }
+                        val signers = keys.map { NostrSignerSync(it) }
+
+                        // Per-subscriber arrival timestamp. We
+                        // overwrite per round; the publish path waits
+                        // until the slot is non-zero before moving on.
+                        val arrivalNs = AtomicLongArray(subs)
+                        val received = AtomicLong()
+                        val eosed = AtomicLong()
+
+                        repeat(subs) { i ->
+                            subClient.subscribe(
+                                "scale-$i",
+                                mapOf(
+                                    relayUrl to
+                                        listOf(
+                                            Filter(
+                                                authors = listOf(pubkeys[i]),
+                                                kinds = listOf(1),
+                                            ),
+                                        ),
+                                ),
+                                object : SubscriptionListener {
+                                    override fun onEvent(
+                                        event: com.vitorpamplona.quartz.nip01Core.core.Event,
+                                        isLive: Boolean,
+                                        relay: NormalizedRelayUrl,
+                                        forFilters: List<Filter>?,
+                                    ) {
+                                        arrivalNs.set(i, System.nanoTime())
+                                        received.incrementAndGet()
+                                    }
+
+                                    override fun onEose(
+                                        relay: NormalizedRelayUrl,
+                                        forFilters: List<Filter>?,
+                                    ) {
+                                        eosed.incrementAndGet()
+                                    }
+                                },
+                            )
+                        }
+
+                        runBlocking {
+                            withTimeout(120_000) {
+                                while (eosed.get() < subs) kotlinx.coroutines.delay(100)
+                            }
+                        }
+
+                        // Events round-robin over the N pubkeys. Per
+                        // the plan we'd like 10k everywhere, but
+                        // [WebSocketSessionPump.MAX_OUTGOING_BUFFER]
+                        // (8192 frames) on the relay's per-session
+                        // outbound queue trips when the subClient's
+                        // single-WS read pipeline can't drain
+                        // (subs + events) frames fast enough. That's
+                        // a test-infra ceiling, not the index's. We
+                        // back off events at high N so the latency
+                        // numbers reflect dispatch cost rather than
+                        // backpressure-driven connection teardown.
+                        // Override with -DfanoutScalingEvents=N to
+                        // force a value (caller's responsibility to
+                        // stay below the pump cap).
+                        val totalEvents =
+                            System
+                                .getProperty("fanoutScalingEvents")
+                                ?.toIntOrNull()
+                                ?: when {
+                                    subs <= 200 -> 2_000
+                                    subs <= 1_000 -> 2_000
+                                    subs <= 2_000 -> 1_000
+                                    else -> 1_000
+                                }
+                        println("$subs subs ready; publishing $totalEvents events round-robin...")
+
+                        val latenciesMs = DoubleArray(totalEvents)
+
+                        runBlocking {
+                            val start = System.nanoTime()
+                            for (i in 0 until totalEvents) {
+                                val target = i % subs
+                                val event = signers[target].sign(TextNoteEvent.build("scale-$i"))
+                                arrivalNs.set(target, 0)
+                                val pubStart = System.nanoTime()
+                                pubClient.publishAndConfirm(event, setOf(relayUrl))
+                                withTimeout(30_000) {
+                                    while (arrivalNs.get(target) == 0L) kotlinx.coroutines.delay(1)
+                                }
+                                latenciesMs[i] = (arrivalNs.get(target) - pubStart) / 1_000_000.0
+                            }
+                            val totalMs = (System.nanoTime() - start) / 1_000_000.0
+
+                            val sorted = latenciesMs.copyOf().also { it.sort() }
+                            val p50 = sorted[sorted.size / 2]
+                            val p99 = sorted[(sorted.size * 99) / 100]
+                            val p999 = sorted[(sorted.size * 999) / 1000]
+                            val mean = sorted.average()
+                            println(
+                                "subs=$subs events=$totalEvents totalMs=${"%.0f".format(totalMs)} " +
+                                    "meanMs=${"%.2f".format(mean)} " +
+                                    "p50Ms=${"%.2f".format(p50)} " +
+                                    "p99Ms=${"%.2f".format(p99)} " +
+                                    "p999Ms=${"%.2f".format(p999)} " +
+                                    "received=${received.get()}/$totalEvents",
+                            )
+                        }
+                    } finally {
+                        subClient.disconnect()
+                        pubClient.disconnect()
+                        scope.cancel()
+                    }
+                }
+            }
+        }
+
     /**
      * One publisher fires N EVENTs back-to-back without awaiting
      * intermediate OKs, then collects all OKs by event id. This is

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,12 +22,13 @@ 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 com.vitorpamplona.quartz.nip01Core.store.IdAndTime
 import kotlinx.coroutines.CompletableDeferred
-import kotlinx.coroutines.channels.BufferOverflow
-import kotlinx.coroutines.flow.MutableSharedFlow
-import kotlinx.coroutines.flow.onSubscription
+import kotlinx.coroutines.awaitCancellation
+import kotlin.concurrent.atomics.AtomicReference
+import kotlin.concurrent.atomics.ExperimentalAtomicApi
 
 /**
  * A reactive event store that combines historical data retrieval with live event streaming.
@@ -37,25 +38,40 @@ 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 [submit] / [insert] to interested subscribers is index-driven via
+ * [FilterIndex]: each [query] registers its filters; on accepted ingest the index returns
+ * the (much smaller) candidate set and we deliver only to those whose filters actually
+ * match. This avoids the quadratic O(N_subscribers × N_filters_per_sub) per-event walk
+ * that a SharedFlow-based broadcast would do.
+ *
  * @property store The underlying persistent storage for events.
+ * @property ingest The group-commit writer pipeline. Accepted events fan out via the
+ *  index; rejected events do not.
  */
+@OptIn(ExperimentalAtomicApi::class)
 class LiveEventStore(
     private val store: IEventStore,
     private val ingest: IngestQueue,
 ) {
-    private val newEventStream =
-        MutableSharedFlow<Event>(
-            replay = 0,
-            extraBufferCapacity = 100, // Optional: adjust for backpressure
-            onBufferOverflow = BufferOverflow.DROP_LATEST, // Default behavior
-        )
+    private val index = FilterIndex<LiveSubscription>()
+
+    /**
+     * One live REQ subscription. Carries the filters (for the
+     * 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,
+    )
 
     /**
      * Fire-and-forget enqueue: hand [event] to the [IngestQueue] and
      * fire [onComplete] once the writer's batch has a per-row
-     * decision. On `Accepted` the live stream is also emitted to so
-     * subscribers see the event. Suspends only when the ingest queue
-     * is full (backpressure).
+     * decision. On `Accepted` the live stream is also fanned out via
+     * [FilterIndex] so subscribers see the event. Suspends only when
+     * the ingest queue is full (backpressure).
      */
     suspend fun submit(
         event: Event,
@@ -63,7 +79,7 @@ class LiveEventStore(
     ) {
         ingest.submit(event) { outcome ->
             if (outcome is IEventStore.InsertOutcome.Accepted) {
-                newEventStream.tryEmit(event)
+                fanout(event)
             }
             onComplete(outcome)
         }
@@ -93,47 +109,86 @@ class LiveEventStore(
         done.await()
     }
 
+    /**
+     * Live fanout for an accepted event. The index returns a
+     * super-set; `Filter.match` enforces negative constraints. Each
+     * candidate's `deliver` is expected to be cheap (typically a
+     * `trySend` to a per-connection outbound queue) — this runs on
+     * the [IngestQueue] drain coroutine, so a slow sub blocks the
+     * batch writer.
+     */
+    private fun fanout(event: Event) {
+        for (sub in index.candidatesFor(event)) {
+            if (sub.filters.any { it.match(event) }) {
+                sub.deliver(event)
+            }
+        }
+    }
+
     suspend fun query(
         filters: List<Filter>,
         onEach: (Event) -> Unit,
         onEose: () -> Unit,
     ) {
-        // Order matters: register the live collector BEFORE replaying
-        // stored events and signalling EOSE. Otherwise an event emitted
-        // between EOSE and `collect` is lost because [newEventStream] has
-        // replay=0. The race is only occasionally visible for kinds the
-        // store persists (insert latency masks it) but fires reliably for
-        // ephemeral kinds (20000-29999) where insert is a no-op — and
-        // ephemeral events MUST still reach matching live subscribers per
-        // NIP-01.
+        // During the historical replay, record ids the store has
+        // emitted so the live path can dedupe. The index registers
+        // *before* the replay starts (otherwise an event accepted
+        // mid-replay would slip past the live path entirely — same
+        // race the previous SharedFlow-based implementation closed
+        // with `onSubscription`).
         //
-        // Side effect of registering the collector first: an event
-        // inserted *during* `store.query` will be both replayed by the
-        // store AND emitted to the live stream. We dedupe by tracking
-        // ids seen during the historical replay and skipping them on
-        // the live path. The set is dropped after EOSE so live-only
-        // events don't accumulate memory.
-        var inHistoricalPhase = true
-        var seenIds: HashSet<String>? = HashSet()
-        val historicalOnEach: (Event) -> Unit = { event ->
-            seenIds?.add(event.id)
-            onEach(event)
-        }
-        newEventStream
-            .onSubscription {
-                store.query(filters, historicalOnEach)
-                onEose()
-                // Free the dedupe set once we've crossed EOSE: from
-                // here on the live stream is the only source of
-                // events, so duplicates aren't possible.
-                inHistoricalPhase = false
-                seenIds = null
-            }.collect { newEvent ->
-                if (inHistoricalPhase && seenIds?.contains(newEvent.id) == true) return@collect
-                if (filters.any { it.match(newEvent) }) {
-                    onEach(newEvent)
+        // The set is read from the [IngestQueue] drain coroutine (in
+        // `deliver`, called synchronously from `fanout`) and written
+        // from this coroutine (the historical-replay closure below).
+        // It must be a persistent / immutable Set under an
+        // AtomicReference — wrapping a mutable HashSet would race
+        // because AtomicReference only protects the reference, not
+        // the set's internal state. Each `add` is a CAS-loop that
+        // publishes a new immutable Set; `deliver`'s `load()` always
+        // sees a fully-constructed snapshot.
+        //
+        // Once cleared to null after EOSE, `deliver` short-circuits
+        // and every live event is forwarded.
+        val seenIds = AtomicReference<Set<String>?>(emptySet())
+
+        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 ->
+                // CAS-loop on an immutable Set. The historical
+                // replay is single-writer in steady state, so the
+                // CAS typically succeeds first try; the loop only
+                // matters if we lose a race on `seenIds.store(null)`
+                // (post-EOSE), in which case `current == null` and
+                // we exit cleanly.
+                while (true) {
+                    val current = seenIds.load() ?: break
+                    if (event.id in current) break
+                    if (seenIds.compareAndSet(current, current + event.id)) break
                 }
+                onEach(event)
             }
+            onEose()
+            // Drop the dedupe set so the live path stops paying for
+            // it. From this point the index drives delivery and
+            // duplicates are no longer possible.
+            seenIds.store(null)
+            // Suspend until the caller's coroutine is cancelled
+            // (e.g. NIP-01 CLOSE or connection drop). The `finally`
+            // unregisters from the index.
+            awaitCancellation()
+        } finally {
+            index.unregister(sub)
+        }
     }
 
     suspend fun count(filters: List<Filter>) = store.count(filters)

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

@@ -0,0 +1,308 @@
+/*
+ * 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 tagsAllFilterMatchesEventsCarryingAllTagValues() {
+        // tagsAll keys the index on the first single-letter entry
+        // exactly like `tags`. Events that match the index lookup
+        // still have to pass `Filter.match` for the AND-of-values
+        // semantics — the index is a super-set.
+        val index = FilterIndex<Sub>()
+        val s = Sub("s")
+        index.register(Filter(tagsAll = mapOf("p" to listOf(pTag1))), s)
+
+        assertTrue(s in index.candidatesFor(event(tags = arrayOf(arrayOf("p", pTag1)))))
+        assertFalse(s in index.candidatesFor(event(tags = arrayOf(arrayOf("p", pTag2)))))
+    }
+
+    @Test
+    fun multiCharTagKeyFallsThroughToKindThenUnindexed() {
+        // The index only buckets single-letter tag keys (NIP-12).
+        // A filter with only multi-char tag keys should fall through
+        // to the next dimension.
+        val index = FilterIndex<Sub>()
+        val withKind = Sub("withKind")
+        val withoutKind = Sub("withoutKind")
+
+        index.register(Filter(tags = mapOf("alt" to listOf("foo")), kinds = listOf(7)), withKind)
+        index.register(Filter(tags = mapOf("alt" to listOf("foo"))), withoutKind)
+
+        // withKind goes to KindKey(7); withoutKind to Unindexed.
+        assertTrue(withKind in index.candidatesFor(event(kind = 7)))
+        assertFalse(withKind in index.candidatesFor(event(kind = 1)))
+        assertTrue(withoutKind in index.candidatesFor(event(kind = 1)))
+        assertTrue(withoutKind in index.candidatesFor(event(kind = 7)))
+    }
+
+    @Test
+    fun reRegisterAddsAdditionalKeysWithoutDuplicating() {
+        // Calling register twice on the same subscriber unions the
+        // bucket assignments — useful when an observer wants to
+        // listen on multiple narrowing dimensions accumulated over
+        // time. The bucket sets must dedupe so the subscriber
+        // appears once in `candidatesFor`.
+        val index = FilterIndex<Sub>()
+        val s = Sub("s")
+        index.register(Filter(authors = listOf(authorA)), s)
+        index.register(Filter(kinds = listOf(7)), s)
+
+        // Author hit + kind hit = same subscriber, returned once.
+        val cands = index.candidatesFor(event(pubkey = authorA, kind = 7))
+        assertEquals(1, cands.size)
+        assertTrue(s in cands)
+
+        // Unregister cleans both dimensions.
+        index.unregister(s)
+        assertTrue(index.candidatesFor(event(pubkey = authorA)).isEmpty())
+        assertTrue(index.candidatesFor(event(kind = 7)).isEmpty())
+    }
+
+    @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())
+    }
+}