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

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

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/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/MergeQueryExecutor.kt

@@ -24,63 +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 {
-    // TODO: the same collect-all + TEMP-B-TREE-sort pattern exists one index
-    // over, on the tag path: `kinds + tags(#e IN [hundreds]) + limit` (the
-    // reactions/replies watcher archetype) unions per-value streams that are
-    // each sorted off `(tag_hash, kind, created_at)` and sorts the union.
-    // `streamCount` currently rejects any filter with tags, so those queries
-    // never merge. Measured by `TagAuthorIndexBenchmark`: `#e IN 300,
-    // limit 500` costs 12.8 ms cold at 200k events and 14.2 ms at 1M
-    // (6.7 ms with indexTagsWithKindAndPubkey on) — tolerable, but it
-    // scales with matching history like the follow-feed shape did; extend
-    // the merge to per-tag-value streams if the relayBench
-    // `reactions-watch` scenario shows it in the profile vs strfry.
+    /** 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
@@ -94,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
@@ -128,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)
@@ -139,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 =
@@ -160,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)
@@ -199,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
@@ -215,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()
@@ -213,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)
+                }
             }
         }
     }
@@ -274,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/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/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/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()
+        }
+    }
+}