Merge pull request #3663 from vitorpamplona/claude/sqlite-query-performance-bh2j27

SQLite query scaling: FTS contentless, tag-merge, pooled statements

README.md

@@ -17,7 +17,7 @@ Join the social network you control.
 [![Maven Central](https://img.shields.io/maven-central/v/com.vitorpamplona.quartz/quartz?label=Quartz%20%28Maven%20Central%29&labelColor=27303D&color=0877d2)](https://central.sonatype.com/artifact/com.vitorpamplona.quartz/quartz)
 [![JitPack snapshots](https://img.shields.io/badge/Quartz%20snapshots-JitPack-27303D?labelColor=27303D&color=0877d2)](https://jitpack.io/#vitorpamplona/amethyst)
 [![CI](https://img.shields.io/github/actions/workflow/status/vitorpamplona/amethyst/build.yml?labelColor=27303D)](https://github.com/vitorpamplona/amethyst/actions/workflows/build.yml)
-[![License: Apache-2.0](https://img.shields.io/github/license/vitorpamplona/amethyst?labelColor=27303D&color=0877d2)](/LICENSE)
+[![License: MIT](https://img.shields.io/github/license/vitorpamplona/amethyst?labelColor=27303D&color=0877d2)](/LICENSE)
 [![Ask DeepWiki](https://deepwiki.com/badge.svg)](https://deepwiki.com/vitorpamplona/amethyst)
 
 ## Download and Install

SKILL.md

@@ -24,7 +24,9 @@ Build customized Amethyst Nostr clients for Android. Fork, rebrand, customize, a
 
 2. **Android SDK**
    - Command-line tools from https://developer.android.com/studio#command-line-tools-only
-   - Required components: build-tools, platform-tools, platforms;android-35
+   - Required components: build-tools, platform-tools, platforms;android-37
+   - The exact SDK level is `android-compileSdk` in `gradle/libs.versions.toml` —
+     check there if this number has drifted.
 
 3. **Git** for cloning the repository
 
@@ -66,48 +68,54 @@ keyPassword=your-password
 
 ### 3. Configure Signing
 
-Add to `amethyst/build.gradle` inside the `android {}` block:
+Add to `amethyst/build.gradle.kts` inside the `android {}` block:
 
-```gradle
-def keystorePropertiesFile = rootProject.file("keystore.properties")
-def keystoreProperties = new Properties()
+```kotlin
+val keystorePropertiesFile = rootProject.file("keystore.properties")
+val keystoreProperties = Properties()
 if (keystorePropertiesFile.exists()) {
-    keystoreProperties.load(new FileInputStream(keystorePropertiesFile))
+    keystorePropertiesFile.inputStream().use { keystoreProperties.load(it) }
 }
 
 signingConfigs {
-    release {
+    create("release") {
         if (keystorePropertiesFile.exists()) {
-            storeFile rootProject.file(keystoreProperties['storeFile'])
-            storePassword keystoreProperties['storePassword']
-            keyAlias keystoreProperties['keyAlias']
-            keyPassword keystoreProperties['keyPassword']
+            storeFile = rootProject.file(keystoreProperties["storeFile"] as String)
+            storePassword = keystoreProperties["storePassword"] as String
+            keyAlias = keystoreProperties["keyAlias"] as String
+            keyPassword = keystoreProperties["keyPassword"] as String
         }
     }
 }
 ```
 
+This needs `import java.util.Properties` at the top of the file.
+
 Update the release buildType to use the signing config:
-```gradle
+```kotlin
 buildTypes {
-    release {
-        signingConfig signingConfigs.release
+    getByName("release") {
+        signingConfig = signingConfigs.getByName("release")
         // ... existing config
     }
 }
 ```
 
+Verify with `./gradlew :amethyst:signingReport` — the release variants should
+report your keystore rather than `~/.android/debug.keystore`.
+
 ### 4. Disable Google Services (Required for F-Droid)
 
 **⚠️ CRITICAL:** The Google Services plugin fails when you change the package name. For F-Droid builds, disable it.
 
-Edit `amethyst/build.gradle`, comment out the plugin:
-```gradle
+Edit `amethyst/build.gradle.kts`, comment out the plugin:
+```kotlin
 plugins {
     alias(libs.plugins.androidApplication)
-    alias(libs.plugins.jetbrainsKotlinAndroid)
     // alias(libs.plugins.googleServices)  // DISABLED for F-Droid
     alias(libs.plugins.jetbrainsComposeCompiler)
+    alias(libs.plugins.serialization)
+    alias(libs.plugins.googleKsp)
 }
 ```
 
@@ -141,8 +149,8 @@ Edit `amethyst/src/main/res/values/strings.xml`:
 
 ### Change Package ID
 
-Edit `amethyst/build.gradle`:
-```gradle
+Edit `amethyst/build.gradle.kts`:
+```kotlin
 android {
     defaultConfig {
         applicationId = "com.yourcompany.yourapp"
@@ -152,8 +160,8 @@ android {
 
 ### Change Project Name
 
-Edit `settings.gradle`:
-```gradle
+Edit `settings.gradle.kts`:
+```kotlin
 rootProject.name = "YourAppName"
 ```
 
@@ -167,36 +175,28 @@ Replace icon files in:
 
 Make your app identify itself on posts with `["client", "YourAppName"]`.
 
-**1. Create tag builder extension:**
+You do **not** need to add the tag per event type. The client tag is applied
+centrally by `NostrSignerWithClientTag`, a signer decorator that appends the tag
+to everything it signs (and respects the user's "add client tag" privacy
+setting). Changing the name is a one-constant edit:
 
-Create `quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/tags/clientTag/TagArrayBuilderExt.kt`:
+Edit `amethyst/src/main/java/com/vitorpamplona/amethyst/model/accountsCache/AccountCacheState.kt`:
 ```kotlin
-package com.vitorpamplona.quartz.nip01Core.tags.clientTag
-
-import com.vitorpamplona.quartz.nip01Core.core.Event
-import com.vitorpamplona.quartz.nip01Core.core.TagArrayBuilder
-
-fun <T : Event> TagArrayBuilder<T>.client(clientName: String) = 
-    addUnique(arrayOf(ClientTag.TAG_NAME, clientName))
+const val CLIENT_TAG_NAME = "YourAppName"
 ```
 
-**2. Add to TextNoteEvent:**
+That constant is passed to `NostrSignerWithClientTag` when the account's signer
+is built, so every signed event carries your name.
 
-Edit `quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip10Notes/TextNoteEvent.kt`:
-
-Add import:
-```kotlin
-import com.vitorpamplona.quartz.nip01Core.tags.clientTag.client
-```
-
-In both `build()` functions, add after `alt(...)`:
-```kotlin
-client("YourAppName")
-```
+The tag itself lives in
+`quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip89AppHandlers/clientTag/`
+(`ClientTag`, `TagArrayBuilderExt`, `NostrSignerWithClientTag`) — you only need to
+touch it if you want the optional NIP-89 handler address / relay hint variants.
 
 ### Modify Default Relays
 
-Edit relay configuration in `quartz/src/main/java/com/vitorpamplona/quartz/nip01Core/relay/` or the UI settings files.
+Edit `commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/defaults/Constants.kt`
+(see also `AmethystDefaults.kt` and `DefaultDmIndexerRelays.kt` in the same folder).
 
 ## Troubleshooting
 

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

@@ -41,6 +41,7 @@ import com.vitorpamplona.amethyst.ui.actions.mediaServers.ServerName
 import com.vitorpamplona.quartz.concord.cord02Community.ConcordCommunityListEvent
 import com.vitorpamplona.quartz.experimental.ephemChat.list.EphemeralChatListEvent
 import com.vitorpamplona.quartz.experimental.nipA3.PaymentTargetsEvent
+import com.vitorpamplona.quartz.marmot.mip00KeyPackages.KeyPackageRelayListEvent
 import com.vitorpamplona.quartz.nip01Core.core.Event
 import com.vitorpamplona.quartz.nip01Core.core.HexKey
 import com.vitorpamplona.quartz.nip01Core.core.JsonMapper
@@ -56,6 +57,7 @@ import com.vitorpamplona.quartz.nip28PublicChat.list.ChannelListEvent
 import com.vitorpamplona.quartz.nip37Drafts.privateOutbox.PrivateOutboxRelayListEvent
 import com.vitorpamplona.quartz.nip47WalletConnect.Nip47WalletConnect
 import com.vitorpamplona.quartz.nip50Search.SearchRelayListEvent
+import com.vitorpamplona.quartz.nip51Lists.favoriteAlgoFeedsList.FavoriteAlgoFeedsListEvent
 import com.vitorpamplona.quartz.nip51Lists.geohashList.GeohashListEvent
 import com.vitorpamplona.quartz.nip51Lists.hashtagList.HashtagListEvent
 import com.vitorpamplona.quartz.nip51Lists.muteList.MuteListEvent
@@ -171,6 +173,8 @@ private object PrefKeys {
     const val LATEST_RELAY_GROUP_LIST = "latestRelayGroupList"
     const val LATEST_CONCORD_LIST = "latestConcordList"
     const val LATEST_TRUST_PROVIDER_LIST = "latestTrustProviderList"
+    const val LATEST_KEY_PACKAGE_RELAY_LIST = "latestKeyPackageRelayList"
+    const val LATEST_FAVORITE_ALGO_FEEDS_LIST = "latestFavoriteAlgoFeedsList"
     const val CALLS_ENABLED = "calls_enabled"
     const val HIDE_DELETE_REQUEST_DIALOG = "hide_delete_request_dialog"
     const val HIDE_BLOCK_ALERT_DIALOG = "hide_block_alert_dialog"
@@ -581,6 +585,8 @@ object LocalPreferences {
                     putOrRemove(PrefKeys.LATEST_RELAY_GROUP_LIST, settings.backupRelayGroupList)
                     putOrRemove(PrefKeys.LATEST_CONCORD_LIST, settings.backupConcordList)
                     putOrRemove(PrefKeys.LATEST_TRUST_PROVIDER_LIST, settings.backupTrustProviderList)
+                    putOrRemove(PrefKeys.LATEST_KEY_PACKAGE_RELAY_LIST, settings.backupKeyPackageRelayList)
+                    putOrRemove(PrefKeys.LATEST_FAVORITE_ALGO_FEEDS_LIST, settings.backupFavoriteAlgoFeedsList)
                     putOrRemove(PrefKeys.LATEST_PAYMENT_TARGETS, settings.backupNipA3PaymentTargets)
                     putOrRemove(PrefKeys.LATEST_CASHU_WALLET, settings.backupCashuWallet)
                     putOrRemove(PrefKeys.LATEST_NUTZAP_INFO, settings.backupNutzapInfo)
@@ -768,6 +774,8 @@ object LocalPreferences {
                     val latestRelayGroupListStr = getString(PrefKeys.LATEST_RELAY_GROUP_LIST, null)
                     val latestConcordListStr = getString(PrefKeys.LATEST_CONCORD_LIST, null)
                     val latestTrustProviderListStr = getString(PrefKeys.LATEST_TRUST_PROVIDER_LIST, null)
+                    val latestKeyPackageRelayListStr = getString(PrefKeys.LATEST_KEY_PACKAGE_RELAY_LIST, null)
+                    val latestFavoriteAlgoFeedsListStr = getString(PrefKeys.LATEST_FAVORITE_ALGO_FEEDS_LIST, null)
                     val latestPaymentTargetsStr = getString(PrefKeys.LATEST_PAYMENT_TARGETS, null)
                     val latestCashuWalletStr = getString(PrefKeys.LATEST_CASHU_WALLET, null)
                     val latestNutzapInfoStr = getString(PrefKeys.LATEST_NUTZAP_INFO, null)
@@ -829,6 +837,8 @@ object LocalPreferences {
                     val latestRelayGroupList = async { parseEventOrNull<SimpleGroupListEvent>(latestRelayGroupListStr) }
                     val latestConcordList = async { parseEventOrNull<ConcordCommunityListEvent>(latestConcordListStr) }
                     val latestTrustProviderList = async { parseEventOrNull<TrustProviderListEvent>(latestTrustProviderListStr) }
+                    val latestKeyPackageRelayList = async { parseEventOrNull<KeyPackageRelayListEvent>(latestKeyPackageRelayListStr) }
+                    val latestFavoriteAlgoFeedsList = async { parseEventOrNull<FavoriteAlgoFeedsListEvent>(latestFavoriteAlgoFeedsListStr) }
                     val latestPaymentTargets = async { parseEventOrNull<PaymentTargetsEvent>(latestPaymentTargetsStr) }
                     val latestCashuWallet =
                         async {
@@ -882,6 +892,8 @@ object LocalPreferences {
                     val latestRelayGroupListResolved = latestRelayGroupList.await()
                     val latestConcordListResolved = latestConcordList.await()
                     val latestTrustProviderListResolved = latestTrustProviderList.await()
+                    val latestKeyPackageRelayListResolved = latestKeyPackageRelayList.await()
+                    val latestFavoriteAlgoFeedsListResolved = latestFavoriteAlgoFeedsList.await()
                     val latestPaymentTargetsResolved = latestPaymentTargets.await()
                     val latestCashuWalletResolved = latestCashuWallet.await()
                     val latestNutzapInfoResolved = latestNutzapInfo.await()
@@ -977,6 +989,8 @@ object LocalPreferences {
                         backupRelayGroupList = latestRelayGroupListResolved,
                         backupConcordList = latestConcordListResolved,
                         backupTrustProviderList = latestTrustProviderListResolved,
+                        backupKeyPackageRelayList = latestKeyPackageRelayListResolved,
+                        backupFavoriteAlgoFeedsList = latestFavoriteAlgoFeedsListResolved,
                         lastReadPerRoute = MutableStateFlow(lastReadPerRouteResolved),
                         hasDonatedInVersion = MutableStateFlow(hasDonatedInVersion),
                         dismissedPollNoteIds = MutableStateFlow(dismissedPollNoteIds),

desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/relay/LocalRelayStore.kt

@@ -26,6 +26,7 @@ import com.vitorpamplona.amethyst.desktop.cache.DesktopLocalCache
 import com.vitorpamplona.quartz.nip01Core.core.Event
 import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
 import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
+import com.vitorpamplona.quartz.nip01Core.store.sqlite.DefaultIndexingStrategy
 import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore
 import com.vitorpamplona.quartz.utils.Log
 import kotlinx.coroutines.CoroutineScope
@@ -42,6 +43,18 @@ class LocalRelayStore(
 ) : AutoCloseable {
     companion object {
         val LOCAL_RELAY_URL: NormalizedRelayUrl = NormalizedRelayUrl("ws://localhost/amethyst-local/")
+
+        /**
+         * Client defaults plus the authors-without-kinds index: shared
+         * ViewModels (`Nip65RelayListViewModel`, `PrivateOutboxRelayListViewModel`,
+         * `VanishRequestsState`) replay `authors`-only filters against this
+         * store, which full-scan without `(pubkey, created_at)` — the
+         * `(kind, pubkey, …)` index can't serve them, pubkey is its second
+         * column. A personal store is small, so the extra insert cost is
+         * negligible; existing DBs get the index built on next open via
+         * `ensureOptionalIndexes`.
+         */
+        val INDEX_STRATEGY = DefaultIndexingStrategy(indexEventsByPubkeyAlone = true)
     }
 
     private fun dbDir(pubKeyHex: String): File = File(homeDir, ".amethyst/accounts/${pubKeyHex.take(8)}")
@@ -87,14 +100,14 @@ class LocalRelayStore(
             dir.mkdirs()
             val path = File(dir, "events.db").absolutePath
             try {
-                store = EventStore(dbName = path, relay = LOCAL_RELAY_URL)
+                store = EventStore(dbName = path, relay = LOCAL_RELAY_URL, indexStrategy = INDEX_STRATEGY)
                 _lastError.value = null
                 refreshStats()
             } catch (e: Exception) {
                 Log.w("LocalRelayStore") { "DB open failed, recreating: ${e.message}" }
                 try {
                     deleteDbFiles(path)
-                    store = EventStore(dbName = path, relay = LOCAL_RELAY_URL)
+                    store = EventStore(dbName = path, relay = LOCAL_RELAY_URL, indexStrategy = INDEX_STRATEGY)
                     _lastError.value = "Database was recreated: ${e.message}"
                 } catch (e2: Exception) {
                     _lastError.value = "Cannot open local store: ${e2.message}"

geode/src/main/kotlin/com/vitorpamplona/geode/RelayIndexingStrategy.kt

@@ -57,6 +57,14 @@ fun relayIndexingStrategy(
     // index unconditionally; without it the filter walks the whole
     // time index.
     indexEventsByPubkeyAlone = true,
+    // The tag ∩ author ∩ kind shape (DM rooms, reports-by-follows,
+    // follows-scoped community feeds — 65 client assembler call sites)
+    // otherwise reads every row for the tag/kind before filtering the
+    // author. TagAuthorIndexBenchmark @ 1M events: 14.2 ms -> 0.66 ms
+    // (~21x, growing with corpus size) with insert cost inside run noise
+    // (49.0 vs 47.4 µs/event). Existing DBs build the index on next open
+    // via ensureOptionalIndexes.
+    indexTagsWithKindAndPubkey = true,
     indexFullTextSearch = fullTextSearch,
     // Tokenize off the commit path; NostrServer drives the catch-up
     // worker and search queries drain it first, so NIP-50 stays

quartz/RELAY.md

@@ -83,6 +83,8 @@ val store = EventStore(
 
 By default, all single-letter tags with values are indexed. Override `shouldIndex(kind, tag)` for custom behavior. More indexes = faster queries but larger database.
 
+Flag flips are safe on existing databases: any flag-gated index the strategy wants but the on-disk schema lacks is created on the next open (idempotent `CREATE INDEX IF NOT EXISTS`, one-time build cost) — no schema version bump involved. Disabling a flag never drops an existing index.
+
 `indexFullTextSearch` defaults to `true` and controls the NIP-50 full-text index (`event_fts`). Set it to `false` when search is served elsewhere (e.g. a Vespa backend, or a `SearchEventSource` as shown below): inserts skip the FTS tokenization cost, no `event_fts` table/trigger is created, and any filter carrying a non-empty `search` term returns no matches.
 
 ## Non-Storage Relays (search, redirector, computed)

quartz/build.gradle.kts

@@ -94,6 +94,8 @@ kotlin {
         // Forward the negentropy-benchmark corpus size to the test JVM.
         System.getProperty("negBenchN")?.let { systemProperty("negBenchN", it) }
         System.getProperty("followBenchScale")?.let { systemProperty("followBenchScale", it) }
+        System.getProperty("tagBenchScale")?.let { systemProperty("tagBenchScale", it) }
+        System.getProperty("fsBenchScale")?.let { systemProperty("fsBenchScale", it) }
         // Opt-in JFR profiling of a benchmark run (-PnegProfile=/tmp/neg.jfr).
         (project.findProperty("negProfile") as? String)?.let {
             jvmArgs("-XX:+FlightRecorder", "-XX:StartFlightRecording=filename=$it,settings=profile,dumponexit=true")

quartz/plans/2026-07-21-sqlite-query-scaling.md

@@ -0,0 +1,168 @@
+# SQLite query scaling: contentless FTS + rowid search, tag-path merge, pooled statements
+
+**Status: shipped.** Follow-up to a scale-curve report (`SQLite vs Vespa`,
+throughput vs corpus size) showing the app-side SQLite store degrading with
+corpus size on several query shapes — NIP-50 search ~18× (25k→400k), batch
+ingest 16×, author-timeline 13×, follow-feed 2.4× — while point reads stayed
+flat. Two of those (author-timeline, follow-feed) were already addressed
+(the pubkey-alone index and `MergeQueryExecutor`) and mostly reflect running
+the report under the *client* `DefaultIndexingStrategy` rather than
+`relayIndexingStrategy()`. This change adds the **large-IN tag watcher** to the
+merge executor, fixes the **FTS delete path** (which degraded with corpus
+size) and shrinks the FTS index, fixes **NIP-50 search ordering** (it was
+sorting by `created_at`, not relevance), and fixes a **statement-cache** miss
+the merge paths hit.
+
+It does **not** fix NIP-50 *search* latency: bm25 relevance scoring (like the
+old `created_at` sort) must visit every match, so search cost still grows with
+the match set (§1). Corpus-independent search is an external-engine job.
+
+Everything here is read/size work; the write path and on-disk index set are
+unchanged except the FTS table, which gets *smaller* and deletes faster.
+
+## 1. FTS — contentless index (fast deletes + smaller), NOT a search-scaling fix
+
+The old index was `fts5(event_header_row_id, content)`: it stored a second
+copy of the tokenized text, its rowid was auto-assigned (unrelated to the
+event), and — the real problem — the `fts_foreign_key` delete trigger deleted
+by the `event_header_row_id` *column*, which FTS5 cannot seek, so it **scanned
+the whole index per delete**.
+
+Changes (`FullTextSearchModule`, `QueryBuilder`, DB version 4→5):
+
+- **rowid = `event_headers.row_id`.** Set explicitly on every insert; the
+  delete trigger, reindex, and catch-up paths key off it. Deletes become an
+  O(log n) primary-key seek instead of an O(n) column scan. This is the win:
+  every event removal fires the trigger — replaceable rotation, kind-5,
+  expiration, right-to-vanish — so on the old schema deletion throughput
+  degraded with corpus size.
+- **Contentless** (`fts5(content, content='', contentless_delete=1)`). The
+  indexed text is *derived* (`SearchableEvent.indexableContent()`, not a raw
+  column), so FTS5 **external-content** — which re-reads the source column from
+  the base table — cannot express it; **contentless** is the correct primitive.
+  It drops the stored content copy (index shrinks) and `contentless_delete=1`
+  keeps the delete trigger working.
+- **Segment compaction.** `reindexAll` finishes with `'optimize'`, and the
+  periodic `SQLiteEventStore.optimize()` (geode's maintenance tick) folds in a
+  bounded `'merge'`, so incremental / deferred-catch-up inserts don't leave the
+  index as many small segments.
+
+Delete cost — `FtsSearchScalingBenchmark.deleteByColumnVsByRowid`, 500 deletes,
+in-memory:
+
+| rows | by column (old) | by rowid (new) |
+|---|---:|---:|
+| 2k | 91.8 ms | 6.6 ms |
+| 8k | 362.0 ms | 4.7 ms |
+
+By-column grows ~linearly with the table (O(n)/delete); by-rowid is flat —
+~78× at 8k rows and widening.
+
+**Search ordering fixed to relevance (NIP-50), which the store was getting
+wrong.** NIP-50 says results are returned "in descending order by quality of
+search result ... not by the usual `.created_at`", with the limit applied after
+the score — but the store sorted search by `created_at DESC` (pre-existing).
+*Every* search filter now orders by FTS5 bm25 (`ORDER BY event_fts.rank`,
+`created_at DESC` as a tie-break): the tag-free shape (`makeSimpleSearch`) and
+`search + tag` (the `prepareRowIDSubQueries`/`makeQueryIn` path, which carries
+the rank column through the row-id subquery via `projectRank` and applies the
+LIMIT by rank). Only the negentropy snapshot keeps `created_at` — a sync set,
+not a ranked result. Verified against stronger-but-older matches outranking
+weaker-but-newer ones, tag-scoped, with the limit cutting by score
+(`SearchRelevanceOrderTest`, `Fts5CapabilityProbe`).
+
+An earlier draft of *this* change instead added a `searchOrderByRowId` flag
+(order by the FTS rowid for O(limit) search); that is *ingestion* order — wrong
+events under a limit once ingestion diverges from time order, and not relevance
+either — so it was removed.
+
+This is a **correctness** fix, not a scaling one: bm25 (like the created_at
+sort) must score every match, so search latency still grows with the match set
+(the report's 18× curve) and `optimize` doesn't change the asymptotics
+(measured within noise at 100k/200k). Corpus-independent search is genuinely an
+external-engine job (the Vespa side of the report), not this index.
+
+Migration (v4→v5): the old rowids can't be remapped, so `event_fts` is dropped
+and rebuilt — synchronous stores rebuild in the upgrade transaction (client
+corpora are small), deferred stores reset the catch-up watermark to 0 and let
+the background worker repopulate (no long migration transaction).
+`ContentlessFtsMigrationTest` fabricates a real v4 DB (old schema + garbage
+rows + `user_version=4`) and asserts the reopen rebuilds search, wipes the
+stale rows, maps rowid→row_id, and keeps the delete trigger working.
+
+## 2. Tag watcher — extend `MergeQueryExecutor` to the tag path
+
+`kinds=[7] AND #e=[hundreds of note ids] LIMIT n` (reactions/replies) had the
+same shape the follow-feed fix already solved for authors: the per-value
+streams come sorted off `(tag_hash[, kind], created_at)`, but their union does
+not, so SQLite collected every matching row and TEMP-B-TREE-sorted to the limit
+— O(matching history), growing with the corpus (`TagAuthorIndexBenchmark`:
+12.8 ms cold at 200k → 14.2 ms at 1M).
+
+`MergeQueryExecutor` now opens one lazy newest-first cursor per `(value[,
+kind])` off `query_by_tags_hash_kind` (or `query_by_tags_hash` when there is no
+kind, gated by `indexTagsByCreatedAtAlone`) and heap-merges to the limit —
+**O(limit + streams)**. Unlike the author path, one event can carry several
+queried tag values, so the tag merge **dedups by event id** through a `seen`
+set (the author path, one pubkey per event, skips it). Eligibility is narrow
+(one non-`d` tag key, ≥2 values, a limit, no ids/authors/d-tag/search, no
+AND-tags); everything else falls through to the single-SQL plan. Counts and
+deletes are unchanged (still single-SQL). `TagMergeCorrectnessTest` pins it
+against a Kotlin reference including cross-stream dedup, since/until windows,
+the raw path, and tie handling (tag cursors order off `event_tags`, which has
+no id column, so same-second ties are a valid newest-N but not id-exact).
+
+## 3. Pooled statement cache — so the merges actually cache
+
+`StatementCachingConnection` kept **one** handle per SQL string. The k-way
+merge opens *many* identical-SQL cursors at once (one per author/tag stream),
+so every stream past the first missed the cache and prepared uncached, and a
+repeated follow-feed / reactions REQ re-prepared all of them each poll. The
+cache now keeps a small **pool** per SQL (bounded by a global cap, default
+raised 256→512), so concurrent same-SQL checkouts reuse cached handles and
+repeated polls reuse their per-stream cursors. `StatementCachingConnectionTest`
+covers sequential reuse, concurrent distinctness/independence, freed-handle
+reuse, and cap overflow → uncached fallback.
+
+## Correctness / scope
+
+- DB version bump 4→5 with a real-upgrade test.
+- New capability test (`Fts5CapabilityProbe`) pins the FTS5 features the
+  contentless index needs (contentless_delete, absent-rowid delete no-op,
+  rowid ordering, merge/optimize) to the bundled SQLite (3.50.1), so a driver
+  downgrade fails loudly.
+- All existing `store.sqlite` tests pass unchanged except `QueryAssemblerTest`,
+  whose asserted EXPLAIN output updated for the new join column
+  (`event_fts.rowid`) and the contentless table's virtual-index marker
+  (`0:M2`→`0:M1`).
+- Search ordering: bm25 relevance for every search REQ shape — tag-free,
+  `search + tag`, and multi-filter all-search (e.g. the client's
+  search-across-kinds, unioned then deduped by event keeping the best score).
+  Mixed search/non-search multi-filter REQs stay `created_at` (a non-search
+  branch has no defined relevance).
+
+## Audit follow-ups (post-review hardening)
+
+A two-reviewer adversarial audit of the branch found no correctness/data-loss/
+crash bugs; it produced these robustness fixes and one gap-closure:
+
+- **Multi-filter search ordering** (gap): a REQ of several filters that all
+  carry a search term was `created_at`-ordered (the union path). Now
+  relevance-ordered via `unionSubqueriesIfNeeded(projectRank)` — each branch
+  projects `rank`, `UNION ALL` + `GROUP BY row_id MIN(rank)` dedups across
+  branches keeping the best score. Only when every branch is a search branch
+  (and FTS is on); count/delete unions stay single-column.
+- **Merge stream-prep leak** (F1): `prepareAuthorStreams`/`prepareTagStreams`
+  now build cursors through `buildStreams`, which closes any already-prepared
+  statements if a later prepare throws — otherwise a mid-loop failure stranded
+  checked-out, un-reset handles (read locks) in the pooled connection.
+- **Stream-count overflow** (F3): `authors×kinds` / `values×kinds` computed as
+  `Long` so a pathological product can't wrap `Int` back into the eligible
+  band and route a huge fan-out into the merge.
+- **Finalizer footgun** (F2): the pooled statement's `finalize()` was renamed
+  `finalizeStatement()` — a no-arg `finalize()` is the JVM's `Object.finalize`,
+  so the GC could double-close the native handle after explicit close.
+
+Migration cost noted (low/operational, not correctness): the synchronous
+v4→v5 rebuild runs in the upgrade transaction; atomic and kill-safe (rolls
+back to v4), but a very large client store pays a one-time first-open stall.

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/EoseMessage.kt

@@ -25,6 +25,22 @@ class EoseMessage(
 ) : Message {
     override fun label() = LABEL
 
+    /**
+     * Wire form is `["EOSE","<subId>"]` — sent once per REQ, so it is on
+     * the per-subscription floor. Splice it directly when [subId] needs no
+     * escaping (the common case: client-chosen sub ids are short ASCII),
+     * skipping the generic serializer's node tree. Byte-identical output;
+     * any exotic subId falls back.
+     */
+    override fun toJson(): String {
+        if (!isEscapeFreeAscii(subId)) return super.toJson()
+        return buildString(subId.length + 12) {
+            append("[\"EOSE\",\"")
+            append(subId)
+            append("\"]")
+        }
+    }
+
     companion object {
         const val LABEL = "EOSE"
     }

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/OkMessage.kt

@@ -29,6 +29,25 @@ class OkMessage(
 ) : Message {
     override fun label() = LABEL
 
+    /**
+     * Wire form is `["OK","<eventId>",<true|false>,"<message>"]` — sent
+     * once per published EVENT. [eventId] is validated hex (always
+     * escape-free); splice directly when [message] also needs no escaping,
+     * which covers the empty-string success ack and the plain-ASCII
+     * rejection reasons. Byte-identical output; a reason with quotes or
+     * non-ASCII falls back to the generic serializer.
+     */
+    override fun toJson(): String {
+        if (!isEscapeFreeAscii(message)) return super.toJson()
+        return buildString(eventId.length + message.length + 20) {
+            append("[\"OK\",\"")
+            append(eventId)
+            append(if (success) "\",true,\"" else "\",false,\"")
+            append(message)
+            append("\"]")
+        }
+    }
+
     companion object {
         const val LABEL = "OK"
 

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/WireJson.kt

@@ -0,0 +1,36 @@
+/*
+ * 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.commands.toClient
+
+/**
+ * True when every char of [s] is printable ASCII (0x20–0x7e) and not a JSON
+ * metacharacter (`"` / `\`) — i.e. exactly the bytes a JSON string encoder
+ * would emit verbatim between the quotes. Frame builders use this to gate a
+ * direct-`buildString` fast path against the generic serializer: when it holds
+ * the spliced output is byte-identical, and any exotic value (control chars,
+ * quotes, non-ASCII) falls back to the escaping serializer.
+ */
+internal fun isEscapeFreeAscii(s: String): Boolean {
+    for (c in s) {
+        if (c < ' ' || c > '~' || c == '"' || c == '\\') return false
+    }
+    return true
+}

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

@@ -22,6 +22,11 @@ package com.vitorpamplona.quartz.nip01Core.relay.filters
 
 import com.vitorpamplona.quartz.nip01Core.core.Event
 import com.vitorpamplona.quartz.nip01Core.core.HexKey
+import kotlinx.collections.immutable.PersistentMap
+import kotlinx.collections.immutable.PersistentSet
+import kotlinx.collections.immutable.persistentHashMapOf
+import kotlinx.collections.immutable.persistentHashSetOf
+import kotlinx.collections.immutable.toPersistentHashSet
 import kotlin.concurrent.atomics.AtomicReference
 import kotlin.concurrent.atomics.ExperimentalAtomicApi
 
@@ -62,9 +67,10 @@ import kotlin.concurrent.atomics.ExperimentalAtomicApi
  * 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).
+ * Writes (subscription register / unregister) build the next snapshot
+ * from persistent (HAMT) maps — O(keys × log S) with structural
+ * sharing rather than a full O(S) copy of both maps, since on a relay
+ * a write happens on every REQ open and close.
  *
  * ## What the index does NOT cover
  *
@@ -109,14 +115,26 @@ class FilterIndex<S : Any> {
     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.
+     * Single immutable snapshot. Subscribers are held in one map per
+     * indexable dimension so [candidatesFor] — called once per accepted
+     * ingest event, the hot read — can probe each dimension with the
+     * event's own field (`event.id`, `event.pubKey`, `tag[0]`/`tag[1]`,
+     * `event.kind`) and allocate no key-wrapper objects. [assignments] is
+     * the reverse map ([S] → the [BucketKey]s it occupies) used by
+     * [unregister]; the wrappers live only here, built on the rare
+     * register path.
+     *
+     * Persistent (HAMT) maps/sets: a register/unregister produces the
+     * next snapshot in O(keys × log S) with structural sharing, instead
+     * of copying full maps — registration happens on every REQ open/close.
      */
     private data class State<S>(
-        val buckets: Map<BucketKey, Set<S>> = emptyMap(),
-        val assignments: Map<S, Set<BucketKey>> = emptyMap(),
+        val ids: PersistentMap<HexKey, PersistentSet<S>> = persistentHashMapOf(),
+        val authors: PersistentMap<HexKey, PersistentSet<S>> = persistentHashMapOf(),
+        val tags: PersistentMap<String, PersistentMap<String, PersistentSet<S>>> = persistentHashMapOf(),
+        val kinds: PersistentMap<Int, PersistentSet<S>> = persistentHashMapOf(),
+        val unindexed: PersistentSet<S> = persistentHashSetOf(),
+        val assignments: PersistentMap<S, PersistentSet<BucketKey>> = persistentHashMapOf(),
     )
 
     private val state: AtomicReference<State<S>> = AtomicReference(State())
@@ -181,18 +199,23 @@ class FilterIndex<S : Any> {
         while (true) {
             val current = state.load()
             val keys = current.assignments[subscriber] ?: return
-            val newBuckets = current.buckets.toMutableMap()
+            var ids = current.ids
+            var authors = current.authors
+            var tags = current.tags
+            var kinds = current.kinds
+            var unindexed = current.unindexed
             for (key in keys) {
-                val cur = newBuckets[key] ?: continue
-                val next = cur - subscriber
-                if (next.isEmpty()) {
-                    newBuckets.remove(key)
-                } else {
-                    newBuckets[key] = next
+                when (key) {
+                    is IdKey -> ids = ids.removeSub(key.id, subscriber)
+                    is AuthorKey -> authors = authors.removeSub(key.author, subscriber)
+                    is KindKey -> kinds = kinds.removeSub(key.kind, subscriber)
+                    is TagKey -> tags = tags.removeTagSub(key.letter, key.value, subscriber)
+                    Unindexed -> unindexed = unindexed.remove(subscriber)
                 }
             }
-            val newAssignments = current.assignments - subscriber
-            if (state.compareAndSet(current, State(newBuckets, newAssignments))) return
+            val next =
+                State(ids, authors, tags, kinds, unindexed, current.assignments.remove(subscriber))
+            if (state.compareAndSet(current, next)) return
         }
     }
 
@@ -202,19 +225,22 @@ class FilterIndex<S : Any> {
      * candidate to handle negative constraints.
      *
      * Iteration order is insertion-stable per call but otherwise
-     * unspecified.
+     * unspecified. Allocates only the result set — dimensions are
+     * probed with the event's own fields, no key wrappers.
      */
     fun candidatesFor(event: Event): Set<S> {
         val s = state.load()
-        if (s.buckets.isEmpty()) return emptySet()
+        if (s.assignments.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) }
+        if (s.unindexed.isNotEmpty()) result.addAll(s.unindexed)
+        s.ids[event.id]?.let { result.addAll(it) }
+        s.authors[event.pubKey]?.let { result.addAll(it) }
+        s.kinds[event.kind]?.let { result.addAll(it) }
+        if (s.tags.isNotEmpty()) {
+            for (tag in event.tags) {
+                if (tag.size >= 2 && tag[0].length == 1) {
+                    s.tags[tag[0]]?.get(tag[1])?.let { result.addAll(it) }
+                }
             }
         }
         return result
@@ -235,19 +261,75 @@ class FilterIndex<S : Any> {
         keys: List<BucketKey>,
     ) {
         if (keys.isEmpty()) return
-        val keySet = keys.toSet()
+        val keySet = keys.toPersistentHashSet()
         while (true) {
             val current = state.load()
-            val newBuckets = current.buckets.toMutableMap()
+            var ids = current.ids
+            var authors = current.authors
+            var tags = current.tags
+            var kinds = current.kinds
+            var unindexed = current.unindexed
             for (key in keySet) {
-                val cur = newBuckets[key] ?: emptySet()
-                if (subscriber in cur) continue
-                newBuckets[key] = cur + subscriber
+                when (key) {
+                    is IdKey -> ids = ids.addSub(key.id, subscriber)
+                    is AuthorKey -> authors = authors.addSub(key.author, subscriber)
+                    is KindKey -> kinds = kinds.addSub(key.kind, subscriber)
+                    is TagKey -> tags = tags.addTagSub(key.letter, key.value, subscriber)
+                    Unindexed -> unindexed = unindexed.add(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
+            val merged = existing?.addAll(keySet) ?: keySet
+            val next = State(ids, authors, tags, kinds, unindexed, current.assignments.put(subscriber, merged))
+            if (state.compareAndSet(current, next)) return
+        }
+    }
+
+    // Per-dimension add/remove of one subscriber, returning the same map
+    // instance when nothing changed so the CAS builds minimal new nodes.
+    private fun <K> PersistentMap<K, PersistentSet<S>>.addSub(
+        key: K,
+        sub: S,
+    ): PersistentMap<K, PersistentSet<S>> {
+        val cur = this[key] ?: persistentHashSetOf()
+        val next = cur.add(sub)
+        return if (next === cur) this else put(key, next)
+    }
+
+    private fun <K> PersistentMap<K, PersistentSet<S>>.removeSub(
+        key: K,
+        sub: S,
+    ): PersistentMap<K, PersistentSet<S>> {
+        val cur = this[key] ?: return this
+        val next = cur.remove(sub)
+        return when {
+            next === cur -> this
+            next.isEmpty() -> remove(key)
+            else -> put(key, next)
+        }
+    }
+
+    private fun PersistentMap<String, PersistentMap<String, PersistentSet<S>>>.addTagSub(
+        letter: String,
+        value: String,
+        sub: S,
+    ): PersistentMap<String, PersistentMap<String, PersistentSet<S>>> {
+        val inner = this[letter] ?: persistentHashMapOf()
+        val newInner = inner.addSub(value, sub)
+        return if (newInner === inner) this else put(letter, newInner)
+    }
+
+    private fun PersistentMap<String, PersistentMap<String, PersistentSet<S>>>.removeTagSub(
+        letter: String,
+        value: String,
+        sub: S,
+    ): PersistentMap<String, PersistentMap<String, PersistentSet<S>>> {
+        val inner = this[letter] ?: return this
+        val newInner = inner.removeSub(value, sub)
+        return when {
+            newInner === inner -> this
+            newInner.isEmpty() -> remove(letter)
+            else -> put(letter, newInner)
         }
     }
 

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

@@ -49,6 +49,7 @@ import com.vitorpamplona.quartz.utils.Log
 import com.vitorpamplona.quartz.utils.cache.LargeCache
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.CoroutineStart
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.channels.ClosedSendChannelException
 import kotlinx.coroutines.launch
@@ -291,8 +292,16 @@ class RelaySession(
         // Policy may rewrite filters to match the user's access level.
         val filters = (result as PolicyResult.Accepted).cmd.filters
 
+        // UNDISPATCHED: the stored replay runs inline on this coroutine —
+        // the reader-pool acquire doesn't suspend when a connection is
+        // free, so EVENT frames and EOSE go out without a scheduler hop
+        // (SmallReqFloorBenchmark: the hop was most of the dispatch
+        // slice on small REQs). The coroutine first parks at the live
+        // tail (awaitCancellation), which is when launch returns and the
+        // job lands in [subscriptions]; commands on this connection are
+        // processed sequentially, so nothing can target the sub earlier.
         val job =
-            scope.launch {
+            scope.launch(start = CoroutineStart.UNDISPATCHED) {
                 try {
                     if (policy.filtersOutgoingEvents) {
                         // Screened path: every event is materialized so the
@@ -331,7 +340,20 @@ class RelaySession(
                                     },
                                 )
                             },
-                            onEachLive = { event -> send(EventMessage(cmd.subId, event)) },
+                            // Live events arrive with their wire body already
+                            // serialized (once per event, shared across every
+                            // matching subscription): splice it into the same
+                            // per-sub frame prefix as the stored replay, no
+                            // per-event EventMessage or re-serialize.
+                            onEachLive = { _, body ->
+                                sendRaw(
+                                    buildString(framePrefix.length + body.length + 1) {
+                                        append(framePrefix)
+                                        append(body)
+                                        append(']')
+                                    },
+                                )
+                            },
                             onEose = { send(EoseMessage(cmd.subId)) },
                         )
                     }

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

@@ -74,13 +74,57 @@ class LiveEventStore(
      * 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].
+     * the index dispatches into. [deliver] receives the event and its
+     * pre-serialized wire body (memoized once per fanout across all
+     * matching subscribers). Identity-keyed inside [FilterIndex].
      */
     private class LiveSubscription(
         val filters: List<Filter>,
-        val deliver: (Event) -> Unit,
+        val deliver: (Event, String) -> Unit,
     )
 
+    /**
+     * Replay-dedupe set for one REQ. During the historical replay the
+     * store's ids are [record]ed here so the concurrent live path can
+     * drop an event the replay also emitted; after EOSE the set is
+     * [release]d and the live path forwards everything.
+     *
+     * Written from the replay coroutine and read from the [IngestQueue]
+     * drain coroutine (via `fanout`), so every access takes a tiny spin
+     * lock — [locked] is `inline`, so the per-row `record` / `isDuplicate`
+     * calls allocate no closure. The backing `HashSet` is created empty
+     * up front (so the register-before-replay race guarantee holds) but
+     * the JVM defers its table allocation to the first `add`, so a
+     * zero-row replay costs only the empty set object, not a sized table.
+     * It MUST stay a mutable set under a lock, never a copy-on-add
+     * immutable set — `set + id` per row made large replays O(n²).
+     */
+    private class SeenIds {
+        private val lock = AtomicBoolean(false)
+        private var ids: HashSet<String>? = HashSet()
+
+        private inline fun <R> locked(block: () -> R): R {
+            while (lock.exchange(true)) {
+                while (lock.load()) { }
+            }
+            try {
+                return block()
+            } finally {
+                lock.store(false)
+            }
+        }
+
+        fun record(id: String) {
+            locked { ids?.add(id) }
+        }
+
+        fun isDuplicate(id: String): Boolean = locked { ids?.contains(id) ?: false }
+
+        fun release() {
+            locked { ids = null }
+        }
+    }
+
     /**
      * Fire-and-forget enqueue: hand [event] to the [IngestQueue] and
      * fire [onComplete] once the writer's batch has a per-row
@@ -149,9 +193,17 @@ class LiveEventStore(
      * batch writer.
      */
     private fun fanout(event: Event) {
-        for (sub in index.candidatesFor(event)) {
+        val candidates = index.candidatesFor(event)
+        if (candidates.isEmpty()) return
+        // Serialize the wire body at most once for this event, no matter
+        // how many subscriptions match it — the old path re-serialized the
+        // whole event per matching subscriber, so a note landing in N live
+        // feeds paid N identical Jackson passes. Lazy so a fanout that
+        // matches nothing (index over-approximates) serializes nothing.
+        var body: String? = null
+        for (sub in candidates) {
             if (sub.filters.any { it.match(event) }) {
-                sub.deliver(event)
+                sub.deliver(event, body ?: event.toJson().also { body = it })
             }
         }
     }
@@ -178,61 +230,33 @@ class LiveEventStore(
         onEose: () -> Unit,
     ) {
         drainFtsIfSearching(filters)
-        // 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`).
-        //
-        // 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),
-        // so access is guarded by a tiny spin lock (contains/add,
-        // never I/O). It MUST be a mutable set under a lock, not an
-        // immutable Set under an AtomicReference with copy-on-add:
-        // `set + id` copies the whole set per streamed event, which
-        // made large replays accidentally O(n²) — a 100k-event REQ
-        // crawled at ~700 events/s and the rate degraded as the
-        // response grew (see the plan doc's giant-REQ finding).
-        //
-        // Once cleared to null after EOSE, `deliver` short-circuits
-        // and every live event is forwarded.
-        val seenLock = AtomicBoolean(false)
-        var seenIds: HashSet<String>? = HashSet(1024)
-
-        fun <R> seenLocked(block: () -> R): R {
-            while (seenLock.exchange(true)) {
-                while (seenLock.load()) { }
-            }
-            try {
-                return block()
-            } finally {
-                seenLock.store(false)
-            }
-        }
+        // 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`), and [SeenIds] bridges the two coroutines:
+        // the replay records ids here, the live `deliver` drops duplicates,
+        // and after EOSE the set is released so every live event forwards.
+        val seen = SeenIds()
 
         val sub =
             LiveSubscription(
                 filters = filters,
-                deliver = { event ->
-                    val duplicate = seenLocked { seenIds?.contains(event.id) ?: false }
-                    if (duplicate) return@LiveSubscription
-                    onEach(event)
+                deliver = { event, _ ->
+                    if (!seen.isDuplicate(event.id)) onEach(event)
                 },
             )
 
         index.register(filters, sub)
         try {
             store.query<Event>(filters.strippingSearchExtensions()) { event ->
-                seenLocked { seenIds?.add(event.id) }
+                seen.record(event.id)
                 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.
-            seenLocked { seenIds = null }
+            seen.release()
             // Suspend until the caller's coroutine is cancelled
             // (e.g. NIP-01 CLOSE or connection drop). The `finally`
             // unregisters from the index.
@@ -255,42 +279,28 @@ class LiveEventStore(
         ctx: RequestContext,
         filters: List<Filter>,
         onEachStored: (RawEvent) -> Unit,
-        onEachLive: (Event) -> Unit,
+        onEachLive: (Event, String) -> Unit,
         onEose: () -> Unit,
     ) {
         drainFtsIfSearching(filters)
-        val seenLock = AtomicBoolean(false)
-        var seenIds: HashSet<String>? = HashSet(1024)
-
-        fun <R> seenLocked(block: () -> R): R {
-            while (seenLock.exchange(true)) {
-                while (seenLock.load()) { }
-            }
-            try {
-                return block()
-            } finally {
-                seenLock.store(false)
-            }
-        }
+        val seen = SeenIds()
 
         val sub =
             LiveSubscription(
                 filters = filters,
-                deliver = { event ->
-                    val duplicate = seenLocked { seenIds?.contains(event.id) ?: false }
-                    if (duplicate) return@LiveSubscription
-                    onEachLive(event)
+                deliver = { event, body ->
+                    if (!seen.isDuplicate(event.id)) onEachLive(event, body)
                 },
             )
 
         index.register(filters, sub)
         try {
             store.rawQuery(filters.strippingSearchExtensions()) { raw ->
-                seenLocked { seenIds?.add(raw.id) }
+                seen.record(raw.id)
                 onEachStored(raw)
             }
             onEose()
-            seenLocked { seenIds = null }
+            seen.release()
             awaitCancellation()
         } finally {
             index.unregister(sub)

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

@@ -78,9 +78,9 @@ interface SessionBackend {
         ctx: RequestContext,
         filters: List<Filter>,
         onEachStored: (RawEvent) -> Unit,
-        onEachLive: (Event) -> Unit,
+        onEachLive: (Event, String) -> Unit,
         onEose: () -> Unit,
-    ): Unit = query(ctx, filters, onEachLive, onEose)
+    ): Unit = query(ctx, filters, { onEachLive(it, it.toJson()) }, onEose)
 
     /** Answers a NIP-45 COUNT with an exact cardinality for the caller in [ctx]. */
     suspend fun count(

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/EventIndexesModule.kt

@@ -147,13 +147,38 @@ class EventIndexesModule(
      */
     fun migrateV2AddPubkeyIndex(db: SQLiteConnection) {
         if (!indexStrategy.indexEventsByPubkeyAlone) return
-        val orderBy =
-            if (indexStrategy.useAndIndexIdOnOrderBy) {
-                "created_at DESC, id ASC"
-            } else {
-                "created_at DESC"
-            }
-        db.execSQL("CREATE INDEX IF NOT EXISTS query_by_pubkey_created ON event_headers (pubkey, $orderBy)")
+        db.execSQL("CREATE INDEX IF NOT EXISTS query_by_pubkey_created ON event_headers (pubkey, ${orderByColumns()})")
+    }
+
+    private fun orderByColumns() =
+        if (indexStrategy.useAndIndexIdOnOrderBy) {
+            "created_at DESC, id ASC"
+        } else {
+            "created_at DESC"
+        }
+
+    /**
+     * Materializes any flag-gated index the current [indexStrategy] wants
+     * but the on-disk schema predates. Flags are runtime configuration, not
+     * schema — a deployment can flip one without a `user_version` bump — so
+     * this runs idempotently on every open. The first open after enabling a
+     * flag pays a one-time index build over the existing rows; subsequent
+     * opens are no-ops. A disabled flag never drops an existing index (that
+     * stays an operator decision).
+     */
+    fun ensureOptionalIndexes(db: SQLiteConnection) {
+        if (indexStrategy.indexEventsByCreatedAtAlone) {
+            db.execSQL("CREATE INDEX IF NOT EXISTS query_by_created_at_id ON event_headers (${orderByColumns()})")
+        }
+        if (indexStrategy.indexEventsByPubkeyAlone) {
+            db.execSQL("CREATE INDEX IF NOT EXISTS query_by_pubkey_created ON event_headers (pubkey, ${orderByColumns()})")
+        }
+        if (indexStrategy.indexTagsByCreatedAtAlone) {
+            db.execSQL("CREATE INDEX IF NOT EXISTS query_by_tags_hash ON event_tags (tag_hash, created_at DESC)")
+        }
+        if (indexStrategy.indexTagsWithKindAndPubkey) {
+            db.execSQL("CREATE INDEX IF NOT EXISTS query_by_tags_hash_kind_pubkey ON event_tags (tag_hash, kind, pubkey_hash, created_at DESC)")
+        }
     }
 
     val sqlInsertHeader =

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/FullTextSearchModule.kt

@@ -31,6 +31,37 @@ import com.vitorpamplona.quartz.utils.EventFactory
 /**
  * NIP-50 full-text search index over event content.
  *
+ * The index is a **contentless** FTS5 table (`content=''`,
+ * `contentless_delete=1`) whose `rowid` is the event's
+ * `event_headers.row_id`. Two consequences:
+ *
+ *  - **Deletes.** The `fts_foreign_key` trigger deletes by `rowid` (an FTS5
+ *    primary-key seek, O(log n)). The old `fts5(event_header_row_id,
+ *    content)` schema deleted by a *regular* column, which FTS5 cannot seek —
+ *    it scans the whole index per delete (O(n)), so deletion throughput
+ *    degraded with corpus size. Every event removal fires this trigger
+ *    (replaceable rotation, kind-5, expiration, right-to-vanish), so the
+ *    seek matters. Measured `Fts5CapabilityProbe`/`FtsSearchScalingBenchmark`:
+ *    ~78× at 8k rows and widening with the table.
+ *  - **Size.** A contentless table stores only the inverted index, not a
+ *    second copy of the tokenized text. The indexed text is *derived*
+ *    ([SearchableEvent.indexableContent], not any raw column), so FTS5
+ *    external-content — which reads the source column from the base table —
+ *    cannot express it; contentless is the correct primitive.
+ *
+ * Search results are ordered by **relevance**, per NIP-50 ("descending order
+ * by quality of search result ... not by the usual `.created_at`", limit
+ * applied after the score) — via FTS5 bm25 (`ORDER BY event_fts.rank`), with
+ * `created_at DESC` only as a tie-break. This holds for *every* search filter:
+ * the tag-free shape ([QueryBuilder.makeSimpleSearch]), `search + tag` (whose
+ * row-id subquery carries the rank through via `projectRank`), and a
+ * multi-filter all-search REQ (unioned, deduped by event keeping the best
+ * score). Only the negentropy snapshot — and a multi-filter REQ mixing search
+ * and non-search branches — keep `created_at` (a sync set / a branch with no
+ * defined relevance). bm25 must score every match, so search latency still grows with the
+ * match set regardless of ordering; corpus-independent search needs an
+ * external engine, not this index.
+ *
  * When [enabled] is `false` the module becomes an inert no-op: no
  * `event_fts` virtual table and no `fts_foreign_key` delete trigger are
  * created, inserts skip the per-event tokenization cost, and the reindex
@@ -52,29 +83,53 @@ class FullTextSearchModule(
 ) : IModule {
     val tableName = "event_fts"
     val triggerName = "fts_foreign_key"
-    val eventHeaderRowIdName = "event_header_row_id"
+
+    /**
+     * The FTS column that links back to `event_headers`. It is the implicit
+     * `rowid` of the (contentless) FTS table, which we set equal to
+     * `event_headers.row_id` on every insert — so the join is
+     * `event_headers.row_id = event_fts.rowid`, with no stored column.
+     */
+    val rowIdColumn = "rowid"
     val contentName = "content"
     val stateTableName = "fts_catchup_state"
 
+    /**
+     * Whether the on-disk `event_fts` is an FTS5 table (vs the fts4/fts3
+     * fallback). Only FTS5 supports the contentless schema and the
+     * `merge`/`optimize` maintenance commands; the fallback stores content
+     * and skips maintenance. Read only on the (single-threaded) writer.
+     * Cached lazily from `sqlite_master` so a reopened DB — where [create]
+     * never runs — still resolves it.
+     */
+    private var isFts5: Boolean? = null
+
     override fun create(db: SQLiteConnection) {
         if (!enabled) return
         val ftsVersion = versionFinder(db)
-        db.execSQL(
-            """
-                    CREATE VIRTUAL TABLE $tableName
-                    USING fts$ftsVersion($eventHeaderRowIdName, $contentName)
-                """,
-        )
+        isFts5 = ftsVersion >= 5
+        // FTS5: contentless index (no stored content copy) with delete
+        // support. fts4/fts3 (bundled driver never selects them) fall back to
+        // a plain content-storing table — rowid-explicit insert and
+        // delete-by-rowid work there too, only the size win is FTS5-only.
+        val columns =
+            if (ftsVersion >= 5) {
+                "$contentName, content='', contentless_delete=1"
+            } else {
+                contentName
+            }
+        db.execSQL("CREATE VIRTUAL TABLE $tableName USING fts$ftsVersion($columns)")
 
-        // Foreign key cleanup for full text search
+        // Foreign key cleanup for full text search. Deletes by the FTS rowid
+        // (= event_headers.row_id); a header with no FTS row (non-searchable
+        // kind) deletes nothing, which is a harmless no-op.
         db.execSQL(
             """
                 CREATE TRIGGER $triggerName
                 AFTER DELETE ON event_headers
                 FOR EACH ROW
                 BEGIN
-                    DELETE FROM $tableName
-                    WHERE old.row_id = $tableName.$eventHeaderRowIdName;
+                    DELETE FROM $tableName WHERE $tableName.rowid = old.row_id;
                 END;
             """,
         )
@@ -82,6 +137,16 @@ class FullTextSearchModule(
         createStateTable(db)
     }
 
+    private fun resolveIsFts5(db: SQLiteConnection): Boolean {
+        isFts5?.let { return it }
+        val sql =
+            db.prepare("SELECT sql FROM sqlite_master WHERE type = 'table' AND name = ?").use { stmt ->
+                stmt.bindText(1, tableName)
+                if (stmt.step()) stmt.getText(0) else ""
+            }
+        return sql.contains("fts5", ignoreCase = true).also { isFts5 = it }
+    }
+
     /**
      * Watermark for the deferred path: everything with
      * `row_id <= last_row_id` is guaranteed indexed. Idempotent — also
@@ -124,13 +189,13 @@ class FullTextSearchModule(
 
     val insertFTS =
         """
-        INSERT OR ROLLBACK INTO $tableName ($eventHeaderRowIdName, $contentName)
+        INSERT OR ROLLBACK INTO $tableName (rowid, $contentName)
         VALUES (?, ?)
         """.trimIndent()
 
     val deleteFTSByRowId =
         """
-        DELETE FROM $tableName WHERE $eventHeaderRowIdName = ?
+        DELETE FROM $tableName WHERE rowid = ?
         """.trimIndent()
 
     fun insert(
@@ -200,7 +265,19 @@ class FullTextSearchModule(
         dropTrigger(db)
         drop(db)
         create(db)
+        populateAll(db)
+        // The rebuild just wrote every row as its own tiny segment; compact
+        // them into one so the first search after a reindex isn't a scan over
+        // hundreds of segments.
+        optimize(db)
+    }
 
+    /**
+     * Scan every stored searchable event and insert its derived content into
+     * the (already created, empty) FTS index, keyed by `event_headers.row_id`.
+     * The caller owns the transaction and the create/drop lifecycle.
+     */
+    private fun populateAll(db: SQLiteConnection) {
         val kinds = searchableKindsPresent(db)
         if (kinds.isEmpty()) return
 
@@ -230,6 +307,58 @@ class FullTextSearchModule(
         }
     }
 
+    /**
+     * v4 → v5 migration: the pre-v5 index was `fts5(event_header_row_id,
+     * content)` with an auto-assigned rowid unrelated to `event_headers`, and
+     * it stored a second copy of the content. v5 is the contentless,
+     * rowid = row_id schema. The old rowids can't be remapped in place, so the
+     * table is dropped and repopulated. Runs inside the migration transaction.
+     *
+     *  - **synchronous** stores rebuild now (client corpora are small).
+     *  - **deferred** stores reset the catch-up watermark to 0 so the relay's
+     *    background worker repopulates without a long migration transaction.
+     */
+    fun migrateV4ToContentless(db: SQLiteConnection) {
+        if (!enabled) return
+        dropTrigger(db)
+        drop(db)
+        create(db)
+        if (deferIndexing) {
+            db.prepare("UPDATE $stateTableName SET last_row_id = 0 WHERE id = 1").use { it.step() }
+        } else {
+            populateAll(db)
+            optimize(db)
+        }
+    }
+
+    /**
+     * Full FTS5 segment compaction — merges the b-tree segments left by
+     * incremental inserts into one, so a `MATCH` touches a single segment
+     * instead of dozens. Expensive (rewrites the whole index); call it once
+     * after a rebuild, not per batch. No-op on the fts4/fts3 fallback.
+     */
+    fun optimize(db: SQLiteConnection) {
+        if (!enabled || !resolveIsFts5(db)) return
+        db.prepare("INSERT INTO $tableName($tableName) VALUES ('optimize')").use { it.step() }
+    }
+
+    /**
+     * Bounded incremental segment merge — does at most [pages] pages of merge
+     * work, so it stays cheap enough to run on a periodic maintenance tick
+     * while the index keeps growing from deferred catch-up. No-op on the
+     * fts4/fts3 fallback.
+     */
+    fun mergeSegments(
+        db: SQLiteConnection,
+        pages: Int = 16,
+    ) {
+        if (!enabled || !resolveIsFts5(db)) return
+        db.prepare("INSERT INTO $tableName($tableName, rank) VALUES ('merge', ?)").use { stmt ->
+            stmt.bindLong(1, pages.toLong())
+            stmt.step()
+        }
+    }
+
     /**
      * Process one batch of a resumable rebuild: re-derive the FTS rows
      * for up to [batchSize] events whose `row_id > ` [afterRowId] and
@@ -329,13 +458,11 @@ class FullTextSearchModule(
         // Unlike [reindexBatch] there is NO per-row delete here: rows past
         // the watermark were never indexed (deferred mode skips insert()),
         // and the watermark advances atomically with the FTS rows it
-        // covers, so a crash replay is impossible. The delete would also
-        // be ruinous — `event_header_row_id` is a plain FTS5 column, so
-        // deleting by it scans the whole FTS table per row, which turned
-        // the first catch-up implementation O(n²). Consequence: switching
-        // a database back and forth between deferred and synchronous
-        // strategies requires a [reindexAll] in between (same rule as a
-        // searchable-kinds change).
+        // covers, so a crash replay is impossible. (Delete-by-rowid is now
+        // O(log n) on the contentless index, so this is purely about not
+        // doing redundant work.) Consequence: switching a database back and
+        // forth between deferred and synchronous strategies requires a
+        // [reindexAll] in between (same rule as a searchable-kinds change).
         val limit = batchSize.coerceAtLeast(1)
         val kinds = searchableKindsPresent(db)
         var last = watermark

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/IndexingStrategy.kt

@@ -74,9 +74,21 @@ interface IndexingStrategy {
      * Activate this if you see too many Tag-centric Filters without
      * kind AND pubkey at the same time.
      *
-     * This is a rarely used index (reports by your follows or
-     * NIP-04 DMs for instance) that becomes quite large without
-     * major gains.
+     * This shape (reports by your follows, NIP-04 DM rooms, follows-scoped
+     * community feeds) is not rare on the client side: the 2026-07 filter
+     * assembler survey counted 65 call sites building
+     * `kinds + authors + tags`. Without this index the plan seeks
+     * `(tag_hash, kind)` and reads every row for that tag/kind before
+     * filtering the author.
+     *
+     * Measured by `TagAuthorIndexBenchmark` (jvmTest prodbench): the
+     * DM-room query drops 9.4 ms → 0.6 ms (~15×) at 200k events and
+     * 14.2 ms → 0.66 ms (~21×) at 1M — the gap grows with corpus size —
+     * while batch-insert cost stays inside run noise (49.0 vs 47.4
+     * µs/event at 1M). geode enables it; the client default stays off
+     * because a client store's per-tag row counts are bounded by one
+     * user's data. Flipping it on an existing DB is safe: the index is
+     * built on next open by `EventIndexesModule.ensureOptionalIndexes`.
      *
      * Keep in mind that activating too many indexes increases the size of the
      * DB so much that the indexes themselves won't fit in memory, requiring

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/MergeQueryExecutor.kt

@@ -24,52 +24,94 @@ import androidx.sqlite.SQLiteConnection
 import androidx.sqlite.SQLiteStatement
 
 /**
- * k-way merge executor for the **home-feed** query shape:
- * `authors=[…] (+ kinds=[…]) [+ since/until] limit=N` ordered newest-first.
+ * k-way merge executor for the two "wide fan-out, newest-N" query shapes
+ * whose single-SQL plan reads O(matching history) rather than O(limit):
  *
- * SQLite serves this by seeking every `(kind, pubkey)` combo and feeding
- * *all* matching rows through a LIMIT-bounded sorter — so it reads O(the
- * followed set's whole matching history). For prolific follows on a cold
- * on-disk DB that's the `follow-feed` regression (relayBench: 97 ms vs
- * strfry 17 ms). See `quartz/plans/2026-07-04-follow-feed-read-tradeoff.md`.
+ *  1. **home-feed** — `authors=[…] (+ kinds=[…]) [+ since/until] limit=N`.
+ *     SQLite seeks every `(kind, pubkey)` combo and feeds *all* matching rows
+ *     through a LIMIT-bounded sorter — O(the followed set's whole matching
+ *     history). For prolific follows on a cold on-disk DB that was the
+ *     `follow-feed` regression (relayBench: 97 ms vs strfry 17 ms). See
+ *     `quartz/plans/2026-07-04-follow-feed-read-tradeoff.md`.
+ *  2. **tag watcher** — `#<x>=[hundreds of values] (+ kinds=[…])
+ *     [+ since/until] limit=N`, the reactions/replies archetype
+ *     (`kinds=[7] AND #e=[note ids]`). The per-value streams come sorted off
+ *     `(tag_hash[, kind], created_at)`, but their union does not, so SQLite
+ *     collects every matching row and TEMP-B-TREE sorts to the limit — the
+ *     tag-index analogue of the follow-feed shape. Measured by
+ *     `TagAuthorIndexBenchmark` (jvmTest prodbench): `#e IN 300, limit 500`
+ *     cost 12.8 ms cold at 200k events and 14.2 ms at 1M, growing with
+ *     matching history.
  *
- * Each `(kind, pubkey)` is already a newest-first stream off the
- * `query_by_kind_pubkey_created (kind, pubkey, created_at DESC)` index
- * (or `query_by_pubkey_created` for authors-only). This opens one lazy
- * cursor per stream and merges their heads, stopping at the limit — so it
- * reads only **O(limit + streams)** rows regardless of how much history the
- * authors have, and it reuses the existing indexes (no write/size cost).
+ * Each stream is already a newest-first cursor off an existing index:
+ *  - authors: `query_by_kind_pubkey_created (kind, pubkey, created_at DESC)`
+ *    (or `query_by_pubkey_created` for authors-only);
+ *  - tags: `query_by_tags_hash_kind (tag_hash, kind, created_at DESC)`
+ *    (or `query_by_tags_hash` for the no-kind case, gated by
+ *    [IndexingStrategy.indexTagsByCreatedAtAlone]).
+ *
+ * The merge opens one lazy cursor per stream, merges their heads newest-first,
+ * and stops at the limit — reading **O(limit + streams)** rows regardless of
+ * how much history the authors/tags have, and reusing the existing indexes
+ * (no write/size cost). With the pooled statement cache
+ * ([StatementCachingConnection]) the per-stream cursors are prepared once and
+ * reused across repeated polls of the same REQ.
  *
  * Merge order is `created_at DESC`, tie-broken by `id ASC`. NIP-01 leaves
- * same-`created_at` ties unspecified, so the returned set is a valid
- * newest-N either way. The `id ASC` tie-break is exact — byte-for-byte the
- * same events the single-SQL path returns — only when the store indexes id
- * ([IndexingStrategy.useAndIndexIdOnOrderBy]): then each per-stream cursor
- * streams in `(created_at DESC, id ASC)` straight off the index, so a
- * stream's same-second head really is its id-minimum. Without that index
- * the per-stream cursor yields same-second rows in rowid order, so the
- * result is still a valid newest-N but may differ from the single-SQL path
- * exactly at a same-second boundary.
+ * same-`created_at` ties unspecified, so the returned set is a valid newest-N
+ * either way. The `id ASC` tie-break is exact — byte-for-byte the same events
+ * the single-SQL path returns — only when the store indexes id
+ * ([IndexingStrategy.useAndIndexIdOnOrderBy]) **and** the stream cursor can
+ * order by id off the index. The author streams can (id is on
+ * `event_headers`); the tag streams cannot (the cursor orders off
+ * `event_tags`, which has no id column), so a tag stream yields same-second
+ * rows in rowid order — still a valid newest-N, but same-second ties may
+ * differ from an id-ordered reference.
+ *
+ * **Cross-stream duplicates.** An author appears in exactly one author stream
+ * (one pubkey per event), so the home-feed merge never double-counts. A single
+ * event can carry several of the queried tag values (or a repeated tag), so it
+ * can surface in several tag streams — the single-SQL path dedups with
+ * `SELECT DISTINCT`. The tag merge therefore dedups by event id through a
+ * `seen` set; the author merge skips that set entirely.
  */
 internal object MergeQueryExecutor {
+    /** Author-stream projection: a single `event_headers` scan, unqualified. */
     const val COLS = "id, pubkey, created_at, kind, tags, content, sig"
 
+    /** Tag-stream projection: `event_tags` joins `event_headers`, so qualify. */
+    private const val EH_COLS =
+        "event_headers.id, event_headers.pubkey, event_headers.created_at, event_headers.kind, event_headers.tags, event_headers.content, event_headers.sig"
+
     /**
      * Above this many streams, fall back to the single-SQL plan: the
-     * per-stream cursor setup stops paying off, and huge author lists are
-     * collecting a lot no matter what. `kinds.size × authors.size`.
+     * per-stream cursor setup stops paying off, and huge fan-outs are
+     * collecting a lot no matter what. `kinds.size × (authors|values).size`.
      */
     const val MAX_STREAMS = 2048
 
     /**
-     * Stream count if [filter] is merge-eligible, else `-1`. Eligible = a
-     * simple (no tag/search/id/d-tag) query with authors + a limit, whose
-     * per-stream index exists. `kinds` optional: with it, one stream per
-     * `(kind, author)`; without, one per author (needs the pubkey index).
+     * Stream count if [filter] is merge-eligible under *either* shape, else
+     * `-1`. Routing check for [QueryBuilder]; [run] re-derives which shape.
      */
     fun streamCount(
         filter: QueryBuilder.FilterWithDTags,
         indexStrategy: IndexingStrategy,
+    ): Int {
+        val authorStreams = authorStreamCount(filter, indexStrategy)
+        if (authorStreams > 0) return authorStreams
+        return tagStreamCount(filter, indexStrategy)
+    }
+
+    /**
+     * Author-shape stream count, or `-1`. Eligible = a simple (no tag/search/
+     * id/d-tag) query with authors + a limit, whose per-stream index exists.
+     * `kinds` optional: with it, one stream per `(kind, author)`; without, one
+     * per author (needs the pubkey index).
+     */
+    fun authorStreamCount(
+        filter: QueryBuilder.FilterWithDTags,
+        indexStrategy: IndexingStrategy,
     ): Int {
         if (!filter.isSimpleQuery()) return -1
         if (filter.ids != null) return -1
@@ -83,27 +125,69 @@ internal object MergeQueryExecutor {
         // distinct set.
         val distinctAuthors = authors.distinct().size
         val kinds = filter.kinds
+        // Long product: a pathological authors×kinds could overflow Int and
+        // wrap back into the eligible band, routing a huge fan-out here.
         val streams =
             if (kinds != null && kinds.isNotEmpty()) {
-                distinctAuthors * kinds.distinct().size
+                distinctAuthors.toLong() * kinds.distinct().size
             } else {
                 // authors-only needs the (pubkey, created_at) index to stream.
                 if (!indexStrategy.indexEventsByPubkeyAlone) return -1
-                distinctAuthors
+                distinctAuthors.toLong()
             }
         // A single stream is already the optimal single index seek — let the
         // normal path handle it; only merge when there's something to merge.
-        return if (streams in 2..MAX_STREAMS) streams else -1
+        return if (streams in 2..MAX_STREAMS.toLong()) streams.toInt() else -1
     }
 
-    /** Prepares one bound, newest-first cursor per stream. */
-    private fun prepareStreams(
+    /**
+     * Tag-shape stream count, or `-1`. Eligible = a single non-`d` tag key
+     * with `IN` (any-of) semantics and ≥2 distinct values, plus a limit, no
+     * ids/authors/d-tag/search, and no `AND`-tags (`tagsAll`) — the large-IN
+     * watcher shape. `kinds` optional: with it, one stream per
+     * `(value, kind)` off `query_by_tags_hash_kind`; without, one per value
+     * off `query_by_tags_hash` (needs [IndexingStrategy.indexTagsByCreatedAtAlone]).
+     *
+     * Authors are excluded on purpose: `tag ∩ author ∩ kind` is a covered
+     * single seek under [IndexingStrategy.indexTagsWithKindAndPubkey], not a
+     * fan-out, and mixing an author predicate into per-tag streams would not
+     * reduce the read.
+     */
+    fun tagStreamCount(
+        filter: QueryBuilder.FilterWithDTags,
+        indexStrategy: IndexingStrategy,
+    ): Int {
+        if (filter.ids != null) return -1
+        if (filter.authors != null) return -1
+        if (filter.dTags != null) return -1
+        if (filter.search != null && filter.search.isNotEmpty()) return -1
+        if (filter.limit == null || filter.limit <= 0) return -1
+        // AND-tags can't be expressed as a union of per-value streams.
+        if (filter.nonDTagsAll != null && filter.nonDTagsAll.isNotEmpty()) return -1
+        val inTags = filter.nonDTagsIn ?: return -1
+        // A second tag key would AND across keys — not a single union.
+        if (inTags.size != 1) return -1
+        val values = inTags.values.first().distinct()
+        if (values.size < 2) return -1
+        val kinds = filter.kinds?.distinct()?.takeIf { it.isNotEmpty() }
+        val streams =
+            if (kinds != null) {
+                values.size.toLong() * kinds.size
+            } else {
+                if (!indexStrategy.indexTagsByCreatedAtAlone) return -1
+                values.size.toLong()
+            }
+        return if (streams in 2..MAX_STREAMS.toLong()) streams.toInt() else -1
+    }
+
+    /** Prepares one bound, newest-first cursor per author stream. */
+    private fun prepareAuthorStreams(
         db: SQLiteConnection,
         filter: QueryBuilder.FilterWithDTags,
         indexStrategy: IndexingStrategy,
     ): List<SQLiteStatement> {
         // Dedup so a repeated pubkey/kind can't open two identical cursors and
-        // double-emit (see streamCount).
+        // double-emit (see authorStreamCount).
         val authors = filter.authors!!.distinct()
         val kinds = filter.kinds?.distinct()?.takeIf { it.isNotEmpty() }
         val since = filter.since
@@ -117,8 +201,7 @@ internal object MergeQueryExecutor {
         val orderBy =
             if (indexStrategy.useAndIndexIdOnOrderBy) "created_at DESC, id ASC" else "created_at DESC"
 
-        val stmts = ArrayList<SQLiteStatement>((kinds?.size ?: 1) * authors.size)
-        if (kinds != null) {
+        return if (kinds != null) {
             val sql =
                 buildString {
                     append("SELECT ").append(COLS)
@@ -128,16 +211,14 @@ internal object MergeQueryExecutor {
                     if (since != null) append(" AND created_at >= ?")
                     append(" ORDER BY ").append(orderBy)
                 }
-            for (kind in kinds) {
-                for (author in authors) {
-                    val stmt = db.prepare(sql)
-                    var p = 1
-                    stmt.bindLong(p++, kind.toLong())
-                    stmt.bindText(p++, author)
-                    if (until != null) stmt.bindLong(p++, until)
-                    if (since != null) stmt.bindLong(p++, since)
-                    stmts.add(stmt)
-                }
+            buildStreams(kinds.size * authors.size) { i ->
+                val stmt = db.prepare(sql)
+                var p = 1
+                stmt.bindLong(p++, kinds[i / authors.size].toLong())
+                stmt.bindText(p++, authors[i % authors.size])
+                if (until != null) stmt.bindLong(p++, until)
+                if (since != null) stmt.bindLong(p++, since)
+                stmt
             }
         } else {
             val sql =
@@ -149,30 +230,135 @@ internal object MergeQueryExecutor {
                     if (since != null) append(" AND created_at >= ?")
                     append(" ORDER BY ").append(orderBy)
                 }
-            for (author in authors) {
+            buildStreams(authors.size) { i ->
                 val stmt = db.prepare(sql)
                 var p = 1
-                stmt.bindText(p++, author)
+                stmt.bindText(p++, authors[i])
                 if (until != null) stmt.bindLong(p++, until)
                 if (since != null) stmt.bindLong(p++, since)
-                stmts.add(stmt)
+                stmt
+            }
+        }
+    }
+
+    /** Prepares one bound, newest-first cursor per tag-value stream. */
+    private fun prepareTagStreams(
+        db: SQLiteConnection,
+        filter: QueryBuilder.FilterWithDTags,
+        hasher: TagNameValueHasher,
+    ): List<SQLiteStatement> {
+        val entry = filter.nonDTagsIn!!.entries.first()
+        val tagName = entry.key
+        val values = entry.value.distinct()
+        val kinds = filter.kinds?.distinct()?.takeIf { it.isNotEmpty() }
+        val since = filter.since
+        val until = filter.until
+
+        // The tag cursors stream off event_tags (which has no id column), so
+        // the tie order can only be created_at DESC — see the class doc.
+        return if (kinds != null) {
+            val sql =
+                buildString {
+                    append("SELECT ").append(EH_COLS)
+                    append(" FROM event_tags INDEXED BY query_by_tags_hash_kind")
+                    append(" JOIN event_headers ON event_headers.row_id = event_tags.event_header_row_id")
+                    append(" WHERE event_tags.tag_hash = ? AND event_tags.kind = ?")
+                    if (until != null) append(" AND event_tags.created_at <= ?")
+                    if (since != null) append(" AND event_tags.created_at >= ?")
+                    append(" ORDER BY event_tags.created_at DESC")
+                }
+            buildStreams(values.size * kinds.size) { i ->
+                val stmt = db.prepare(sql)
+                var p = 1
+                stmt.bindLong(p++, hasher.hash(tagName, values[i / kinds.size]))
+                stmt.bindLong(p++, kinds[i % kinds.size].toLong())
+                if (until != null) stmt.bindLong(p++, until)
+                if (since != null) stmt.bindLong(p++, since)
+                stmt
+            }
+        } else {
+            val sql =
+                buildString {
+                    append("SELECT ").append(EH_COLS)
+                    append(" FROM event_tags INDEXED BY query_by_tags_hash")
+                    append(" JOIN event_headers ON event_headers.row_id = event_tags.event_header_row_id")
+                    append(" WHERE event_tags.tag_hash = ?")
+                    if (until != null) append(" AND event_tags.created_at <= ?")
+                    if (since != null) append(" AND event_tags.created_at >= ?")
+                    append(" ORDER BY event_tags.created_at DESC")
+                }
+            buildStreams(values.size) { i ->
+                val stmt = db.prepare(sql)
+                var p = 1
+                stmt.bindLong(p++, hasher.hash(tagName, values[i]))
+                if (until != null) stmt.bindLong(p++, until)
+                if (since != null) stmt.bindLong(p++, since)
+                stmt
             }
         }
-        return stmts
     }
 
     /**
-     * Runs the merge, calling [onRow] with each winning cursor positioned on
-     * the row to emit, newest-first, up to `limit`. [onRow] must read the
-     * current row (it stays valid until the next step).
+     * Runs the merge for whichever shape [filter] matches, calling [onRow]
+     * with each winning cursor positioned on the row to emit, newest-first,
+     * up to `limit`. [onRow] must read the current row (it stays valid until
+     * the next step). [hasher] is only consulted for the tag shape.
      */
     fun run(
         db: SQLiteConnection,
         filter: QueryBuilder.FilterWithDTags,
         indexStrategy: IndexingStrategy,
+        hasher: (SQLiteConnection) -> TagNameValueHasher,
+        onRow: (SQLiteStatement) -> Unit,
+    ) {
+        if (authorStreamCount(filter, indexStrategy) > 0) {
+            // One pubkey per event ⇒ author streams never overlap: no dedup.
+            mergeStreams(prepareAuthorStreams(db, filter, indexStrategy), filter.limit!!, dedup = false, onRow)
+        } else {
+            // A single event can match several tag values ⇒ dedup by id.
+            mergeStreams(prepareTagStreams(db, filter, hasher(db)), filter.limit!!, dedup = true, onRow)
+        }
+    }
+
+    /**
+     * Prepares [count] cursors via [prepareOne], closing any already-prepared
+     * statements if a later prepare throws — otherwise a mid-loop failure would
+     * strand checked-out, un-reset handles in the pooled connection (dead
+     * slots holding read locks). On success the caller ([mergeStreams]) owns
+     * closing them.
+     */
+    private inline fun buildStreams(
+        count: Int,
+        prepareOne: (Int) -> SQLiteStatement,
+    ): List<SQLiteStatement> {
+        val stmts = ArrayList<SQLiteStatement>(count)
+        try {
+            for (i in 0 until count) stmts.add(prepareOne(i))
+        } catch (e: Throwable) {
+            for (s in stmts) {
+                try {
+                    s.close()
+                } catch (_: Throwable) {
+                }
+            }
+            throw e
+        }
+        return stmts
+    }
+
+    /**
+     * Heap-free k-way merge over the prepared [stmts]: repeatedly emits the
+     * newest live head (`created_at DESC`, tie `id ASC`) until [limit] rows
+     * are emitted or every stream is drained. When [dedup] is set an event id
+     * already emitted is skipped (its cursor still advances), so a row that
+     * surfaces in several streams is emitted once.
+     */
+    private fun mergeStreams(
+        stmts: List<SQLiteStatement>,
+        limit: Int,
+        dedup: Boolean,
         onRow: (SQLiteStatement) -> Unit,
     ) {
-        val stmts = prepareStreams(db, filter, indexStrategy)
         try {
             val k = stmts.size
             val headCreatedAt = LongArray(k)
@@ -188,8 +374,8 @@ internal object MergeQueryExecutor {
                 }
             }
 
+            val seen = if (dedup) HashSet<String>() else null
             var emitted = 0
-            val limit = filter.limit!!
             while (emitted < limit) {
                 // Pick the newest live head: created_at DESC, then id ASC.
                 var best = -1
@@ -204,8 +390,11 @@ internal object MergeQueryExecutor {
                 }
                 if (best == -1) break
 
-                onRow(stmts[best]) // cursor is still on the head row
-                emitted++
+                // Emit unless this id was already emitted by another stream.
+                if (seen == null || seen.add(headId[best]!!)) {
+                    onRow(stmts[best]) // cursor is still on the head row
+                    emitted++
+                }
 
                 // Advance the winner to its next row.
                 if (stmts[best].step()) {

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/QueryBuilder.kt

@@ -49,7 +49,7 @@ class QueryBuilder(
         val merge = filter.toFilterWithDTags()
         if (MergeQueryExecutor.streamCount(merge, indexStrategy) > 0) {
             val out = ArrayList<T>(merge.limit!!)
-            MergeQueryExecutor.run(db, merge, indexStrategy) { out.add(it.toEvent()) }
+            MergeQueryExecutor.run(db, merge, indexStrategy, hasher) { out.add(it.toEvent()) }
             return out
         }
         return db.runQuery(toSql(filter, hasher(db)))
@@ -62,7 +62,7 @@ class QueryBuilder(
     ) {
         val merge = filter.toFilterWithDTags()
         if (MergeQueryExecutor.streamCount(merge, indexStrategy) > 0) {
-            MergeQueryExecutor.run(db, merge, indexStrategy) { onEach(it.toEvent()) }
+            MergeQueryExecutor.run(db, merge, indexStrategy, hasher) { onEach(it.toEvent()) }
             return
         }
         db.runQuery(toSql(filter, hasher(db)), onEach)
@@ -102,7 +102,7 @@ class QueryBuilder(
         val merge = filter.toFilterWithDTags()
         if (MergeQueryExecutor.streamCount(merge, indexStrategy) > 0) {
             val out = ArrayList<RawEvent>(merge.limit!!)
-            MergeQueryExecutor.run(db, merge, indexStrategy) { out.add(it.toRawEvent()) }
+            MergeQueryExecutor.run(db, merge, indexStrategy, hasher) { out.add(it.toRawEvent()) }
             return out
         }
         return db.runRawQuery(toSql(filter, hasher(db)))
@@ -115,7 +115,7 @@ class QueryBuilder(
     ) {
         val merge = filter.toFilterWithDTags()
         if (MergeQueryExecutor.streamCount(merge, indexStrategy) > 0) {
-            MergeQueryExecutor.run(db, merge, indexStrategy) { onEach(it.toRawEvent()) }
+            MergeQueryExecutor.run(db, merge, indexStrategy, hasher) { onEach(it.toRawEvent()) }
             return
         }
         db.runRawQuery(toSql(filter, hasher(db)), onEach)
@@ -202,13 +202,17 @@ class QueryBuilder(
             )
         }
 
-        val rowIdSubqueries = prepareRowIDSubQueries(filter, hasher)
+        // A search term that survived the simple-search branch above always
+        // combines with tags here (search + #t etc.). NIP-50 still orders by
+        // relevance, so carry the FTS rank through the subquery and out.
+        val rankSearch = newFilter.search != null && newFilter.search.isNotEmpty()
+        val rowIdSubqueries = prepareRowIDSubQueries(filter, hasher, projectRank = rankSearch)
 
         return if (rowIdSubqueries == null) {
             QuerySpec(makeEverythingQuery())
         } else {
             QuerySpec(
-                makeQueryIn(rowIdSubqueries.sql),
+                makeQueryIn(rowIdSubqueries.sql, orderByRank = rankSearch),
                 rowIdSubqueries.args,
             )
         }
@@ -220,7 +224,13 @@ class QueryBuilder(
     ): QuerySpec {
         if (filters.size == 1) return toSql(filters.first(), hasher)
 
-        val rowIdSubqueries = unionSubqueriesIfNeeded(filters, hasher)
+        // A multi-filter search REQ (e.g. the client's search-across-kinds,
+        // all filters sharing one term) must still be NIP-50 relevance-ordered.
+        // Only when EVERY branch is a search branch (and FTS is on, so each has
+        // a rank column) — a non-search branch has no defined relevance, so a
+        // mixed REQ falls back to created_at.
+        val rankSearch = fts.enabled && filters.all { !it.search.isNullOrEmpty() }
+        val rowIdSubqueries = unionSubqueriesIfNeeded(filters, hasher, projectRank = rankSearch)
 
         return if (rowIdSubqueries == null) {
             QuerySpec(
@@ -229,7 +239,7 @@ class QueryBuilder(
             )
         } else {
             QuerySpec(
-                makeQueryIn(rowIdSubqueries.sql),
+                makeQueryIn(rowIdSubqueries.sql, orderByRank = rankSearch),
                 rowIdSubqueries.args,
             )
         }
@@ -427,7 +437,7 @@ class QueryBuilder(
         val sql =
             buildString {
                 append("SELECT event_headers.id, event_headers.created_at FROM event_headers")
-                append("\nINNER JOIN ${fts.tableName} ON event_headers.row_id = ${fts.tableName}.${fts.eventHeaderRowIdName}")
+                append("\nINNER JOIN ${fts.tableName} ON event_headers.row_id = ${fts.tableName}.rowid")
                 if (clause.conditions.isNotEmpty()) {
                     append("\nWHERE ${clause.conditions}")
                 }
@@ -470,14 +480,20 @@ class QueryBuilder(
 
     private fun makeEverythingQuery() = "SELECT id, pubkey, created_at, kind, tags, content, sig FROM event_headers ORDER BY created_at DESC${if (indexStrategy.useAndIndexIdOnOrderBy) ", id ASC" else ""}"
 
-    private fun makeQueryIn(rowIdQuery: String) =
-        """
+    // [orderByRank] presents the joined result in NIP-50 relevance order: the
+    // subquery (built with `projectRank`) exposes the FTS bm25 score as a
+    // `rank` column, and `created_at DESC` is only a tie-break. Off, it keeps
+    // the default newest-first ordering for every non-search shape.
+    private fun makeQueryIn(
+        rowIdQuery: String,
+        orderByRank: Boolean = false,
+    ) = """
         SELECT id, pubkey, created_at, kind, tags, content, sig FROM event_headers
         INNER JOIN (
             $rowIdQuery
         ) AS filtered
         ON event_headers.row_id = filtered.row_id
-        ORDER BY created_at DESC${if (indexStrategy.useAndIndexIdOnOrderBy) ", id ASC" else ""}
+        ORDER BY ${if (orderByRank) "filtered.rank, " else ""}created_at DESC${if (indexStrategy.useAndIndexIdOnOrderBy) ", id ASC" else ""}
         """.trimIndent()
 
     private fun <T : Event> SQLiteConnection.runQuery(query: QuerySpec): List<T> =
@@ -713,16 +729,34 @@ class QueryBuilder(
     fun unionSubqueriesIfNeeded(
         filters: List<Filter>,
         hasher: TagNameValueHasher,
+        // See [prepareRowIDSubQueries]. When set, every branch is a search
+        // branch that also projects a `rank` column; the union keeps one row
+        // per event with its BEST (min) bm25 score so the caller can present
+        // the whole multi-filter search REQ NIP-50-ranked. Callers must only
+        // pass true when all filters carry a search term (else a branch has no
+        // rank column). Off for count/delete, which stay single-column.
+        projectRank: Boolean = false,
     ): QuerySpec? {
         val inner =
             filters.mapNotNull { filter ->
-                prepareRowIDSubQueries(filter, hasher)
+                prepareRowIDSubQueries(filter, hasher, projectRank)
             }
 
         if (inner.isEmpty()) return null
 
-        return if (inner.size == 1) {
-            inner.first()
+        if (inner.size == 1) return inner.first()
+
+        return if (projectRank) {
+            // UNION ALL keeps every (row_id, rank) so an event matching two
+            // branches under different terms isn't dropped before MIN; GROUP BY
+            // then dedups by event keeping the best score.
+            QuerySpec(
+                sql =
+                    "SELECT row_id, MIN(rank) as rank FROM (\n            " +
+                        inner.joinToString("\n            UNION ALL\n            ") { "SELECT row_id, rank FROM (${it.sql})" } +
+                        "\n            ) GROUP BY row_id",
+                args = inner.flatMap { it.args },
+            )
         } else {
             QuerySpec(
                 sql = inner.joinToString("\n            UNION\n            ") { "SELECT row_id FROM (${it.sql})" },
@@ -748,6 +782,13 @@ class QueryBuilder(
     fun prepareRowIDSubQueries(
         filter: Filter,
         hasher: TagNameValueHasher,
+        // When set on a search filter, the subquery also projects the FTS
+        // bm25 score as a `rank` column and orders its own LIMIT by relevance,
+        // so the caller ([makeQueryIn] with `orderByRank`) can present results
+        // NIP-50-ranked. Off (the default) for count/delete/union/negentropy,
+        // which never expose a second column (a two-column subquery breaks
+        // `row_id IN (…)`) and don't rank.
+        projectRank: Boolean = false,
     ): QuerySpec? {
         if (filter.isEmpty()) return null
 
@@ -761,6 +802,10 @@ class QueryBuilder(
 
         val mustJoinSearch = filter.search != null && fts.enabled
 
+        // Only emit the rank column when there is actually an FTS join to take
+        // it from; a `projectRank` request on a tag-only filter is ignored.
+        val emitRank = projectRank && mustJoinSearch
+
         val nonDTagsIn = filter.tags?.filter { it.key != "d" } ?: emptyMap()
 
         val nonDTagsAll = filter.tagsAll?.filter { it.key != "d" } ?: emptyMap()
@@ -789,7 +834,11 @@ class QueryBuilder(
             buildString {
                 // always do tags if there are any
                 if (reverseLookup) {
-                    append("SELECT DISTINCT(event_tags.event_header_row_id) as row_id FROM event_tags")
+                    append("SELECT DISTINCT(event_tags.event_header_row_id) as row_id")
+                    // rank is functionally determined by the row_id (one FTS
+                    // row per event), so it doesn't change what DISTINCT folds.
+                    if (emitRank) append(", ${fts.tableName}.rank as rank")
+                    append(" FROM event_tags")
 
                     // it's quite rare to have 2 tags in the filter, but possible
                     nonDTagsIn.keys.forEachIndexed { index, tagName ->
@@ -815,13 +864,15 @@ class QueryBuilder(
                     }
 
                     if (mustJoinSearch) {
-                        append(" INNER JOIN ${fts.tableName} ON ${fts.tableName}.${fts.eventHeaderRowIdName} = event_tags.event_header_row_id")
+                        append(" INNER JOIN ${fts.tableName} ON ${fts.tableName}.rowid = event_tags.event_header_row_id")
                     }
                 } else if (mustJoinSearch) {
-                    append("SELECT ${fts.tableName}.${fts.eventHeaderRowIdName} as row_id FROM ${fts.tableName}")
+                    append("SELECT ${fts.tableName}.rowid as row_id")
+                    if (emitRank) append(", ${fts.tableName}.rank as rank")
+                    append(" FROM ${fts.tableName}")
 
                     if (hasHeaders) {
-                        append(" INNER JOIN event_headers ON event_headers.row_id = ${fts.tableName}.${fts.eventHeaderRowIdName}")
+                        append(" INNER JOIN event_headers ON event_headers.row_id = ${fts.tableName}.rowid")
                     }
                 } else {
                     // no tags and no search.
@@ -934,15 +985,17 @@ class QueryBuilder(
                     append(" WHERE ${clause.conditions}")
                 }
                 if (filter.limit != null) {
-                    if (reverseLookup) {
+                    if (emitRank) {
+                        // NIP-50: the LIMIT keeps the most RELEVANT rows, not
+                        // the newest, so the inner cut is by rank too.
+                        append(" ORDER BY rank")
+                    } else if (reverseLookup) {
                         append(" ORDER BY event_tags.created_at DESC")
-                        append(" LIMIT ")
-                        append(filter.limit)
                     } else {
                         append(" ORDER BY event_headers.created_at DESC")
-                        append(" LIMIT ")
-                        append(filter.limit)
                     }
+                    append(" LIMIT ")
+                    append(filter.limit)
                 }
             }
 
@@ -1001,11 +1054,17 @@ class QueryBuilder(
         val sql =
             buildString {
                 append("SELECT event_headers.id, event_headers.pubkey, event_headers.created_at, event_headers.kind, event_headers.tags, event_headers.content, event_headers.sig FROM event_headers")
-                append("\nINNER JOIN ${fts.tableName} ON event_headers.row_id = ${fts.tableName}.${fts.eventHeaderRowIdName}")
+                append("\nINNER JOIN ${fts.tableName} ON event_headers.row_id = ${fts.tableName}.rowid")
                 if (clause.conditions.isNotEmpty()) {
                     append("\nWHERE ${clause.conditions}")
                 }
-                append("\nORDER BY event_headers.created_at DESC")
+                // NIP-50: search results are ordered by relevance ("quality of
+                // search result"), not created_at, and the limit is applied
+                // after the score. FTS5 exposes bm25 as the `rank` column (more
+                // negative = more relevant), so ORDER BY rank ascending is
+                // best-match-first. created_at DESC is only a tie-break so
+                // equally-relevant matches come newest-first deterministically.
+                append("\nORDER BY ${fts.tableName}.rank, event_headers.created_at DESC")
                 if (indexStrategy.useAndIndexIdOnOrderBy) {
                     append(", event_headers.id ASC")
                 }

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/SQLiteEventStore.kt

@@ -60,7 +60,7 @@ class SQLiteEventStore(
     val extraPragmas: List<String> = emptyList(),
 ) {
     companion object {
-        const val DATABASE_VERSION = 4
+        const val DATABASE_VERSION = 5
     }
 
     val seedModule = SeedModule()
@@ -160,6 +160,11 @@ class SQLiteEventStore(
                         setUserVersion(this, DATABASE_VERSION)
                     }
                 }
+                // Flag-gated indexes are runtime config, not schema: a
+                // deployment that flips an IndexingStrategy flag on an
+                // existing DB gets the index built here (idempotent,
+                // one-time cost), with no user_version bump involved.
+                eventIndexModule.ensureOptionalIndexes(db)
             },
         )
     }
@@ -208,6 +213,12 @@ class SQLiteEventStore(
                     // watermark seeds at the current MAX(row_id).
                     fullTextSearchModule.createStateTable(db)
                 }
+                4 -> {
+                    // Upgrade from version 4 to 5: the FTS index became a
+                    // contentless table keyed by event_headers.row_id. The old
+                    // rowids can't be remapped, so drop and repopulate.
+                    fullTextSearchModule.migrateV4ToContentless(db)
+                }
             }
         }
     }
@@ -269,6 +280,12 @@ class SQLiteEventStore(
         pool.useWriter { db ->
             db.execSQL("PRAGMA analysis_limit = 400;")
             db.execSQL("PRAGMA optimize;")
+            // Fold a bounded FTS segment merge into the same periodic
+            // maintenance tick: incremental (and deferred catch-up) inserts
+            // leave the NIP-50 index as many small segments, and a MATCH
+            // queries every one. Bounded so the tick stays cheap; a no-op when
+            // there is nothing to merge or FTS is off.
+            fullTextSearchModule.mergeSegments(db)
         }
 
     /**

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/StatementCachingConnection.kt

@@ -34,46 +34,72 @@ import androidx.sqlite.SQLiteStatement
  * real reset + clearBindings happens on the next checkout). Statements are
  * only truly finalized when the connection itself closes.
  *
+ * Each SQL string caches a small **pool** of handles rather than a single
+ * one, so overlapping checkouts of the *same* SQL all reuse cached handles.
+ * That is exactly the k-way-merge query shape ([MergeQueryExecutor]): it
+ * opens one identical-SQL cursor per author/tag stream — dozens to hundreds
+ * live at once — which a single-handle cache could not serve (every stream
+ * past the first fell back to an uncached prepare). The pool lets a repeated
+ * follow-feed / reactions-watcher REQ reuse its per-stream cursors instead
+ * of re-preparing them each poll.
+ *
  * Constraints, by design of the call sites:
  *  - **Not thread-safe** — same contract as the underlying connection,
- *    which the pool already serializes (single writer under a mutex).
- *  - **No overlapping use of the same SQL** — checking out one SQL string
- *    twice without closing the first use would alias one native handle.
- *    Insert/query paths never nest the same statement; a checkout while
- *    the previous one is still open falls back to an uncached statement.
+ *    which the pool already serializes (single writer under a mutex; each
+ *    reader held by one coroutine at a time).
  */
 class StatementCachingConnection(
     private val delegate: SQLiteConnection,
     /**
-     * Ceiling on retained statements. Query SQL embeds one `?` per filter
-     * element, so shape variety is client-controlled — without a cap a
-     * long-lived relay connection would accumulate native handles without
-     * bound. Once full, unseen SQL just prepares uncached. 256 comfortably
-     * covers the write path's fixed set plus the recurring filter shapes.
+     * Ceiling on retained statements across all SQL strings. Query SQL
+     * embeds one `?` per filter element, so shape variety is
+     * client-controlled — without a cap a long-lived relay connection would
+     * accumulate native handles without bound. Once full, unseen SQL (or an
+     * extra concurrent copy of a cached SQL) just prepares uncached. 512
+     * covers the write path's fixed set, the recurring single-shot filter
+     * shapes, and a few hundred concurrent per-stream merge cursors.
      */
-    private val maxCachedStatements: Int = 256,
+    private val maxCachedStatements: Int = 512,
 ) : SQLiteConnection by delegate {
-    private val cache = HashMap<String, CachedStatement>()
+    // One reusable pool per SQL string. Several entries of the same pool may
+    // be checked out simultaneously (the merge path); a `prepare` reuses the
+    // first free entry, grows the pool while under the global cap, and only
+    // then falls back to an uncached statement.
+    private val cache = HashMap<String, ArrayList<CachedStatement>>()
+    private var cachedCount = 0
 
     override fun prepare(sql: String): SQLiteStatement {
-        val cached =
-            cache[sql] ?: run {
-                if (cache.size >= maxCachedStatements) return delegate.prepare(sql)
-                CachedStatement(delegate.prepare(sql)).also { cache[sql] = it }
+        val pool = cache[sql]
+        if (pool != null) {
+            for (i in pool.indices) {
+                val stmt = pool[i]
+                if (!stmt.checkedOut) {
+                    stmt.checkedOut = true
+                    stmt.clearBindings()
+                    return stmt
+                }
+            }
+            // Every pooled handle for this SQL is in use — grow if the global
+            // budget allows, else serve an uncached statement.
+            if (cachedCount >= maxCachedStatements) return delegate.prepare(sql)
+            return CachedStatement(delegate.prepare(sql)).also {
+                it.checkedOut = true
+                pool.add(it)
+                cachedCount++
             }
-        if (cached.checkedOut) {
-            // Same SQL prepared while the previous handle is still in use —
-            // stay correct with a plain uncached statement.
-            return delegate.prepare(sql)
         }
-        cached.checkedOut = true
-        cached.clearBindings()
-        return cached
+        if (cachedCount >= maxCachedStatements) return delegate.prepare(sql)
+        return CachedStatement(delegate.prepare(sql)).also {
+            it.checkedOut = true
+            cache[sql] = arrayListOf(it)
+            cachedCount++
+        }
     }
 
     override fun close() {
-        cache.values.forEach { runCatching { it.finalize() } }
+        cache.values.forEach { pool -> pool.forEach { runCatching { it.finalizeStatement() } } }
         cache.clear()
+        cachedCount = 0
         delegate.close()
     }
 
@@ -94,6 +120,10 @@ class StatementCachingConnection(
             checkedOut = false
         }
 
-        fun finalize() = delegate.close()
+        // Not named `finalize`: a no-arg `finalize()` is treated by the JVM as
+        // Object.finalize(), so the GC would call it and double-close the
+        // native handle after our explicit close(). This is only ever invoked
+        // explicitly from the connection's close().
+        fun finalizeStatement() = delegate.close()
     }
 }

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip50Search/SearchQuery.kt

@@ -202,8 +202,20 @@ fun Filter.strippingSearchExtensions(): Filter {
 /**
  * Applies [strippingSearchExtensions] to every filter, returning this
  * same list when no filter carried extension tokens.
+ *
+ * This runs on every REQ/COUNT/snapshot, and the overwhelming majority
+ * carry no `search` term at all, so the no-search case must not allocate:
+ * bail before building any list when nothing could be stripped.
  */
 fun List<Filter>.strippingSearchExtensions(): List<Filter> {
+    var hasSearch = false
+    for (i in indices) {
+        if (!this[i].search.isNullOrEmpty()) {
+            hasSearch = true
+            break
+        }
+    }
+    if (!hasSearch) return this
     var changed = false
     val out =
         map {

quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/QueryAssemblerTest.kt

@@ -238,9 +238,9 @@ class QueryAssemblerTest : BaseDBTest() {
                     INNER JOIN (
                         SELECT row_id FROM (SELECT event_headers.row_id as row_id FROM event_headers ORDER BY event_headers.created_at DESC LIMIT 10)
                         UNION
-                        SELECT row_id FROM (SELECT event_fts.event_header_row_id as row_id FROM event_fts INNER JOIN event_headers ON event_headers.row_id = event_fts.event_header_row_id WHERE (event_headers.kind IN ("1", "1111")) AND (event_headers.pubkey = "7c5eb72a4584fdaaeaa145b25c92ea9917704224951219dbd43acef9e91fb88d") AND (event_fts MATCH "keywords") ORDER BY event_headers.created_at DESC LIMIT 100)
+                        SELECT row_id FROM (SELECT event_fts.rowid as row_id FROM event_fts INNER JOIN event_headers ON event_headers.row_id = event_fts.rowid WHERE (event_headers.kind IN ("1", "1111")) AND (event_headers.pubkey = "7c5eb72a4584fdaaeaa145b25c92ea9917704224951219dbd43acef9e91fb88d") AND (event_fts MATCH "keywords") ORDER BY event_headers.created_at DESC LIMIT 100)
                         UNION
-                        SELECT row_id FROM (SELECT event_fts.event_header_row_id as row_id FROM event_fts INNER JOIN event_headers ON event_headers.row_id = event_fts.event_header_row_id WHERE (event_headers.kind = "20") AND (event_fts MATCH "cats") ORDER BY event_headers.created_at DESC LIMIT 30)
+                        SELECT row_id FROM (SELECT event_fts.rowid as row_id FROM event_fts INNER JOIN event_headers ON event_headers.row_id = event_fts.rowid WHERE (event_headers.kind = "20") AND (event_fts MATCH "cats") ORDER BY event_headers.created_at DESC LIMIT 30)
                     ) AS filtered
                     ON event_headers.row_id = filtered.row_id
                     ORDER BY $orderBy
@@ -252,13 +252,13 @@ class QueryAssemblerTest : BaseDBTest() {
                     │       │   └── SCAN (subquery-1)
                     │       ├── UNION USING TEMP B-TREE
                     │       │   ├── CO-ROUTINE (subquery-3)
-                    │       │   │   ├── SCAN event_fts VIRTUAL TABLE INDEX 0:M2
+                    │       │   │   ├── SCAN event_fts VIRTUAL TABLE INDEX 0:M1
                     │       │   │   ├── SEARCH event_headers USING INTEGER PRIMARY KEY (rowid=?)
                     │       │   │   └── USE TEMP B-TREE FOR ORDER BY
                     │       │   └── SCAN (subquery-3)
                     │       └── UNION USING TEMP B-TREE
                     │           ├── CO-ROUTINE (subquery-5)
-                    │           │   ├── SCAN event_fts VIRTUAL TABLE INDEX 0:M2
+                    │           │   ├── SCAN event_fts VIRTUAL TABLE INDEX 0:M1
                     │           │   ├── SEARCH event_headers USING INTEGER PRIMARY KEY (rowid=?)
                     │           │   └── USE TEMP B-TREE FOR ORDER BY
                     │           └── SCAN (subquery-5)
@@ -275,9 +275,9 @@ class QueryAssemblerTest : BaseDBTest() {
                     INNER JOIN (
                         SELECT row_id FROM (SELECT event_headers.row_id as row_id FROM event_headers ORDER BY event_headers.created_at DESC LIMIT 10)
                         UNION
-                        SELECT row_id FROM (SELECT event_fts.event_header_row_id as row_id FROM event_fts INNER JOIN event_headers ON event_headers.row_id = event_fts.event_header_row_id WHERE (event_headers.kind IN ("1", "1111")) AND (event_headers.pubkey = "7c5eb72a4584fdaaeaa145b25c92ea9917704224951219dbd43acef9e91fb88d") AND (event_fts MATCH "keywords") ORDER BY event_headers.created_at DESC LIMIT 100)
+                        SELECT row_id FROM (SELECT event_fts.rowid as row_id FROM event_fts INNER JOIN event_headers ON event_headers.row_id = event_fts.rowid WHERE (event_headers.kind IN ("1", "1111")) AND (event_headers.pubkey = "7c5eb72a4584fdaaeaa145b25c92ea9917704224951219dbd43acef9e91fb88d") AND (event_fts MATCH "keywords") ORDER BY event_headers.created_at DESC LIMIT 100)
                         UNION
-                        SELECT row_id FROM (SELECT event_fts.event_header_row_id as row_id FROM event_fts INNER JOIN event_headers ON event_headers.row_id = event_fts.event_header_row_id WHERE (event_headers.kind = "20") AND (event_fts MATCH "cats") ORDER BY event_headers.created_at DESC LIMIT 30)
+                        SELECT row_id FROM (SELECT event_fts.rowid as row_id FROM event_fts INNER JOIN event_headers ON event_headers.row_id = event_fts.rowid WHERE (event_headers.kind = "20") AND (event_fts MATCH "cats") ORDER BY event_headers.created_at DESC LIMIT 30)
                     ) AS filtered
                     ON event_headers.row_id = filtered.row_id
                     ORDER BY $orderBy
@@ -290,13 +290,13 @@ class QueryAssemblerTest : BaseDBTest() {
                     │       │   └── SCAN (subquery-1)
                     │       ├── UNION USING TEMP B-TREE
                     │       │   ├── CO-ROUTINE (subquery-3)
-                    │       │   │   ├── SCAN event_fts VIRTUAL TABLE INDEX 0:M2
+                    │       │   │   ├── SCAN event_fts VIRTUAL TABLE INDEX 0:M1
                     │       │   │   ├── SEARCH event_headers USING INTEGER PRIMARY KEY (rowid=?)
                     │       │   │   └── USE TEMP B-TREE FOR ORDER BY
                     │       │   └── SCAN (subquery-3)
                     │       └── UNION USING TEMP B-TREE
                     │           ├── CO-ROUTINE (subquery-5)
-                    │           │   ├── SCAN event_fts VIRTUAL TABLE INDEX 0:M2
+                    │           │   ├── SCAN event_fts VIRTUAL TABLE INDEX 0:M1
                     │           │   ├── SEARCH event_headers USING INTEGER PRIMARY KEY (rowid=?)
                     │           │   └── USE TEMP B-TREE FOR ORDER BY
                     │           └── SCAN (subquery-5)
@@ -712,10 +712,10 @@ class QueryAssemblerTest : BaseDBTest() {
                 assertEquals(
                     """
                     SELECT event_headers.id, event_headers.pubkey, event_headers.created_at, event_headers.kind, event_headers.tags, event_headers.content, event_headers.sig FROM event_headers
-                    INNER JOIN event_fts ON event_headers.row_id = event_fts.event_header_row_id
+                    INNER JOIN event_fts ON event_headers.row_id = event_fts.rowid
                     WHERE (event_fts MATCH "keywords") AND (event_headers.pubkey IN ("7c5eb72a4584fdaaeaa145b25c92ea9917704224951219dbd43acef9e91fb88d", "f3ac434d61bc0f491a814782ccfdf9c439dae1f0bde9097ad4a245f4c495cd14", "12ae0fd81c85e1e7d9ed096397dc3129849425fe6f8afce7213ebf38ddfc6ca9"))
-                    ORDER BY event_headers.created_at DESC, event_headers.id ASC
-                    ├── SCAN event_fts VIRTUAL TABLE INDEX 0:M2
+                    ORDER BY event_fts.rank, event_headers.created_at DESC, event_headers.id ASC
+                    ├── SCAN event_fts VIRTUAL TABLE INDEX 0:M1
                     ├── SEARCH event_headers USING INTEGER PRIMARY KEY (rowid=?)
                     └── USE TEMP B-TREE FOR ORDER BY
                     """.trimIndent(),
@@ -725,10 +725,10 @@ class QueryAssemblerTest : BaseDBTest() {
                 assertEquals(
                     """
                     SELECT event_headers.id, event_headers.pubkey, event_headers.created_at, event_headers.kind, event_headers.tags, event_headers.content, event_headers.sig FROM event_headers
-                    INNER JOIN event_fts ON event_headers.row_id = event_fts.event_header_row_id
+                    INNER JOIN event_fts ON event_headers.row_id = event_fts.rowid
                     WHERE (event_fts MATCH "keywords") AND (event_headers.pubkey IN ("7c5eb72a4584fdaaeaa145b25c92ea9917704224951219dbd43acef9e91fb88d", "f3ac434d61bc0f491a814782ccfdf9c439dae1f0bde9097ad4a245f4c495cd14", "12ae0fd81c85e1e7d9ed096397dc3129849425fe6f8afce7213ebf38ddfc6ca9"))
-                    ORDER BY event_headers.created_at DESC
-                    ├── SCAN event_fts VIRTUAL TABLE INDEX 0:M2
+                    ORDER BY event_fts.rank, event_headers.created_at DESC
+                    ├── SCAN event_fts VIRTUAL TABLE INDEX 0:M1
                     ├── SEARCH event_headers USING INTEGER PRIMARY KEY (rowid=?)
                     └── USE TEMP B-TREE FOR ORDER BY
                     """.trimIndent(),
@@ -741,14 +741,14 @@ class QueryAssemblerTest : BaseDBTest() {
     fun testKindAndSearch() =
         forEachDB { db ->
             val filter = Filter(kinds = listOf(1, 1111, 10000), search = "keywords")
-            val orderBy = if (db.indexStrategy.useAndIndexIdOnOrderBy) "event_headers.created_at DESC, event_headers.id ASC" else "event_headers.created_at DESC"
+            val orderBy = if (db.indexStrategy.useAndIndexIdOnOrderBy) "event_fts.rank, event_headers.created_at DESC, event_headers.id ASC" else "event_fts.rank, event_headers.created_at DESC"
             assertEquals(
                 """
                 SELECT event_headers.id, event_headers.pubkey, event_headers.created_at, event_headers.kind, event_headers.tags, event_headers.content, event_headers.sig FROM event_headers
-                INNER JOIN event_fts ON event_headers.row_id = event_fts.event_header_row_id
+                INNER JOIN event_fts ON event_headers.row_id = event_fts.rowid
                 WHERE (event_fts MATCH "keywords") AND (event_headers.kind IN ("1", "1111", "10000"))
                 ORDER BY $orderBy
-                ├── SCAN event_fts VIRTUAL TABLE INDEX 0:M2
+                ├── SCAN event_fts VIRTUAL TABLE INDEX 0:M1
                 ├── SEARCH event_headers USING INTEGER PRIMARY KEY (rowid=?)
                 └── USE TEMP B-TREE FOR ORDER BY
                 """.trimIndent(),

quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/SearchRelevanceOrderTest.kt

@@ -0,0 +1,202 @@
+/*
+ * 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.store.sqlite
+
+import com.vitorpamplona.quartz.nip01Core.core.Event
+import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
+import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerSync
+import com.vitorpamplona.quartz.nip10Notes.TextNoteEvent
+import com.vitorpamplona.quartz.utils.EventFactory
+import kotlinx.coroutines.runBlocking
+import kotlin.test.Test
+import kotlin.test.assertEquals
+
+/**
+ * NIP-50: search results are ordered by "quality of search result" (relevance),
+ * **not** by `created_at`, and the limit is applied after the score. The store
+ * uses FTS5 bm25 (`ORDER BY event_fts.rank`), so a stronger match outranks a
+ * newer one.
+ */
+class SearchRelevanceOrderTest {
+    private val signer = NostrSignerSync()
+
+    private fun note(
+        content: String,
+        createdAt: Long,
+    ) = signer.sign(TextNoteEvent.build(content, createdAt = createdAt))
+
+    @Test
+    fun strongerMatchOutranksNewer() =
+        runBlocking {
+            val store = EventStore(dbName = null)
+            try {
+                // Older, but the term appears 3× in a short doc → most relevant.
+                val strong = note("bitcoin bitcoin bitcoin", createdAt = 1_000)
+                // Newer, term once buried in a long doc → least relevant.
+                val weak = note("bitcoin is one topic among many other unrelated words here padding", createdAt = 9_000)
+                // Newer still, but does not match at all.
+                val nonMatch = note("completely different subject entirely", createdAt = 9_999)
+
+                store.insert(weak)
+                store.insert(strong)
+                store.insert(nonMatch)
+
+                val results = store.query<Event>(Filter(search = "bitcoin", limit = 10)).map { it.id }
+                // Relevance order (strong before weak) — the opposite of
+                // created_at DESC (which would put weak first) — and the
+                // non-matching note is absent.
+                assertEquals(listOf(strong.id, weak.id), results)
+            } finally {
+                store.close()
+            }
+        }
+
+    @Test
+    fun limitAppliesAfterRelevanceScore() =
+        runBlocking {
+            val store = EventStore(dbName = null)
+            try {
+                // Three docs of decreasing relevance but increasing created_at,
+                // so created_at order and relevance order are exact opposites.
+                val best = note("apple apple apple apple", createdAt = 1)
+                val mid = note("apple apple filler words", createdAt = 2)
+                val worst = note("apple among lots of other unrelated filler words here", createdAt = 3)
+                store.insert(best)
+                store.insert(mid)
+                store.insert(worst)
+
+                // limit=2 after scoring keeps the two MOST RELEVANT, not the
+                // two newest.
+                val top2 = store.query<Event>(Filter(search = "apple", limit = 2)).map { it.id }
+                assertEquals(listOf(best.id, mid.id), top2)
+            } finally {
+                store.close()
+            }
+        }
+
+    /** A `#t` tag makes this a `search + tag` filter — the combined path. */
+    private var idSeq = 0
+
+    private fun hexId(n: Int): String {
+        val s = n.toString(16)
+        return "0".repeat(64 - s.length) + s
+    }
+
+    private fun tagged(
+        content: String,
+        createdAt: Long,
+        topic: String,
+    ): Event =
+        EventFactory.create(
+            hexId(++idSeq),
+            "00".repeat(32),
+            createdAt,
+            1,
+            arrayOf(arrayOf("t", topic)),
+            content,
+            "0".repeat(128),
+        )
+
+    private fun evk(
+        content: String,
+        createdAt: Long,
+        kind: Int,
+    ): Event = EventFactory.create(hexId(++idSeq), "00".repeat(32), createdAt, kind, arrayOf(), content, "0".repeat(128))
+
+    @Test
+    fun multiFilterSearchIsRelevanceOrderedAcrossBranches() =
+        runBlocking {
+            val store = EventStore(dbName = null)
+            try {
+                // Relevance A > B > C, created_at A < B < C (reverse), and the
+                // branches split by kind: A,C are kind 1 (TextNote), B is kind
+                // 1111 (Comment) — both searchable kinds.
+                val a = evk("apple apple apple apple", 1, 1)
+                val b = evk("apple apple apple", 2, 1111)
+                val c = evk("apple filler filler filler filler filler", 3, 1)
+                store.batchInsert(listOf(a, b, c))
+
+                // The client's search-across-kinds shape: one term, two filters.
+                val filters =
+                    listOf(
+                        Filter(search = "apple", kinds = listOf(1), limit = 100),
+                        Filter(search = "apple", kinds = listOf(1111), limit = 100),
+                    )
+                val ids = store.query<Event>(filters).map { it.id }
+                assertEquals(listOf(a.id, b.id, c.id), ids, "multi-filter search must be relevance-ordered across branches")
+            } finally {
+                store.close()
+            }
+        }
+
+    @Test
+    fun multiFilterSearchDedupsEventsMatchingSeveralBranches() =
+        runBlocking {
+            val store = EventStore(dbName = null)
+            try {
+                val x = evk("banana banana banana", 1, 1)
+                val y = evk("banana one", 2, 1)
+                store.batchInsert(listOf(x, y))
+
+                // Overlapping branches: both kind-1 events match BOTH filters.
+                // GROUP BY row_id must fold each to a single ranked row.
+                val filters =
+                    listOf(
+                        Filter(search = "banana", kinds = listOf(1, 6), limit = 100),
+                        Filter(search = "banana", kinds = listOf(1, 2), limit = 100),
+                    )
+                val ids = store.query<Event>(filters).map { it.id }
+                assertEquals(ids.size, ids.toSet().size, "no event may appear twice across branches")
+                assertEquals(listOf(x.id, y.id), ids, "deduped, relevance-ordered")
+                assertEquals(2, store.count(filters), "count parity across the union")
+            } finally {
+                store.close()
+            }
+        }
+
+    @Test
+    fun searchWithATagIsAlsoRelevanceOrdered() =
+        runBlocking {
+            val store = EventStore(dbName = null)
+            try {
+                // All tagged #t=nostr; relevance decreases as created_at rises,
+                // so created_at order would be the exact reverse of relevance.
+                val strong = tagged("nostr nostr nostr", createdAt = 1, topic = "nostr")
+                val weak = tagged("nostr among many other unrelated filler words here padding", createdAt = 2, topic = "nostr")
+                // Matches the term but wrong tag → excluded by the tag filter.
+                val wrongTag = tagged("nostr nostr nostr nostr", createdAt = 3, topic = "other")
+                // Right tag but doesn't match the term → excluded by search.
+                val noMatch = tagged("bitcoin only", createdAt = 4, topic = "nostr")
+
+                store.batchInsert(listOf(strong, weak, wrongTag, noMatch))
+
+                val filter = Filter(search = "nostr", tags = mapOf("t" to listOf("nostr")), limit = 10)
+                val ids = store.query<Event>(filter).map { it.id }
+                assertEquals(listOf(strong.id, weak.id), ids, "search + tag must be relevance-ordered, tag-scoped")
+
+                // limit after score keeps the most relevant one.
+                val top1 = store.query<Event>(filter.copy(limit = 1)).map { it.id }
+                assertEquals(listOf(strong.id), top1)
+            } finally {
+                store.close()
+            }
+        }
+}

quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/StatementCachingConnectionTest.kt

@@ -0,0 +1,113 @@
+/*
+ * 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.store.sqlite
+
+import androidx.sqlite.SQLiteStatement
+import androidx.sqlite.driver.bundled.BundledSQLiteDriver
+import kotlin.test.AfterTest
+import kotlin.test.BeforeTest
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertNotSame
+import kotlin.test.assertSame
+
+class StatementCachingConnectionTest {
+    private lateinit var conn: StatementCachingConnection
+
+    private val sql = "SELECT ? AS v"
+
+    @BeforeTest
+    fun setup() {
+        conn = StatementCachingConnection(BundledSQLiteDriver().open(":memory:"))
+    }
+
+    @AfterTest
+    fun tearDown() {
+        conn.close()
+    }
+
+    private fun readOne(stmt: SQLiteStatement): Long {
+        stmt.step()
+        return stmt.getLong(0)
+    }
+
+    @Test
+    fun sequentialSameSqlReusesTheSameHandle() {
+        var first: SQLiteStatement? = null
+        conn.prepare(sql).use { stmt ->
+            stmt.bindLong(1, 7)
+            assertEquals(7, readOne(stmt))
+            first = stmt
+        }
+        // Closed (returned to pool) — the next prepare of the same SQL must
+        // hand back the very same cached handle, not a fresh prepare.
+        conn.prepare(sql).use { stmt ->
+            assertSame(first, stmt)
+            stmt.bindLong(1, 9)
+            assertEquals(9, readOne(stmt))
+        }
+    }
+
+    @Test
+    fun concurrentSameSqlHandlesAreDistinctAndIndependent() {
+        // The k-way-merge shape: many identical-SQL cursors live at once.
+        val a = conn.prepare(sql)
+        val b = conn.prepare(sql)
+        val c = conn.prepare(sql)
+        assertNotSame(a, b)
+        assertNotSame(b, c)
+        assertNotSame(a, c)
+
+        a.bindLong(1, 1)
+        b.bindLong(1, 2)
+        c.bindLong(1, 3)
+        // Each cursor keeps its own bindings/position even while the others
+        // are open — no aliasing of one native handle.
+        assertEquals(1, readOne(a))
+        assertEquals(2, readOne(b))
+        assertEquals(3, readOne(c))
+        a.close()
+        b.close()
+        c.close()
+
+        // After release, a fresh concurrent burst reuses the pooled handles.
+        val reused = conn.prepare(sql)
+        assertSame(a, reused, "pool should hand back a freed handle before preparing anew")
+        reused.close()
+    }
+
+    @Test
+    fun overflowingTheGlobalCapFallsBackToUncached() {
+        val small = StatementCachingConnection(BundledSQLiteDriver().open(":memory:"), maxCachedStatements = 2)
+        try {
+            val live = (0 until 5).map { small.prepare(sql) }
+            // All five must be usable even though only two can be cached; the
+            // extra three are plain uncached statements.
+            live.forEachIndexed { i, stmt ->
+                stmt.bindLong(1, i.toLong())
+                assertEquals(i.toLong(), readOne(stmt))
+            }
+            live.forEach { it.close() }
+        } finally {
+            small.close()
+        }
+    }
+}

quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/TagMergeCorrectnessTest.kt

@@ -0,0 +1,295 @@
+/*
+ * 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.store.sqlite
+
+import com.vitorpamplona.quartz.nip01Core.core.Event
+import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
+import com.vitorpamplona.quartz.utils.EventFactory
+import kotlinx.coroutines.runBlocking
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertFalse
+import kotlin.test.assertTrue
+
+/**
+ * Correctness guard for the tag-stream path of [MergeQueryExecutor]: the
+ * `#<x>=[values] (+ kinds) [+ since/until] limit=N` watcher shape
+ * (reactions/replies). The merge must return the same newest-N a
+ * `SELECT DISTINCT … ORDER BY created_at DESC LIMIT N` would, deduping events
+ * that carry several of the queried tag values.
+ *
+ * Where `created_at` is distinct the order is fully determined and asserted
+ * against a Kotlin reference. Where it ties, the tag cursors can only order by
+ * `created_at` (event_tags has no id column), so the result is a valid
+ * newest-N but not id-exact — those cases assert set + size instead.
+ */
+class TagMergeCorrectnessTest {
+    private val hex = "0123456789abcdef"
+
+    private fun mix(seed: Long): Long {
+        var z = seed + -0x61c8864680b583ebL
+        z = (z xor (z ushr 30)) * -0x40a7b892e31b1a47L
+        z = (z xor (z ushr 27)) * -0x6b2fb644ecceee15L
+        return z xor (z ushr 31)
+    }
+
+    private fun hex64(
+        salt: Long,
+        index: Int,
+    ): String {
+        val out = CharArray(64)
+        for (w in 0 until 4) {
+            val v = mix(salt * 1_000_003 + index.toLong() * 4 + w)
+            for (b in 0 until 8) {
+                val byte = ((v ushr (b * 8)) and 0xFF).toInt()
+                out[(w * 8 + b) * 2] = hex[byte ushr 4]
+                out[(w * 8 + b) * 2 + 1] = hex[byte and 0xF]
+            }
+        }
+        return out.concatToString()
+    }
+
+    private val sig = "0".repeat(128)
+    private var idSeq = 0
+
+    private fun ev(
+        createdAt: Long,
+        kind: Int,
+        eTags: List<String>,
+    ): Event =
+        EventFactory.create(
+            hex64(7, idSeq++),
+            hex64(1, idSeq),
+            createdAt,
+            kind,
+            eTags.map { arrayOf("e", it) }.toTypedArray(),
+            "",
+            sig,
+        )
+
+    private val newestFirst =
+        Comparator<Event> { a, b ->
+            if (a.createdAt != b.createdAt) b.createdAt.compareTo(a.createdAt) else a.id.compareTo(b.id)
+        }
+
+    private fun reference(
+        all: List<Event>,
+        values: Set<String>,
+        kinds: Set<Int>?,
+        since: Long?,
+        until: Long?,
+        limit: Int,
+    ): List<String> =
+        all
+            .asSequence()
+            .filter { e -> e.tags.any { it.size >= 2 && it[0] == "e" && it[1] in values } }
+            .filter { kinds == null || it.kind in kinds }
+            .filter { since == null || it.createdAt >= since }
+            .filter { until == null || it.createdAt <= until }
+            .sortedWith(newestFirst)
+            .map { it.id }
+            .distinct()
+            .take(limit)
+            .toList()
+
+    private fun mergeEligible(
+        store: EventStore,
+        filter: Filter,
+    ): Boolean =
+        MergeQueryExecutor.streamCount(
+            with(store.store.queryBuilder) { filter.toFilterWithDTags() },
+            store.store.queryBuilder.indexStrategy,
+        ) > 0
+
+    private fun newStore() =
+        EventStore(
+            dbName = null,
+            indexStrategy =
+                DefaultIndexingStrategy(
+                    indexTagsByCreatedAtAlone = true,
+                    useAndIndexIdOnOrderBy = true,
+                    indexFullTextSearch = false,
+                ),
+        )
+
+    @Test
+    fun distinctCreatedAt_withKinds_matchesReference() =
+        runBlocking {
+            val store = newStore()
+            val notes = (0 until 6).map { hex64(2, it) }
+            val all = ArrayList<Event>()
+            var t = 1_700_000_000L
+            for (round in 0 until 40) {
+                // reactions (kind 7) and replies (kind 1) to a rotating note
+                all.add(ev(t++, 7, listOf(notes[round % notes.size])))
+                all.add(ev(t++, 1, listOf(notes[(round + 1) % notes.size])))
+            }
+            // Noise: kind-7 to notes NOT in the query set, and other tags.
+            for (i in 0 until 100) all.add(ev(t++, 7, listOf(hex64(9, i))))
+            store.batchInsert(all)
+
+            val queried = notes.take(3)
+            val filter = Filter(kinds = listOf(7), tags = mapOf("e" to queried), limit = 25)
+            assertTrue(mergeEligible(store, filter), "tag watcher must be merge-eligible")
+
+            val merged = store.query<Event>(filter).map { it.id }
+            assertEquals(reference(all, queried.toSet(), setOf(7), null, null, 25), merged)
+            store.close()
+        }
+
+    @Test
+    fun crossStreamDuplicate_emittedOnce() =
+        runBlocking {
+            val store = newStore()
+            val a = hex64(3, 0)
+            val b = hex64(3, 1)
+            val all = ArrayList<Event>()
+            var t = 1_700_000_000L
+            // Events tagging BOTH a and b — they appear in both value streams
+            // and must be emitted exactly once.
+            repeat(5) { all.add(ev(t++, 1, listOf(a, b))) }
+            // Events tagging only one of them.
+            repeat(5) { all.add(ev(t++, 1, listOf(a))) }
+            repeat(5) { all.add(ev(t++, 1, listOf(b))) }
+            store.batchInsert(all)
+
+            val filter = Filter(kinds = listOf(1), tags = mapOf("e" to listOf(a, b)), limit = 500)
+            assertTrue(mergeEligible(store, filter))
+
+            val merged = store.query<Event>(filter).map { it.id }
+            assertEquals(merged.size, merged.toSet().size, "no event may be emitted twice")
+            assertEquals(15, merged.size, "5 both + 5 a-only + 5 b-only = 15 distinct")
+            assertEquals(reference(all, setOf(a, b), setOf(1), null, null, 500), merged)
+            store.close()
+        }
+
+    @Test
+    fun noKinds_usesTagCreatedAtIndex() =
+        runBlocking {
+            val store = newStore()
+            val notes = (0 until 5).map { hex64(4, it) }
+            val all = ArrayList<Event>()
+            var t = 1_700_000_000L
+            for (round in 0 until 30) all.add(ev(t++, (round % 3) + 1, listOf(notes[round % notes.size])))
+            for (i in 0 until 60) all.add(ev(t++, 1, listOf(hex64(8, i))))
+            store.batchInsert(all)
+
+            val queried = notes.take(3)
+            val filter = Filter(tags = mapOf("e" to queried), limit = 20)
+            assertTrue(mergeEligible(store, filter), "no-kind tag watcher must be merge-eligible with the tag index")
+
+            val merged = store.query<Event>(filter).map { it.id }
+            assertEquals(reference(all, queried.toSet(), null, null, null, 20), merged)
+            store.close()
+        }
+
+    @Test
+    fun withSinceAndUntil_boundsTheWindow() =
+        runBlocking {
+            val store = newStore()
+            val notes = (0 until 4).map { hex64(5, it) }
+            val all = ArrayList<Event>()
+            val base = 1_700_000_000L
+            for (i in 0 until 300) all.add(ev(base + i.toLong(), 7, listOf(notes[i % notes.size])))
+            store.batchInsert(all)
+
+            val since = base + 50
+            val until = base + 250
+            val filter = Filter(kinds = listOf(7), tags = mapOf("e" to notes), since = since, until = until, limit = 500)
+            assertTrue(mergeEligible(store, filter))
+
+            val merged = store.query<Event>(filter).map { it.id }
+            val ref = reference(all, notes.toSet(), setOf(7), since, until, 500)
+            assertEquals(ref, merged)
+            assertTrue(merged.isNotEmpty() && ref.size < 300)
+            store.close()
+        }
+
+    @Test
+    fun rawPathMatchesDecoded() =
+        runBlocking {
+            val store = newStore()
+            val notes = (0 until 6).map { hex64(6, it) }
+            val all = ArrayList<Event>()
+            var t = 1_700_000_000L
+            for (round in 0 until 25) all.add(ev(t++, 7, listOf(notes[round % notes.size])))
+            store.batchInsert(all)
+
+            val filter = Filter(kinds = listOf(7), tags = mapOf("e" to notes.take(3)), limit = 15)
+            val decoded = store.query<Event>(filter).map { it.id }
+            val raw = store.store.rawQuery(filter).map { it.id }
+            assertEquals(decoded, raw, "the zero-decode raw path must match the decoded query")
+            store.close()
+        }
+
+    @Test
+    fun tiedCreatedAt_matchesReferenceAsSet() =
+        runBlocking {
+            val store = newStore()
+            val notes = (0 until 4).map { hex64(10, it) }
+            val all = ArrayList<Event>()
+            // Many events share a created_at — the tag cursor can't id-order
+            // within a second, so assert a valid newest-N by set + size.
+            var t = 1_700_000_000L
+            for (block in 0 until 15) {
+                val ts = t
+                for (n in notes) all.add(ev(ts, 7, listOf(n)))
+                t += 1
+            }
+            store.batchInsert(all)
+
+            val filter = Filter(kinds = listOf(7), tags = mapOf("e" to notes), limit = 22)
+            assertTrue(mergeEligible(store, filter))
+
+            val merged = store.query<Event>(filter).map { it.id }
+            assertEquals(22, merged.size)
+            assertEquals(merged.size, merged.toSet().size)
+            // The whole result must sit within the newest slice the reference
+            // would return once ties are resolved either way: everything in
+            // `merged` must be at or above the created_at cutoff.
+            val ids = merged.toSet()
+            val chosen = all.filter { it.id in ids }
+            val cutoff = chosen.minOf { it.createdAt }
+            val eligibleAboveCutoff = all.filter { it.createdAt > cutoff }.map { it.id }.toSet()
+            assertTrue(eligibleAboveCutoff.all { it in ids }, "every event newer than the cutoff must be included")
+            store.close()
+        }
+
+    @Test
+    fun ineligibleShapesFallThrough() =
+        runBlocking {
+            val store = newStore()
+            store.batchInsert(listOf(ev(1_700_000_000L, 7, listOf(hex64(2, 0)))))
+
+            // Single value → single seek, not a merge.
+            assertFalse(mergeEligible(store, Filter(kinds = listOf(7), tags = mapOf("e" to listOf(hex64(2, 0))), limit = 10)))
+            // No limit → not merge-eligible.
+            assertFalse(mergeEligible(store, Filter(kinds = listOf(7), tags = mapOf("e" to listOf(hex64(2, 0), hex64(2, 1))))))
+            // Authors present → covered-index seek shape, not a tag merge.
+            assertFalse(
+                mergeEligible(
+                    store,
+                    Filter(kinds = listOf(7), authors = listOf(hex64(1, 1)), tags = mapOf("e" to listOf(hex64(2, 0), hex64(2, 1))), limit = 10),
+                ),
+            )
+            store.close()
+        }
+}

quartz/src/jvmMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/fs/FsQueryPlanner.kt

@@ -25,6 +25,7 @@ import com.vitorpamplona.quartz.nip01Core.core.isAddressable
 import com.vitorpamplona.quartz.nip01Core.core.isReplaceable
 import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
 import com.vitorpamplona.quartz.nip01Core.store.sqlite.TagNameValueHasher
+import java.nio.file.DirectoryStream
 import java.nio.file.Files
 import java.nio.file.Path
 import kotlin.io.path.exists
@@ -36,16 +37,23 @@ import kotlin.io.path.exists
  *
  * Step-2 coverage:
  *  - `ids`            → direct canonical opens
- *  - `tagsAll`/`tags` → tag index union (first key)
- *  - `kinds`          → kind index union
- *  - `authors`        → author index union
+ *  - `tagsAll`/`tags`/`kinds`/`authors` → cheapest index tree drives
+ *    (capped entry-count comparison), the rest post-filter
  *  - otherwise        → full scan via every `idx/kind/<k>/` subtree
  *
- * The planner is intentionally dumb about selectivity — "first available
- * driver wins". A cost-based picker (smallest listing) can slot in
- * later without changing callers. All FilterMatcher semantics (tag
- * AND/OR, since/until, id, author, kind cross-checks) are enforced in
- * the orchestrator, so picking a loose driver is correctness-safe.
+ * Driver choice is cost-based: every legal driver (each `tagsAll` value
+ * alone — AND semantics make any single value a complete driver — each
+ * `tags` key's value union, the kind set, the author set) opens a lazy
+ * directory iterator, all are drained in lockstep, and the first to
+ * exhaust — the smallest listing — drives. A giant tree (`idx/kind/1/`
+ * with a million entries) is therefore never read past ~the smallest
+ * candidate's size. Before the pick, the fixed tags → kinds → authors
+ * order sent `authors + kinds + limit` — the most common CLI shape —
+ * through the kind tree: 149 ms fixed-order vs 4.0 ms cost-based at 30k
+ * events per `FsDriverSelectionBenchmark` (floor: author-only at
+ * 1.4 ms). All FilterMatcher semantics (tag AND/OR,
+ * since/until, id, author, kind cross-checks) are enforced in the
+ * orchestrator, so any driver pick is correctness-safe.
  */
 internal class FsQueryPlanner(
     private val layout: FsLayout,
@@ -74,19 +82,100 @@ internal class FsQueryPlanner(
             return ftsDriver(search)
         }
 
-        firstTagKey(filter)?.let { (name, values) ->
-            return mergeDesc(values.map { v -> walkDir(layout.tagValueDir(name, v, hasher.hash(name, v))) })
+        val candidates = driverCandidates(filter)
+        if (candidates.isEmpty()) return allKindsDriver()
+
+        return mergeDesc(cheapestDriver(candidates).map { walkDir(it) })
+    }
+
+    /**
+     * Every set of index directories that, walked and post-filtered, yields
+     * a superset of the filter's matches:
+     *  - each `tagsAll` value alone (AND semantics — every match carries it),
+     *  - each `tags` key's full value union (OR within a key, AND across),
+     *  - the kind set, and the author set.
+     * Listed in the old fixed-priority order so [cheapestDriver] keeps that
+     * order on cost ties.
+     */
+    private fun driverCandidates(filter: Filter): List<List<Path>> {
+        val out = ArrayList<List<Path>>()
+        filter.tagsAll?.forEach { (name, values) ->
+            values.forEach { v -> out.add(listOf(layout.tagValueDir(name, v, hasher.hash(name, v)))) }
+        }
+        filter.tags?.forEach { (name, values) ->
+            if (values.isNotEmpty()) {
+                out.add(values.map { v -> layout.tagValueDir(name, v, hasher.hash(name, v)) })
+            }
+        }
+        filter.kinds?.takeIf { it.isNotEmpty() }?.let { kinds ->
+            out.add(kinds.map { layout.kindDir(it) })
+        }
+        filter.authors?.takeIf { it.isNotEmpty() }?.let { authors ->
+            out.add(authors.map { layout.authorDir(it) })
+        }
+        return out
+    }
+
+    /**
+     * Smallest candidate by lockstep listing drain: one lazy directory
+     * iterator per candidate, all advanced [COST_BATCH] entries per round —
+     * the first to exhaust its listing is the smallest, so a giant tree is
+     * never read past ~the smallest candidate's size (a candidate that
+     * exhausts on round one costs the others one batch each). A candidate
+     * whose dirs are all missing exhausts immediately: driving from an empty
+     * mandatory predicate correctly yields an empty result. If every
+     * candidate survives [COST_CAP] entries, all are huge and relative
+     * driver choice stops mattering — the first (old fixed-priority order)
+     * wins.
+     */
+    private fun cheapestDriver(candidates: List<List<Path>>): List<Path> {
+        if (candidates.size == 1) return candidates[0]
+        val cursors = candidates.map { EntryCursor(it) }
+        try {
+            var advanced = 0L
+            while (advanced < COST_CAP) {
+                for (i in cursors.indices) {
+                    if (!cursors[i].skip(COST_BATCH)) return candidates[i]
+                }
+                advanced += COST_BATCH
+            }
+            return candidates[0]
+        } finally {
+            cursors.forEach { it.close() }
+        }
+    }
+
+    /** Lazy entry iterator over a candidate's directories, in order. */
+    private class EntryCursor(
+        dirs: List<Path>,
+    ) : AutoCloseable {
+        private val remaining = ArrayDeque(dirs)
+        private var stream: DirectoryStream<Path>? = null
+        private var iter: Iterator<Path> = emptyList<Path>().iterator()
+
+        /** Advances up to [n] entries; false when the listing ends first. */
+        fun skip(n: Int): Boolean {
+            var left = n
+            while (left > 0) {
+                if (iter.hasNext()) {
+                    iter.next()
+                    left--
+                    continue
+                }
+                close()
+                val dir = remaining.removeFirstOrNull() ?: return false
+                if (!Files.isDirectory(dir)) continue
+                stream = Files.newDirectoryStream(dir)
+                iter = stream!!.iterator()
+            }
+            return true
         }
 
-        filter.kinds?.let { kinds ->
-            return mergeDesc(kinds.map { walkDir(layout.kindDir(it)) })
+        override fun close() {
+            stream?.close()
+            stream = null
+            iter = emptyList<Path>().iterator()
         }
-
-        filter.authors?.let { authors ->
-            return mergeDesc(authors.map { walkDir(layout.authorDir(it)) })
-        }
-
-        return allKindsDriver()
     }
 
     /**
@@ -301,14 +390,15 @@ internal class FsQueryPlanner(
         var top: Candidate,
     )
 
-    // ---- helpers ------------------------------------------------------
+    private companion object {
+        /** Entries each candidate's cursor advances per lockstep round. */
+        const val COST_BATCH = 64
 
-    /** First tag filter with at least one value, preferring `tagsAll`. */
-    private fun firstTagKey(filter: Filter): Pair<String, List<String>>? {
-        filter.tagsAll?.firstNonEmpty()?.let { return it }
-        filter.tags?.firstNonEmpty()?.let { return it }
-        return null
+        /**
+         * Stop draining once every candidate has survived this many
+         * entries: past it they are all huge, relative choice stops
+         * mattering, and the first candidate in priority order wins.
+         */
+        const val COST_CAP = 65_536L
     }
-
-    private fun Map<String, List<String>>.firstNonEmpty(): Pair<String, List<String>>? = entries.firstOrNull { it.value.isNotEmpty() }?.let { it.key to it.value }
 }

quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/FollowFeedReadBenchmark.kt

@@ -202,7 +202,7 @@ class FollowFeedReadBenchmark {
             runBlocking {
                 store.store.pool.useReader { c ->
                     var n = 0
-                    MergeQueryExecutor.run(c, filter, store.store.queryBuilder.indexStrategy) { n++ }
+                    MergeQueryExecutor.run(c, filter, store.store.queryBuilder.indexStrategy, store.store.seedModule::hasher) { n++ }
                     n
                 }
             }

quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/FtsSearchScalingBenchmark.kt

@@ -0,0 +1,185 @@
+/*
+ * 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.prodbench
+
+import androidx.sqlite.SQLiteConnection
+import androidx.sqlite.driver.bundled.BundledSQLiteDriver
+import com.vitorpamplona.quartz.nip01Core.core.Event
+import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
+import com.vitorpamplona.quartz.nip01Core.store.sqlite.DefaultIndexingStrategy
+import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore
+import com.vitorpamplona.quartz.utils.EventFactory
+import kotlinx.coroutines.runBlocking
+import kotlin.test.Test
+
+/**
+ * Two measurements around the contentless FTS index, motivating the v4→v5
+ * schema change and being honest about what it does and does not fix.
+ *
+ *  1. **Delete scaling — the win.** The old `fts5(event_header_row_id,
+ *     content)` schema had the `fts_foreign_key` trigger delete by a regular
+ *     FTS column, which FTS5 cannot seek (it scans, O(n) per delete). The
+ *     contentless schema keys deletes off the rowid (= event_headers.row_id),
+ *     an O(log n) primary-key seek. Every event removal fires this trigger
+ *     (replaceable rotation, kind-5, expiration, right-to-vanish).
+ *  2. **Search scaling — the limit.** `MATCH … ORDER BY rank LIMIT n` (NIP-50
+ *     relevance ordering, bm25) must score *every* matching document, so
+ *     search cost grows with the match set regardless of ordering (created_at
+ *     has the same shape). Segment `optimize` compacts the index but does not
+ *     change that; corpus-independent search needs an external engine. Shown
+ *     fragmented vs optimized to size the (secondary) compaction effect.
+ *
+ * Size search with `-DftsBenchScale=N` (default 1). Not an assertion test.
+ */
+class FtsSearchScalingBenchmark {
+    companion object {
+        val SCALE = System.getProperty("ftsBenchScale")?.toInt() ?: 1
+        val SIZES = listOf(50_000, 100_000, 200_000).map { it * SCALE }
+        const val NEEDLE = "zzneedle"
+    }
+
+    private val hex = "0123456789abcdef"
+
+    private fun mix(seed: Long): Long {
+        var z = seed + -0x61c8864680b583ebL
+        z = (z xor (z ushr 30)) * -0x40a7b892e31b1a47L
+        z = (z xor (z ushr 27)) * -0x6b2fb644ecceee15L
+        return z xor (z ushr 31)
+    }
+
+    private fun hex64(index: Int): String {
+        val out = CharArray(64)
+        for (w in 0 until 4) {
+            val v = mix(index.toLong() * 4 + w + 99)
+            for (b in 0 until 8) {
+                val byte = ((v ushr (b * 8)) and 0xFF).toInt()
+                out[(w * 8 + b) * 2] = hex[byte ushr 4]
+                out[(w * 8 + b) * 2 + 1] = hex[byte and 0xF]
+            }
+        }
+        return String(out)
+    }
+
+    private val vocab = (0 until 400).map { "word$it" }
+    private val sig = "0".repeat(128)
+
+    private fun seed(n: Int): List<Event> {
+        val base = 1_700_000_000L
+        val events = ArrayList<Event>(n)
+        for (i in 0 until n) {
+            val r = mix(i.toLong())
+            val content =
+                buildString {
+                    for (w in 0 until 8) append(vocab[((r ushr (w * 3)) and 0x1FF).toInt() % vocab.size]).append(' ')
+                    // ~1% carry the searched term.
+                    if (i % 100 == 0) append(NEEDLE)
+                }
+            events.add(EventFactory.create(hex64(i), hex64(i % 5000), base + i.toLong(), 1, emptyArray(), content, sig))
+        }
+        return events
+    }
+
+    private inline fun timeMs(block: () -> Unit): Double {
+        val start = System.nanoTime()
+        block()
+        return (System.nanoTime() - start) / 1e6
+    }
+
+    private fun SQLiteConnection.exec(sql: String) = prepare(sql).use { it.step() }
+
+    @Test
+    fun deleteByColumnVsByRowid() {
+        // Old-schema (delete by FTS column) vs contentless (delete by rowid),
+        // 500 deletes at two table sizes. By-column should grow with the
+        // table; by-rowid should stay flat.
+        println("─ FtsSearchScalingBenchmark.delete (500 deletes) ─")
+        println("  %-9s %14s %14s".format("rows", "byColumn", "byRowid"))
+        for (n in listOf(2_000 * SCALE, 8_000 * SCALE)) {
+            val db = BundledSQLiteDriver().open(":memory:")
+            try {
+                db.exec("CREATE VIRTUAL TABLE col USING fts5(event_header_row_id, content)")
+                db.exec("CREATE VIRTUAL TABLE row USING fts5(content, content='', contentless_delete=1)")
+                for (i in 1..n) {
+                    db.prepare("INSERT INTO col(event_header_row_id, content) VALUES (?, 'alpha beta gamma')").use {
+                        it.bindLong(1, i.toLong())
+                        it.step()
+                    }
+                    db.prepare("INSERT INTO row(rowid, content) VALUES (?, 'alpha beta gamma')").use {
+                        it.bindLong(1, i.toLong())
+                        it.step()
+                    }
+                }
+                val byColumn =
+                    timeMs {
+                        for (i in 1..500) {
+                            db.prepare("DELETE FROM col WHERE event_header_row_id = ?").use {
+                                it.bindLong(1, i.toLong())
+                                it.step()
+                            }
+                        }
+                    }
+                val byRowid =
+                    timeMs {
+                        for (i in 1..500) {
+                            db.prepare("DELETE FROM row WHERE rowid = ?").use {
+                                it.bindLong(1, i.toLong())
+                                it.step()
+                            }
+                        }
+                    }
+                println("  %-9s %11.2f ms %11.2f ms".format("${n / 1000}k", byColumn, byRowid))
+            } finally {
+                db.close()
+            }
+        }
+    }
+
+    @Test
+    fun searchScaling() =
+        runBlocking {
+            println("─ FtsSearchScalingBenchmark.search (created_at DESC, limit=50) ─")
+            println("  %-9s %16s %16s".format("corpus", "fragmented", "optimized"))
+            for (size in SIZES) {
+                val store = EventStore(dbName = null, indexStrategy = DefaultIndexingStrategy())
+                try {
+                    seed(size).chunked(10_000).forEach { store.batchInsert(it) }
+                    val f = Filter(search = NEEDLE, limit = 50)
+                    val frag = time(store, f)
+                    store.store.reindexFullTextSearch() // rebuild + optimize()
+                    val opt = time(store, f)
+                    println("  %-9s %13.2f ms %13.2f ms".format("${size / 1000}k", frag, opt))
+                } finally {
+                    store.close()
+                }
+            }
+        }
+
+    private suspend fun time(
+        store: EventStore,
+        filter: Filter,
+    ): Double {
+        repeat(3) { store.query<Event>(filter) }
+        val runs = 20
+        val start = System.nanoTime()
+        repeat(runs) { store.query<Event>(filter) }
+        return (System.nanoTime() - start) / 1e6 / runs
+    }
+}

quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/SmallReqFloorBenchmark.kt

@@ -32,11 +32,14 @@ import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore
 import com.vitorpamplona.quartz.utils.EventFactory
 import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.CoroutineStart
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.cancel
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeout
+import java.util.concurrent.atomic.AtomicInteger
 import kotlin.test.Test
 import kotlin.test.assertEquals
 
@@ -65,6 +68,8 @@ class SmallReqFloorBenchmark {
         const val AUTHORS = 2_500 // ~20 events per author, matching author-archive
         const val ROUNDS = 400
         const val WARMUP = 100
+        const val IDLE_SUBS = 1_000
+        const val FANOUT_SUBS = 200
     }
 
     private fun hexId(seed: Int): String = seed.toString(16).padStart(64, '0')
@@ -122,16 +127,19 @@ class SmallReqFloorBenchmark {
             }
 
             // --- B: backend queryRaw to EOSE (live machinery included) ---
+            // UNDISPATCHED mirrors the production path (RelaySession.handleReq
+            // starts the query coroutine undispatched), so B−A is the live
+            // machinery itself, not a benchmark-only scheduler hop.
             suspend fun timeBackend(round: Int): Long {
                 val eose = CompletableDeferred<Long>()
                 val t0 = System.nanoTime()
                 val job =
-                    scope.launch {
+                    scope.launch(start = CoroutineStart.UNDISPATCHED) {
                         live.queryRaw(
                             ctx = ctx,
                             filters = listOf(filterFor(round)),
                             onEachStored = {},
-                            onEachLive = {},
+                            onEachLive = { _, _ -> },
                             onEose = { eose.complete(System.nanoTime() - t0) },
                         )
                     }
@@ -143,6 +151,33 @@ class SmallReqFloorBenchmark {
             val b = LongArray(ROUNDS)
             repeat(ROUNDS) { b[it] = timeBackend(it) }
 
+            // --- B@1k: same, with 1000 idle live subscriptions parked ---
+            // Register/unregister cost scales with the live population
+            // (FilterIndex mutates a shared snapshot per REQ open/close),
+            // which the single-sub stage can't see. Each idle sub filters
+            // on an author absent from the corpus: 0-row replay, then parks
+            // at the live tail and stays registered.
+            val idleJobs =
+                (0 until IDLE_SUBS).map { i ->
+                    val ready = CompletableDeferred<Unit>()
+                    val job =
+                        scope.launch(start = CoroutineStart.UNDISPATCHED) {
+                            live.queryRaw(
+                                ctx = ctx,
+                                filters = listOf(Filter(authors = listOf(hexId(1_000_000 + i)), kinds = listOf(1), limit = 1)),
+                                onEachStored = {},
+                                onEachLive = { _, _ -> },
+                                onEose = { ready.complete(Unit) },
+                            )
+                        }
+                    ready.await()
+                    job
+                }
+            repeat(WARMUP) { timeBackend(it) }
+            val b1k = LongArray(ROUNDS)
+            repeat(ROUNDS) { b1k[it] = timeBackend(it) }
+            idleJobs.forEach { it.cancel() }
+
             // --- C: full session dispatch, REQ json in → EOSE frame out ---
             suspend fun timeSession(round: Int): Long {
                 val eose = CompletableDeferred<Long>()
@@ -160,15 +195,67 @@ class SmallReqFloorBenchmark {
             val c = LongArray(ROUNDS)
             repeat(ROUNDS) { c[it] = timeSession(it) }
 
+            // --- fanout: one live event → FANOUT_SUBS live subscriptions ---
+            // All subs register on `live` directly (via queryRaw, same backend
+            // we submit into) and filter an author with no stored events (0-row
+            // replay, then park live). Submitting one matching event fans out to
+            // every sub; the body is serialized once and spliced per sub, so
+            // this measures the shared-serialization path (#2). skipVerify so
+            // the synthetic sig is accepted. Fewer rounds than A–C: each round
+            // is FANOUT_SUBS deliveries and a real group-commit insert.
+            val fanAuthor = hexId(9_000_001)
+            val delivered = AtomicInteger(0)
+            var fanDone = CompletableDeferred<Long>()
+            var fanStart = 0L
+            val fanJobs =
+                (0 until FANOUT_SUBS).map {
+                    val ready = CompletableDeferred<Unit>()
+                    val job =
+                        scope.launch(start = CoroutineStart.UNDISPATCHED) {
+                            live.queryRaw(
+                                ctx = ctx,
+                                filters = listOf(Filter(authors = listOf(fanAuthor), kinds = listOf(1))),
+                                onEachStored = {},
+                                onEachLive = { _, _ ->
+                                    if (delivered.incrementAndGet() == FANOUT_SUBS) {
+                                        fanDone.complete(System.nanoTime() - fanStart)
+                                    }
+                                },
+                                onEose = { ready.complete(Unit) },
+                            )
+                        }
+                    ready.await()
+                    job
+                }
+            val fanRounds = 60
+            val fanWarmup = 15
+            val fan = LongArray(fanRounds)
+            var fanSeq = 0
+            repeat(fanWarmup + fanRounds) { r ->
+                delivered.set(0)
+                fanDone = CompletableDeferred()
+                val ev = EventFactory.create<Event>(hexId(9_500_000 + fanSeq), fanAuthor, 1_700_000_000L + fanSeq, 1, emptyArray(), "fanout $fanSeq", sig)
+                fanSeq++
+                fanStart = System.nanoTime()
+                live.submit(ev, skipVerify = true) {}
+                val nanos = withTimeout(30_000) { fanDone.await() }
+                if (r >= fanWarmup) fan[r - fanWarmup] = nanos
+            }
+            fanJobs.forEach { it.cancel() }
+
             assertEquals(true, rowsA > 0, "author filters must return rows")
 
             val mA = median(a)
             val mB = median(b)
+            val mB1k = median(b1k)
             val mC = median(c)
             println("SmallReqFloorBenchmark @ ${EVENTS / 1000}k events, ~${rowsA / ROUNDS} rows/req, medians of $ROUNDS")
             println("  A raw store query:        ${"%6.3f".format(mA)} ms")
             println("  B backend queryRaw→EOSE:  ${"%6.3f".format(mB)} ms  (live machinery +${"%6.3f".format(mB - mA)})")
+            println("  B@${IDLE_SUBS} idle subs:      ${"%6.3f".format(mB1k)} ms  (population cost +${"%6.3f".format(mB1k - mB)})")
             println("  C session REQ→EOSE:       ${"%6.3f".format(mC)} ms  (dispatch+frames +${"%6.3f".format(mC - mB)})")
+            val mFan = median(fan)
+            println("  fanout 1→$FANOUT_SUBS live subs: ${"%6.3f".format(mFan)} ms  (${"%.2f".format(mFan * 1000 / FANOUT_SUBS)} µs/sub; body serialized once)")
 
             server.close()
             scope.cancel()

quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/TagAuthorIndexBenchmark.kt

@@ -0,0 +1,184 @@
+/*
+ * 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.prodbench
+
+import com.vitorpamplona.quartz.nip01Core.core.Event
+import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
+import com.vitorpamplona.quartz.nip01Core.store.sqlite.DefaultIndexingStrategy
+import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore
+import com.vitorpamplona.quartz.utils.EventFactory
+import kotlinx.coroutines.runBlocking
+import kotlin.test.Test
+
+/**
+ * Measures the two tag-path query shapes the client filter-assembler survey
+ * (2026-07) found hot but that no existing benchmark covers:
+ *
+ * 1. **tag ∩ author (DM-room shape)** — `kinds=[4] AND authors=[peer] AND
+ *    #p=[me] LIMIT n`. 65 assembler call sites build this shape (every
+ *    NIP-04 chat room, reports-by-follows, follows-scoped community feeds).
+ *    [com.vitorpamplona.quartz.nip01Core.store.sqlite.IndexingStrategy.indexTagsWithKindAndPubkey]
+ *    gates a covering `(tag_hash, kind, pubkey_hash, created_at)` index for
+ *    it, but the flag is off everywhere (including geode). Without it the
+ *    plan seeks `(tag_hash, kind)` and reads EVERY DM the user has ever
+ *    received before filtering to the one peer. This compares query latency
+ *    with the flag off vs on, and the batch-insert cost the extra index adds.
+ *
+ * 2. **large-IN tag watcher (reactions shape)** — `kinds=[7] AND
+ *    #e=[hundreds of note ids] LIMIT n`. The per-value streams come sorted
+ *    off `(tag_hash, kind, created_at)`, but their union does not, so SQLite
+ *    collects every matching row and TEMP-B-TREE sorts to the limit — the
+ *    tag-index analogue of the follow-feed regression
+ *    [MergeQueryExecutor] fixed for author streams. Reported with and
+ *    without a `since` bound to show what EOSE-warm steady state hides.
+ *
+ * Size the seed with `-DtagBenchScale=N` (default 1 ≈ ~200k events).
+ */
+class TagAuthorIndexBenchmark {
+    companion object {
+        val SCALE = System.getProperty("tagBenchScale")?.toInt() ?: 1
+    }
+
+    private val hex = "0123456789abcdef"
+
+    private fun mix(seed: Long): Long {
+        var z = seed + -0x61c8864680b583ebL
+        z = (z xor (z ushr 30)) * -0x40a7b892e31b1a47L
+        z = (z xor (z ushr 27)) * -0x6b2fb644ecceee15L
+        return z xor (z ushr 31)
+    }
+
+    private fun hex64(
+        salt: Long,
+        index: Int,
+    ): String {
+        val out = CharArray(64)
+        for (w in 0 until 4) {
+            val v = mix(salt * 1_000_003 + index.toLong() * 4 + w)
+            for (b in 0 until 8) {
+                val byte = ((v ushr (b * 8)) and 0xFF).toInt()
+                out[(w * 8 + b) * 2] = hex[byte ushr 4]
+                out[(w * 8 + b) * 2 + 1] = hex[byte and 0xF]
+            }
+        }
+        return String(out)
+    }
+
+    private val sig = "0".repeat(128)
+    private var idSeq = 0
+
+    private fun ev(
+        pubkey: String,
+        createdAt: Long,
+        kind: Int,
+        tags: Array<Array<String>>,
+    ): Event = EventFactory.create(hex64(7, idSeq++), pubkey, createdAt, kind, tags, "", sig)
+
+    private fun seedEvents(): List<Event> {
+        idSeq = 0
+        val base = 1_700_000_000L
+        val span = 3_000_000L // ~35 days
+        val me = hex64(9, 0)
+        val events = ArrayList<Event>(220_000 * SCALE)
+
+        // DM inbox: 200 peers, 300 DMs each → 60k kind-4 rows sharing the
+        // same (p:me) tag hash. The room query wants one peer's 300.
+        val peers = (0 until 200).map { hex64(2, it) }
+        for ((i, peer) in peers.withIndex()) {
+            repeat(300 * SCALE) {
+                val ts = base + (mix(i * 131L + it) and 0x7fffffff) % span
+                events.add(ev(peer, ts, 4, arrayOf(arrayOf("p", me))))
+            }
+        }
+
+        // Notification noise: 2000 authors mention me in kind-1 notes, so
+        // (p:me) spans multiple kinds like a real inbox does.
+        repeat(40_000 * SCALE) {
+            val author = hex64(3, it % 2_000)
+            val ts = base + (mix(it * 17L) and 0x7fffffff) % span
+            events.add(ev(author, ts, 1, arrayOf(arrayOf("p", me))))
+        }
+
+        // Reactions: 100k kind-7 events spread over 5000 target notes, for
+        // the large-IN watcher shape.
+        val noteIds = (0 until 5_000).map { hex64(5, it) }
+        repeat(100_000 * SCALE) {
+            val author = hex64(4, it % 3_000)
+            val ts = base + (mix(it * 29L) and 0x7fffffff) % span
+            events.add(ev(author, ts, 7, arrayOf(arrayOf("e", noteIds[it % noteIds.size]))))
+        }
+        return events
+    }
+
+    @Test
+    fun compareTagAuthorIndex() =
+        runBlocking {
+            val events = seedEvents()
+            val me = hex64(9, 0)
+            val peers = (0 until 200).map { hex64(2, it) }
+            val noteIds = (0 until 5_000).map { hex64(5, it) }
+
+            println("─ TagAuthorIndexBenchmark: ${events.size} events (scale=$SCALE) ─")
+
+            val strategies =
+                listOf(
+                    "flag-off" to DefaultIndexingStrategy(indexFullTextSearch = false),
+                    "flag-on " to DefaultIndexingStrategy(indexFullTextSearch = false, indexTagsWithKindAndPubkey = true),
+                )
+
+            for ((label, strategy) in strategies) {
+                val store = EventStore(dbName = null, indexStrategy = strategy)
+
+                val t0 = System.nanoTime()
+                events.chunked(10_000).forEach { store.batchInsert(it) }
+                val insertMs = (System.nanoTime() - t0) / 1e6
+                println("  ═ $label ═  insert: %.0f ms (%.1f µs/event)".format(insertMs, insertMs * 1000 / events.size))
+
+                // 1. DM room: one peer's DMs out of the whole (p:me) inbox.
+                val room = Filter(kinds = listOf(4), authors = listOf(peers[42]), tags = mapOf("p" to listOf(me)), limit = 100)
+                time(store, "dm-room (#p ∩ author ∩ kind, limit 100)", room)
+
+                // 2. Reactions watcher: 300 note ids, cold (no since).
+                val watcher = Filter(kinds = listOf(7), tags = mapOf("e" to noteIds.take(300)), limit = 500)
+                time(store, "reactions (#e IN 300, limit 500, cold)", watcher)
+
+                // 3. Same watcher, EOSE-warm (since bounds the window).
+                val warm = watcher.copy(since = 1_700_000_000L + 2_900_000L)
+                time(store, "reactions (#e IN 300, limit 500, since)", warm)
+
+                store.close()
+            }
+        }
+
+    private suspend fun time(
+        store: EventStore,
+        label: String,
+        filter: Filter,
+    ) {
+        repeat(3) { store.query<Event>(filter) }
+        val runs = 10
+        var rows = 0
+        val start = System.nanoTime()
+        repeat(runs) { rows = store.query<Event>(filter).size }
+        val ms = (System.nanoTime() - start) / 1e6 / runs
+        println("    %-42s %8.2f ms  (%d rows)".format(label, ms, rows))
+    }
+}

quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/fs/FsDriverSelectionBenchmark.kt

@@ -0,0 +1,159 @@
+/*
+ * 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.store.fs
+
+import com.vitorpamplona.quartz.nip01Core.core.Event
+import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
+import com.vitorpamplona.quartz.utils.EventFactory
+import kotlinx.coroutines.runBlocking
+import java.nio.file.Files
+import java.nio.file.Path
+import kotlin.io.path.exists
+import kotlin.test.Test
+
+/**
+ * Guards [FsQueryPlanner]'s cost-based driver pick on the
+ * `authors + kinds + limit` shape — the most common CLI query (27
+ * assembler call sites; every `amy feed`-style author timeline over
+ * non-replaceable kinds).
+ *
+ * Under the pre-pick fixed order (tags → kinds → authors),
+ * `Filter(authors=[pk], kinds=[1], limit=n)` drove from `idx/kind/1/`
+ * (the biggest tree in any real store) and post-filtered the author:
+ * 149 ms at 30k events. The lockstep pick drives from the author tree
+ * and runs at ~4 ms. The benchmark times:
+ *
+ *  - **planner (cost-based pick)**: the filter as the planner runs it —
+ *    should sit near the floor, far below a kind-tree walk.
+ *  - **author-driver emulation**: the author tree walked via an
+ *    authors-only query with the kind check applied by the caller — the
+ *    reference the pick is expected to match or beat.
+ *  - **author-only floor**: `authors + limit` with no kind, the cheapest
+ *    possible walk of the same tree.
+ *
+ * Size the seed with `-DfsBenchScale=N` (default 1 ≈ ~30k events; each
+ * event is a file + ~3 hardlinks, so seeding dominates wall time).
+ */
+class FsDriverSelectionBenchmark {
+    companion object {
+        val SCALE = System.getProperty("fsBenchScale")?.toInt() ?: 1
+    }
+
+    private val hex = "0123456789abcdef"
+
+    private fun mix(seed: Long): Long {
+        var z = seed + -0x61c8864680b583ebL
+        z = (z xor (z ushr 30)) * -0x40a7b892e31b1a47L
+        z = (z xor (z ushr 27)) * -0x6b2fb644ecceee15L
+        return z xor (z ushr 31)
+    }
+
+    private fun hex64(
+        salt: Long,
+        index: Int,
+    ): String {
+        val out = CharArray(64)
+        for (w in 0 until 4) {
+            val v = mix(salt * 1_000_003 + index.toLong() * 4 + w)
+            for (b in 0 until 8) {
+                val byte = ((v ushr (b * 8)) and 0xFF).toInt()
+                out[(w * 8 + b) * 2] = hex[byte ushr 4]
+                out[(w * 8 + b) * 2 + 1] = hex[byte and 0xF]
+            }
+        }
+        return String(out)
+    }
+
+    private val sig = "0".repeat(128)
+    private var idSeq = 0
+
+    private fun ev(
+        pubkey: String,
+        createdAt: Long,
+        kind: Int,
+    ): Event = EventFactory.create(hex64(7, idSeq++), pubkey, createdAt, kind, emptyArray(), "", sig)
+
+    @Test
+    fun compareDrivers() =
+        runBlocking {
+            val root: Path = Files.createTempDirectory("fs-driver-bench-")
+            val store = FsEventStore(root)
+            try {
+                val base = 1_700_000_000L
+                val span = 3_000_000L
+                val target = hex64(9, 0)
+
+                // Background: 300 authors × 100 kind-1 notes.
+                val bg = ArrayList<Event>(30_000 * SCALE + 300)
+                repeat(30_000 * SCALE) {
+                    val author = hex64(1, it % 300)
+                    bg.add(ev(author, base + (mix(it * 31L) and 0x7fffffff) % span, 1))
+                }
+                // Target author: 200 kind-1 notes + 50 kind-7 reactions.
+                repeat(200) { bg.add(ev(target, base + (mix(it * 131L) and 0x7fffffff) % span, 1)) }
+                repeat(50) { bg.add(ev(target, base + (mix(it * 61L) and 0x7fffffff) % span, 7)) }
+
+                val t0 = System.nanoTime()
+                store.transaction { bg.forEach { insert(it) } }
+                val insertMs = (System.nanoTime() - t0) / 1e6
+                println("─ FsDriverSelectionBenchmark: ${bg.size} events (scale=$SCALE), seed %.0f ms ─".format(insertMs))
+
+                // The planner's own pick — expected to choose the author
+                // tree over the ~30k-entry kind-1 tree.
+                val kindDriven = Filter(authors = listOf(target), kinds = listOf(1), limit = 50)
+                time(store, "planner (cost-based pick)") { store.query<Event>(kindDriven).size }
+
+                // Reference: author tree walked explicitly, kind checked
+                // by the caller — the pick should match or beat this.
+                time(store, "author-driver emulation") {
+                    store
+                        .query<Event>(Filter(authors = listOf(target), limit = 250))
+                        .asSequence()
+                        .filter { it.kind == 1 }
+                        .take(50)
+                        .count()
+                }
+
+                // Floor: author-only shape, the cheapest walk of the tree.
+                val authorOnly = Filter(authors = listOf(target), limit = 50)
+                time(store, "author-only floor") { store.query<Event>(authorOnly).size }
+            } finally {
+                store.close()
+                if (root.exists()) {
+                    Files.walk(root).use { it.sorted(Comparator.reverseOrder()).forEach { p -> Files.deleteIfExists(p) } }
+                }
+            }
+        }
+
+    private inline fun time(
+        store: FsEventStore,
+        label: String,
+        run: () -> Int,
+    ) {
+        repeat(3) { run() }
+        val runs = 10
+        var rows = 0
+        val start = System.nanoTime()
+        repeat(runs) { rows = run() }
+        val ms = (System.nanoTime() - start) / 1e6 / runs
+        println("    %-32s %8.2f ms  (%d rows)".format(label, ms, rows))
+    }
+}

quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/ContentlessFtsMigrationTest.kt

@@ -0,0 +1,132 @@
+/*
+ * 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.store.sqlite
+
+import androidx.sqlite.SQLiteConnection
+import androidx.sqlite.driver.bundled.BundledSQLiteDriver
+import com.vitorpamplona.quartz.nip01Core.core.Event
+import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
+import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerSync
+import com.vitorpamplona.quartz.nip10Notes.TextNoteEvent
+import com.vitorpamplona.quartz.utils.Secp256k1Instance
+import kotlinx.coroutines.runBlocking
+import java.nio.file.Files
+import java.nio.file.Path
+import kotlin.io.path.deleteIfExists
+import kotlin.test.AfterTest
+import kotlin.test.BeforeTest
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertTrue
+
+/**
+ * Verifies the real v4 → v5 FTS upgrade path: a database written with the old
+ * `fts5(event_header_row_id, content)` index (auto-assigned rowid, content
+ * stored) must, on next open, drop that table and rebuild a contentless index
+ * keyed by `event_headers.row_id`, with search intact.
+ *
+ * The v4 state is fabricated by opening a fresh v5 store, then rewriting its
+ * `event_fts` to the old schema (with deliberately wrong content, to prove the
+ * rebuild wipes it) and stamping `user_version = 4`.
+ */
+class ContentlessFtsMigrationTest {
+    private val signer = NostrSignerSync()
+    private lateinit var dbFile: Path
+
+    private fun path() = dbFile.toAbsolutePath().toString()
+
+    @BeforeTest
+    fun setup() {
+        Secp256k1Instance
+        dbFile = Files.createTempFile("contentless-fts-migration-", ".db")
+        Files.deleteIfExists(dbFile)
+    }
+
+    @AfterTest
+    fun tearDown() {
+        listOf("", "-wal", "-shm", "-journal").forEach { Path.of(dbFile.toString() + it).deleteIfExists() }
+    }
+
+    @Test
+    fun upgradesOldFtsSchemaAndRebuildsSearch() =
+        runBlocking {
+            val alpha = signer.sign(TextNoteEvent.build("uniqalpha searchable body", createdAt = 1_700_000_000L))
+            val beta = signer.sign(TextNoteEvent.build("uniqbeta searchable body", createdAt = 1_700_000_100L))
+
+            // 1. Fresh v5 store, seed events, close.
+            EventStore(dbName = path(), relay = null).also {
+                it.insert(alpha)
+                it.insert(beta)
+                it.close()
+            }
+
+            // 2. Rewrite event_fts to the pre-v5 schema with WRONG content and
+            //    downgrade user_version to 4 — the state a v4 database is in.
+            BundledSQLiteDriver().open(path()).use { db ->
+                db.exec("DROP TRIGGER IF EXISTS fts_foreign_key")
+                db.exec("DROP TABLE IF EXISTS event_fts")
+                db.exec("CREATE VIRTUAL TABLE event_fts USING fts5(event_header_row_id, content)")
+                db.exec(
+                    """
+                    CREATE TRIGGER fts_foreign_key AFTER DELETE ON event_headers FOR EACH ROW
+                    BEGIN DELETE FROM event_fts WHERE old.row_id = event_fts.event_header_row_id; END
+                    """.trimIndent(),
+                )
+                // Stale/garbage rows: a real v4 index would hold correct data,
+                // but seeding garbage proves the migration rebuilds from
+                // event_headers rather than trusting the old table.
+                db.exec("INSERT INTO event_fts(event_header_row_id, content) VALUES (1, 'uniqstale garbage')")
+                db.exec("PRAGMA user_version = 4")
+            }
+
+            // 3. Reopen with current code → onUpgrade(4→5) → migrateV4ToContentless.
+            val store = EventStore(dbName = path(), relay = null)
+            try {
+                // Rebuilt from event_headers: real content is searchable...
+                assertEquals(alpha.id, store.query<Event>(Filter(search = "uniqalpha")).single().id)
+                assertEquals(beta.id, store.query<Event>(Filter(search = "uniqbeta")).single().id)
+                // ...and the old garbage is gone.
+                assertTrue(store.query<Event>(Filter(search = "uniqstale")).isEmpty())
+
+                // The new rowid IS event_headers.row_id: the join returns the
+                // right event for each FTS rowid.
+                store.store.pool.useReader { db ->
+                    db
+                        .prepare(
+                            "SELECT h.id FROM event_fts f JOIN event_headers h ON h.row_id = f.rowid ORDER BY f.rowid",
+                        ).use { stmt ->
+                            val ids = ArrayList<String>()
+                            while (stmt.step()) ids.add(stmt.getText(0))
+                            assertEquals(listOf(alpha.id, beta.id), ids, "FTS rowid must map to event_headers.row_id")
+                        }
+                }
+
+                // The rebuilt delete trigger still cleans up FTS on delete.
+                store.store.delete(beta.id)
+                assertTrue(store.query<Event>(Filter(search = "uniqbeta")).isEmpty())
+                assertEquals(alpha.id, store.query<Event>(Filter(search = "uniqalpha")).single().id)
+            } finally {
+                store.close()
+            }
+        }
+
+    private fun SQLiteConnection.exec(sql: String) = prepare(sql).use { it.step() }
+}

quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/Fts5CapabilityProbe.kt

@@ -0,0 +1,123 @@
+/*
+ * 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.store.sqlite
+
+import androidx.sqlite.driver.bundled.BundledSQLiteDriver
+import kotlin.test.Test
+import kotlin.test.assertEquals
+
+/**
+ * Pins the FTS5 features the [FullTextSearchModule] contentless index depends
+ * on to the bundled SQLite. All three shipped between SQLite 3.43 and 3.44
+ * (2023); if the bundled driver is ever downgraded below that, this fails
+ * loudly instead of the store silently breaking search on delete.
+ *
+ *  - `content=''` **contentless** table with `contentless_delete=1`: lets the
+ *    index drop the duplicated content column yet still delete rows by rowid
+ *    (the `fts_foreign_key` trigger needs it).
+ *  - explicit `rowid` on insert (= `event_headers.row_id`): the join key and
+ *    the O(log n) delete key.
+ *  - bm25 `rank` on a contentless table, through a join: NIP-50 relevance order.
+ *  - `'merge'` / `'optimize'` maintenance commands: segment compaction.
+ */
+class Fts5CapabilityProbe {
+    @Test
+    fun contentlessDeleteAndRowidOrderingAreSupported() {
+        val db = BundledSQLiteDriver().open(":memory:")
+        try {
+            db.execSQL("CREATE VIRTUAL TABLE cl USING fts5(content, content='', contentless_delete=1)")
+            db.execSQL("INSERT INTO cl(rowid, content) VALUES (100, 'hello world')")
+            db.execSQL("INSERT INTO cl(rowid, content) VALUES (50, 'hello there')")
+            db.execSQL("INSERT INTO cl(rowid, content) VALUES (200, 'hello again')")
+            db.execSQL("DELETE FROM cl WHERE rowid = 50")
+
+            // Deleting an absent rowid must be a harmless no-op: the store's
+            // fts_foreign_key trigger fires on EVERY event_headers delete, but
+            // only searchable events ever got an FTS row.
+            db.execSQL("DELETE FROM cl WHERE rowid = 999999")
+
+            val order = ArrayList<Long>()
+            db.prepare("SELECT rowid FROM cl WHERE cl MATCH 'hello' ORDER BY rowid DESC LIMIT 5").use {
+                while (it.step()) order.add(it.getLong(0))
+            }
+            // Deleted 50 is gone; the rest come back newest-rowid first.
+            assertEquals(listOf(200L, 100L), order)
+        } finally {
+            db.close()
+        }
+    }
+
+    @Test
+    fun bm25RankWorksOnContentlessTableInAJoin() {
+        // NIP-50 orders by relevance, not created_at. Verify FTS5 bm25 `rank`
+        // works on a contentless table and is reachable through the same
+        // join-back-to-base-table shape the store's search query uses.
+        val db = BundledSQLiteDriver().open(":memory:")
+        try {
+            db.execSQL("CREATE TABLE headers (row_id INTEGER PRIMARY KEY, created_at INTEGER, tag TEXT)")
+            db.execSQL("CREATE VIRTUAL TABLE fts USING fts5(content, content='', contentless_delete=1)")
+            // row 10: term appears 3× in a short doc (most relevant).
+            // row 20: term once in a long doc (least relevant) but NEWER.
+            db.execSQL("INSERT INTO headers VALUES (10, 100, 'A')")
+            db.execSQL("INSERT INTO fts(rowid, content) VALUES (10, 'needle needle needle')")
+            db.execSQL("INSERT INTO headers VALUES (20, 999, 'B')")
+            db.execSQL("INSERT INTO fts(rowid, content) VALUES (20, 'needle alpha beta gamma delta epsilon zeta eta')")
+
+            // created_at DESC would return B (999) first; relevance returns A.
+            val byRank = ArrayList<String>()
+            db
+                .prepare(
+                    """
+                    SELECT headers.tag FROM headers
+                    INNER JOIN fts ON headers.row_id = fts.rowid
+                    WHERE fts MATCH 'needle'
+                    ORDER BY fts.rank
+                    LIMIT 10
+                    """.trimIndent(),
+                ).use { while (it.step()) byRank.add(it.getText(0)) }
+            assertEquals(listOf("A", "B"), byRank, "bm25 rank must put the more relevant (shorter, higher-tf) doc first")
+        } finally {
+            db.close()
+        }
+    }
+
+    @Test
+    fun segmentMergeAndOptimizeAreSupported() {
+        val db = BundledSQLiteDriver().open(":memory:")
+        try {
+            db.execSQL("CREATE VIRTUAL TABLE m USING fts5(content)")
+            db.execSQL("INSERT INTO m(rowid, content) VALUES (1, 'a b c')")
+            db.execSQL("INSERT INTO m(rowid, content) VALUES (2, 'd e f')")
+            // Bounded incremental merge, then a full optimize — both must parse
+            // and run without error on the bundled build.
+            db.execSQL("INSERT INTO m(m, rank) VALUES ('merge', -16)")
+            db.execSQL("INSERT INTO m(m) VALUES ('optimize')")
+
+            val hits = ArrayList<Long>()
+            db.prepare("SELECT rowid FROM m WHERE m MATCH 'e' ORDER BY rowid").use {
+                while (it.step()) hits.add(it.getLong(0))
+            }
+            assertEquals(listOf(2L), hits)
+        } finally {
+            db.close()
+        }
+    }
+}

relayBench/src/main/kotlin/com/vitorpamplona/relaybench/Scenarios.kt

@@ -70,11 +70,11 @@ object Scenarios {
         }
 
         val topAuthors = notesByAuthor.entries.sortedWith(compareByDescending<Map.Entry<String, Int>> { it.value }.thenBy { it.key }).map { it.key }
-        val hottestThread =
+        val hotNotes =
             eTagRefs.entries
                 .sortedWith(compareByDescending<Map.Entry<String, Int>> { it.value }.thenBy { it.key })
-                .firstOrNull()
-                ?.key
+                .map { it.key }
+        val hottestThread = hotNotes.firstOrNull()
         val mostMentioned =
             pTagRefs.entries
                 .sortedWith(compareByDescending<Map.Entry<String, Int>> { it.value }.thenBy { it.key })
@@ -86,6 +86,23 @@ object Scenarios {
                 .firstOrNull()
                 ?.key
 
+        // The author that most often tags the most-mentioned pubkey — a
+        // conversation pair for the tag ∩ author (DM-room) query shape.
+        val conversationPeer =
+            mostMentioned?.let { me ->
+                val byAuthor = HashMap<String, Int>()
+                for (e in events) {
+                    if (e.kind != 1 || e.pubKey == me) continue
+                    if (e.tags.any { it.size >= 2 && it[0] == "p" && it[1] == me }) {
+                        byAuthor.merge(e.pubKey, 1, Int::plus)
+                    }
+                }
+                byAuthor.entries
+                    .sortedWith(compareByDescending<Map.Entry<String, Int>> { it.value }.thenBy { it.key })
+                    .firstOrNull()
+                    ?.key
+            }
+
         // Evenly spread sample of note ids — a "fetch these 100 events" batch.
         val idSample =
             if (noteIds.size <= 100) {
@@ -159,6 +176,33 @@ object Scenarios {
                     ),
                 )
             }
+            if (mostMentioned != null && conversationPeer != null) {
+                // The tag ∩ author ∩ kind shape (65 client assembler call
+                // sites: NIP-04 DM rooms, reports-by-follows, follows-scoped
+                // community feeds). Modeled on kind 1 because public corpora
+                // carry no DMs; the index path exercised is identical.
+                add(
+                    Scenario(
+                        "conversation",
+                        "notes by one author tagging the most-mentioned pubkey (DM-room shape)",
+                        Filter(kinds = listOf(1), authors = listOf(conversationPeer), tags = mapOf("p" to listOf(mostMentioned)), limit = 500),
+                    ),
+                )
+            }
+            if (hotNotes.size > 1) {
+                // Large-IN tag watcher: per-value streams come sorted off the
+                // tag index but their union does not, exposing whether the
+                // store collects+sorts or merges. 150 values stays inside
+                // strfry's default 200-element filter cap.
+                val watched = hotNotes.take(150)
+                add(
+                    Scenario(
+                        "reactions-watch",
+                        "reactions on the ${watched.size} hottest notes (visible-feed reaction watcher)",
+                        Filter(kinds = listOf(7), tags = mapOf("e" to watched), limit = 500),
+                    ),
+                )
+            }
             topHashtag?.let {
                 add(
                     Scenario(