Merge pull request #3846 from vitorpamplona/claude/30382-rank-follower-providers-j2cnb9

fix: request follower-count provider's 30382 cards in UserCardsSubAssembler

amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/reqCommand/user/watchers/UserCardsSubAssembler.kt

@@ -84,9 +84,14 @@ class UserCardsSubAssembler(
                         add(it, account.userProfile().pubkeyHex)
                     }
                 }
-                accounts.map { it.trustProviderList.liveUserRankProvider.value }.forEach { account ->
-                    if (account != null) {
-                        add(account.relayUrl, account.pubkey)
+                accounts.map { it.trustProviderList.liveUserRankProvider.value }.forEach { provider ->
+                    if (provider != null) {
+                        add(provider.relayUrl, provider.pubkey)
+                    }
+                }
+                accounts.map { it.trustProviderList.liveUserFollowerCount.value }.forEach { provider ->
+                    if (provider != null) {
+                        add(provider.relayUrl, provider.pubkey)
                     }
                 }
             }

commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/relayClient/subscriptions/ExplainedFilterTest.kt

@@ -108,7 +108,9 @@ class ExplainedFilterTest {
      */
     @Test
     fun `copy preserves the purpose`() {
-        val advanced = explained().copy(since = 1_785_379_272)
+        // Typed as the base Filter so the is-check below stays a runtime assertion — with the
+        // override's covariant return type inferred, the compiler would prove it true statically.
+        val advanced: Filter = explained().copy(since = 1_785_379_272)
 
         assertTrue("copy() must stay an ExplainedFilter", advanced is ExplainedFilter)
         assertEquals(SubPurpose.NOTIFICATIONS, advanced.purposeOrNull())

quartz/RELAY.md

@@ -259,15 +259,25 @@ q.domain       // "example.com"
 q.nsfwIncluded // false  (NIP-50 default is true when the token is absent)
 ```
 
-The SQLite store expects `Filter.search` to be plain FTS text: `:` is FTS5
-column-filter syntax, so a raw `include:spam` reaching MATCH raises "no such
-column: include" instead of matching. Strip the tokens before querying with
-`SearchQuery.stripExtensions(raw)` or `filter.strippingSearchExtensions()` —
-an extensions-only query collapses to an empty search, which imposes no
-constraint (NIP-50: unsupported extensions are ignored, not match-nothing).
-The storage-backed server path (`NostrServer` / `LiveEventStore`) already does
-this, so relays like geode comply out of the box; `EventSource` backends get
-the raw string because a real search backend wants the extensions.
+`Filter.search` reaches every backend **verbatim**, extension tokens included
+— the relay layer never rewrites it. Which extensions are directives and which
+are noise is a property of the store, so the `IEventStore` contract puts the
+decision there: a store that implements an extension (rank profiles, trust
+floors, observer-relative scoring) parses the raw string with
+`SearchQuery.parse`; a store that doesn't must ignore the tokens per NIP-50
+(not match them as literal text, not return nothing). The built-in SQLite and
+filesystem stores do the latter by stripping at their own boundary with
+`filter.strippingSearchExtensions()` — FTS5 treats `:` as column-filter syntax,
+so a raw `include:spam` reaching MATCH would raise "no such column: include" —
+and an extensions-only query collapses to an empty search, which imposes no
+constraint. Relays like geode therefore comply out of the box, and
+`EventSource` backends likewise get the raw string.
+
+Observer-relative stores (web-of-trust ranking, "for-you" relevance) read the
+caller's NIP-42-authenticated pubkeys from the coroutine context via
+`StoreQueryContext` — `LiveEventStore` installs it around every REQ/COUNT store
+call for authenticated connections. It is ranking context only: it may reorder
+results, never change which events match.
 
 A search/redirector relay is just a custom policy (or, for computed results, a
 custom `IEventStore` whose `query` answers the REQ) that reads the parsed query:

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/cache/interning/InterningEventStore.kt

@@ -116,7 +116,8 @@ class InterningEventStore(
     override suspend fun snapshotIdsForNegentropy(
         filters: List<Filter>,
         maxEntries: Int?,
-    ): List<IdAndTime> = inner.snapshotIdsForNegentropy(filters, maxEntries)
+        onProgress: ((collected: Int) -> Unit)?,
+    ): List<IdAndTime> = inner.snapshotIdsForNegentropy(filters, maxEntries, onProgress)
 
     override suspend fun delete(filter: Filter) = inner.delete(filter)
 

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

@@ -53,6 +53,7 @@ import kotlinx.coroutines.CoroutineStart
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.channels.ClosedSendChannelException
 import kotlinx.coroutines.launch
+import kotlin.concurrent.Volatile
 import kotlin.concurrent.atomics.AtomicLong
 import kotlin.concurrent.atomics.ExperimentalAtomicApi
 
@@ -82,8 +83,18 @@ class RelaySession(
      * The authenticated-identity store for this connection. The engine is the
      * only writer (committed in [handleAuth] on a successful NIP-42 AUTH); the
      * policy and the data plane read it through [requestContext].
+     *
+     * Copy-on-write on purpose: readers run concurrently with the engine —
+     * a REQ replay coroutine can hold the set (via `StoreQueryContext` or an
+     * `EventSource` reading [RequestContext.authenticatedUsers]) while a
+     * later AUTH frame commits a new identity. Each read hands out the
+     * current **immutable** set, so an in-flight query keeps a consistent
+     * snapshot instead of racing a mutating `LinkedHashSet`; `@Volatile`
+     * makes the swapped reference visible across threads. AUTH is rare, so
+     * the copy costs nothing on the hot path.
      */
-    private val authenticatedUsers = mutableSetOf<HexKey>()
+    @Volatile
+    private var authenticatedUsers = setOf<HexKey>()
 
     /**
      * The per-connection scope. Handed to the [policy] at connect (so gating
@@ -261,7 +272,7 @@ class RelaySession(
 
         // Single, engine-side commit into the connection scope — after the full
         // chain approved and a verifying policy voted to record.
-        if (record) authenticatedUsers.add(cmd.event.pubKey)
+        if (record) authenticatedUsers = authenticatedUsers + cmd.event.pubKey
 
         send(OkMessage(cmd.event.id, true, ""))
     }

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

@@ -27,10 +27,11 @@ import com.vitorpamplona.quartz.nip01Core.relay.filters.FilterIndex
 import com.vitorpamplona.quartz.nip01Core.store.IEventStore
 import com.vitorpamplona.quartz.nip01Core.store.IdAndTime
 import com.vitorpamplona.quartz.nip01Core.store.RawEvent
-import com.vitorpamplona.quartz.nip50Search.strippingSearchExtensions
+import com.vitorpamplona.quartz.nip01Core.store.StoreQueryContext
 import com.vitorpamplona.quartz.utils.TimeUtils
 import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.awaitCancellation
+import kotlinx.coroutines.withContext
 import kotlin.concurrent.atomics.AtomicBoolean
 import kotlin.concurrent.atomics.AtomicLong
 import kotlin.concurrent.atomics.AtomicReference
@@ -50,14 +51,14 @@ import kotlin.concurrent.atomics.ExperimentalAtomicApi
  * match. This avoids the quadratic O(N_subscribers × N_filters_per_sub) per-event walk
  * that a SharedFlow-based broadcast would do.
  *
- * NIP-50 `search` strings are handed to the store with their `key:value`
- * extension tokens stripped ([strippingSearchExtensions]): the SQLite FTS
- * backend treats `:` as column-filter syntax, so a raw `include:spam`
- * would raise "no such column" instead of matching. Per NIP-50, an
- * unsupported extension is ignored — an extensions-only search therefore
- * becomes unconstrained, not match-nothing. Relays that *do* implement
- * extensions serve search through an [EventSource] backend, which
- * receives the raw string.
+ * NIP-50 `search` strings are handed to the store **verbatim**, extension
+ * tokens included: whether `include:spam` is a directive, ignorable noise,
+ * or poison for a text engine is a property of the store, so each
+ * [IEventStore] implementation makes that call itself (see the NIP-50
+ * contract on [IEventStore] — the built-in SQLite and filesystem stores
+ * strip the tokens at their own boundary). This layer also installs a
+ * [StoreQueryContext] around each store call so observer-relative stores
+ * can read the connection's NIP-42 identity.
  *
  * @property store The underlying persistent storage for events.
  * @property ingest The group-commit writer pipeline. Accepted events fan out via the
@@ -208,6 +209,25 @@ class LiveEventStore(
         }
     }
 
+    /**
+     * Runs [block] with a [StoreQueryContext] carrying the connection's
+     * NIP-42-authenticated pubkeys, so observer-relative stores can read
+     * the caller's identity from the coroutine context. Skipped entirely
+     * for unauthenticated connections — the element's contract is
+     * "present means non-empty".
+     */
+    private suspend inline fun <R> withCallerIdentity(
+        ctx: RequestContext,
+        crossinline block: suspend () -> R,
+    ): R {
+        val users = ctx.authenticatedUsers
+        return if (users.isEmpty()) {
+            block()
+        } else {
+            withContext(StoreQueryContext(users)) { block() }
+        }
+    }
+
     /**
      * With deferred FTS, a search query must first drain the catch-up
      * backlog — that keeps NIP-50 results exactly as fresh as the
@@ -248,9 +268,11 @@ class LiveEventStore(
 
         index.register(filters, sub)
         try {
-            store.query<Event>(filters.strippingSearchExtensions()) { event ->
-                seen.record(event.id)
-                onEach(event)
+            withCallerIdentity(ctx) {
+                store.query<Event>(filters) { event ->
+                    seen.record(event.id)
+                    onEach(event)
+                }
             }
             onEose()
             // Drop the dedupe set so the live path stops paying for
@@ -295,9 +317,11 @@ class LiveEventStore(
 
         index.register(filters, sub)
         try {
-            store.rawQuery(filters.strippingSearchExtensions()) { raw ->
-                seen.record(raw.id)
-                onEachStored(raw)
+            withCallerIdentity(ctx) {
+                store.rawQuery(filters) { raw ->
+                    seen.record(raw.id)
+                    onEachStored(raw)
+                }
             }
             onEose()
             seen.release()
@@ -312,7 +336,7 @@ class LiveEventStore(
         filters: List<Filter>,
     ): Int {
         drainFtsIfSearching(filters)
-        return store.count(filters.strippingSearchExtensions())
+        return withCallerIdentity(ctx) { store.count(filters) }
     }
 
     /**
@@ -320,7 +344,7 @@ class LiveEventStore(
      * needs the full set of event ids matching the filter at the
      * moment the NEG-OPEN arrives, not a streamed/live result.
      */
-    suspend fun snapshotQuery(filter: Filter): List<Event> = store.query(filter.strippingSearchExtensions())
+    suspend fun snapshotQuery(filter: Filter): List<Event> = store.query(filter)
 
     /**
      * Multi-filter snapshot. Unions the per-filter results and
@@ -333,7 +357,7 @@ class LiveEventStore(
         val seen = HashSet<String>()
         val merged = ArrayList<Event>()
         for (f in filters) {
-            for (e in store.query<Event>(f.strippingSearchExtensions())) {
+            for (e in store.query<Event>(f)) {
                 if (seen.add(e.id)) merged += e
             }
         }
@@ -353,7 +377,7 @@ class LiveEventStore(
     override suspend fun snapshotIdsForNegentropy(
         filters: List<Filter>,
         maxEntries: Int?,
-    ): List<IdAndTime> = store.snapshotIdsForNegentropy(filters.strippingSearchExtensions(), maxEntries)
+    ): List<IdAndTime> = store.snapshotIdsForNegentropy(filters, maxEntries)
 
     // ------------------------------------------------------------------
     // NIP-77 snapshot cache

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

@@ -63,7 +63,10 @@ interface RequestContext {
      * The pubkeys that have authenticated on this connection via NIP-42. Empty
      * when the connection is unauthenticated. Backed by the engine-owned scope
      * and read live, so a REQ that arrives after a successful AUTH sees the
-     * freshly recorded pubkey(s).
+     * freshly recorded pubkey(s). Each read returns an **immutable snapshot**
+     * (the engine swaps the set copy-on-write on AUTH), so holding one across
+     * a long replay is safe — it just won't grow if another AUTH lands
+     * mid-query.
      */
     val authenticatedUsers: Set<HexKey>
 }

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/IEventStore.kt

@@ -28,6 +28,35 @@ import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
 import com.vitorpamplona.quartz.nip59Giftwrap.wraps.GiftWrapEvent
 import com.vitorpamplona.quartz.nip65RelayList.AdvertisedRelayListEvent
 
+/**
+ * Storage contract for Nostr events: insert, filter-query, count, delete,
+ * NIP-50 full-text search, and NIP-77 negentropy snapshots.
+ *
+ * ## NIP-50 `search` contract
+ *
+ * Query/count/delete filters arrive with [Filter.search] **verbatim as the
+ * client sent it**, `key:value` extension tokens included (`include:spam`,
+ * `domain:example.com`, `language:en`, …). No layer above the store rewrites
+ * the string, so each implementation decides which extensions it supports:
+ *
+ * - A store that interprets an extension (rank profiles, trust floors,
+ *   observer-relative scoring, …) reads it from the raw string — parse it
+ *   with [com.vitorpamplona.quartz.nip50Search.SearchQuery.parse].
+ * - Extensions the store does NOT support must be **ignored, not matched as
+ *   literal text and not treated as match-nothing** (NIP-50: relays "SHOULD
+ *   ignore extensions they don't support"). Stores whose text engine would
+ *   choke on the raw tokens strip them first with
+ *   [com.vitorpamplona.quartz.nip50Search.strippingSearchExtensions] — this
+ *   is what the built-in SQLite and filesystem stores do. An extensions-only
+ *   search therefore collapses to an unconstrained query.
+ *
+ * ## Caller identity
+ *
+ * Ranked/observer-relative stores can read the caller's NIP-42-authenticated
+ * identity from the coroutine context via [StoreQueryContext]; the relay
+ * layer installs it around every REQ/COUNT-driven store call. It is ranking
+ * context only and absent for unauthenticated callers.
+ */
 interface IEventStore : AutoCloseable {
     companion object {
         /**
@@ -37,6 +66,12 @@ interface IEventStore : AutoCloseable {
          * small enough that a pause request is honoured promptly.
          */
         const val DEFAULT_FTS_REINDEX_BATCH = 1000
+
+        /**
+         * Suggested [snapshotIdsForNegentropy] `onProgress` cadence: report
+         * the running count roughly every this many collected entries.
+         */
+        const val NEGENTROPY_PROGRESS_EVERY = 1000
     }
 
     /**
@@ -173,6 +208,13 @@ interface IEventStore : AutoCloseable {
      * guard). The +1 sentinel lets the caller distinguish "exactly
      * capped" from "too many to fit".
      *
+     * [onProgress] is a liveness hook for corpora large enough that the
+     * walk takes minutes: implementations SHOULD invoke it every
+     * [NEGENTROPY_PROGRESS_EVERY]-ish collected entries with the running
+     * count. Callers must not rely on any particular cadence — a store
+     * that answers from an index may legitimately never call it. Pass
+     * `null` (the default) to opt out at zero cost.
+     *
      * Default implementation falls back to the full-decode path so
      * non-SQLite stores stay correct; SQLite overrides with a direct
      * `SELECT id, created_at` against the `query_by_created_at_id`
@@ -182,8 +224,13 @@ interface IEventStore : AutoCloseable {
     suspend fun snapshotIdsForNegentropy(
         filters: List<Filter>,
         maxEntries: Int? = null,
+        onProgress: ((collected: Int) -> Unit)? = null,
     ): List<IdAndTime> {
-        val all = query<Event>(filters).map { IdAndTime(it.createdAt, it.id) }
+        val all = ArrayList<IdAndTime>()
+        query<Event>(filters) { event ->
+            all.add(IdAndTime(event.createdAt, event.id))
+            if (onProgress != null && all.size % NEGENTROPY_PROGRESS_EVERY == 0) onProgress(all.size)
+        }
         return if (maxEntries != null && all.size > maxEntries + 1) {
             all.subList(0, maxEntries + 1)
         } else {
@@ -258,6 +305,11 @@ interface IEventStore : AutoCloseable {
      *
      * Process roughly [batchSize] events starting from [resumeFrom]
      * (`null` = from the beginning) and return a [FtsReindexProgress].
+     * [resumeFrom] / [FtsReindexProgress.cursor] is an **opaque,
+     * store-defined string**: callers persist it and pass it back
+     * unchanged, and must never parse, order, or compare it (SQLite
+     * encodes a kind + row id; another store may carry an engine
+     * continuation token).
      * Drive it in a loop, feeding [FtsReindexProgress.cursor] back in,
      * until [FtsReindexProgress.done] is `true`:
      *

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/ObservableEventStore.kt

@@ -192,7 +192,8 @@ class ObservableEventStore(
     override suspend fun snapshotIdsForNegentropy(
         filters: List<Filter>,
         maxEntries: Int?,
-    ): List<IdAndTime> = inner.snapshotIdsForNegentropy(filters, maxEntries)
+        onProgress: ((collected: Int) -> Unit)?,
+    ): List<IdAndTime> = inner.snapshotIdsForNegentropy(filters, maxEntries, onProgress)
 
     override suspend fun liveNegentropySnapshot(maxEntries: Int) = inner.liveNegentropySnapshot(maxEntries)
 

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/StoreQueryContext.kt

@@ -0,0 +1,65 @@
+/*
+ * 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
+
+import com.vitorpamplona.quartz.nip01Core.core.HexKey
+import kotlin.coroutines.AbstractCoroutineContextElement
+import kotlin.coroutines.CoroutineContext
+
+/**
+ * The published caller-identity seam between a relay layer and an
+ * [IEventStore]: who is asking, carried on the coroutine context so it
+ * crosses the store boundary — and any decorator in between — without
+ * widening every `query`/`count` signature.
+ *
+ * The storage-backed relay path (`LiveEventStore`) installs this element
+ * around every REQ/COUNT-driven store call when the connection has
+ * NIP-42-authenticated pubkeys. A store whose results are
+ * observer-relative — web-of-trust ranking, "for-you" relevance, trust
+ * floors — reads it back:
+ *
+ * ```
+ * val observer = coroutineContext[StoreQueryContext]?.observer
+ * ```
+ *
+ * Contract: this is **ranking context only**. It may reorder or score
+ * results; it must never change *which* events match a filter — access
+ * control belongs to the relay policy layer, not the store. The element
+ * is absent for unauthenticated callers (and for direct store use outside
+ * a relay), so every read needs a null-tolerant fallback such as an
+ * operator-configured default observer.
+ */
+class StoreQueryContext(
+    /**
+     * The pubkeys authenticated on the calling connection via NIP-42, in
+     * no particular order. Never empty — the relay layer skips installing
+     * the element instead of installing an empty one.
+     */
+    val authenticatedUsers: Set<HexKey>,
+) : AbstractCoroutineContextElement(StoreQueryContext) {
+    companion object Key : CoroutineContext.Key<StoreQueryContext>
+
+    /**
+     * Convenience for the common single-identity case: one of
+     * [authenticatedUsers], or `null` when the set is empty.
+     */
+    val observer: HexKey? get() = authenticatedUsers.firstOrNull()
+}

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

@@ -82,7 +82,8 @@ class EventStore(
     override suspend fun snapshotIdsForNegentropy(
         filters: List<Filter>,
         maxEntries: Int?,
-    ): List<IdAndTime> = store.snapshotIdsForNegentropy(filters, maxEntries)
+        onProgress: ((collected: Int) -> Unit)?,
+    ): List<IdAndTime> = store.snapshotIdsForNegentropy(filters, maxEntries, onProgress)
 
     override suspend fun liveNegentropySnapshot(maxEntries: Int) = store.liveNegentropySnapshot(maxEntries)
 

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

@@ -28,6 +28,7 @@ import com.vitorpamplona.quartz.nip01Core.core.Kind
 import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper
 import com.vitorpamplona.quartz.nip01Core.core.isAddressable
 import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
+import com.vitorpamplona.quartz.nip01Core.store.IEventStore
 import com.vitorpamplona.quartz.nip01Core.store.IdAndTime
 import com.vitorpamplona.quartz.nip01Core.store.RawEvent
 import com.vitorpamplona.quartz.nip01Core.store.sqlite.sql.where
@@ -258,6 +259,7 @@ class QueryBuilder(
         filters: List<Filter>,
         db: SQLiteConnection,
         maxEntries: Int? = null,
+        onProgress: ((collected: Int) -> Unit)? = null,
     ): List<IdAndTime> {
         val inner =
             if (filters.size == 1) {
@@ -278,7 +280,7 @@ class QueryBuilder(
             } else {
                 inner
             }
-        return db.runIdAndTimeQuery(query)
+        return db.runIdAndTimeQuery(query, onProgress)
     }
 
     private fun toSnapshotIdsSql(
@@ -454,7 +456,10 @@ class QueryBuilder(
         return QuerySpec(sql, clause.args)
     }
 
-    private fun SQLiteConnection.runIdAndTimeQuery(query: QuerySpec): List<IdAndTime> =
+    private fun SQLiteConnection.runIdAndTimeQuery(
+        query: QuerySpec,
+        onProgress: ((collected: Int) -> Unit)? = null,
+    ): List<IdAndTime> =
         prepare(query.sql).use { stmt ->
             query.args.forEachIndexed { index, arg ->
                 stmt.bindText(index + 1, arg)
@@ -462,6 +467,7 @@ class QueryBuilder(
             val results = ArrayList<IdAndTime>()
             while (stmt.step()) {
                 results.add(IdAndTime(stmt.getLong(1), stmt.getText(0)))
+                if (onProgress != null && results.size % IEventStore.NEGENTROPY_PROGRESS_EVERY == 0) onProgress(results.size)
             }
             results
         }

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

@@ -39,6 +39,7 @@ import com.vitorpamplona.quartz.nip01Core.store.IdAndTime
 import com.vitorpamplona.quartz.nip01Core.store.RawEvent
 import com.vitorpamplona.quartz.nip09Deletions.DeletionEvent
 import com.vitorpamplona.quartz.nip40Expiration.isExpired
+import com.vitorpamplona.quartz.nip50Search.strippingSearchExtensions
 import com.vitorpamplona.quartz.nip62RequestToVanish.RequestToVanishEvent
 import com.vitorpamplona.quartz.nip65RelayList.AdvertisedRelayListEvent
 import com.vitorpamplona.quartz.nip77Negentropy.LiveNegentropyIndex
@@ -537,48 +538,58 @@ class SQLiteEventStore(
         }
     }
 
-    suspend fun <T : Event> query(filter: Filter): List<T> = pool.useReader { queryBuilder.query(filter, it) }
+    // Every filter-accepting read/delete strips NIP-50 extension tokens at
+    // this boundary: FTS5 treats `:` as column-filter syntax, so a raw
+    // `include:spam` reaching MATCH raises "no such column" instead of
+    // matching. This store implements no extensions, and NIP-50 says
+    // unsupported extensions are ignored — an extensions-only search
+    // therefore collapses to an unconstrained query, not match-nothing.
+    // Stores that DO interpret extensions receive the raw string from the
+    // relay layer and make their own call (see the IEventStore contract).
 
-    suspend fun <T : Event> query(filters: List<Filter>): List<T> = pool.useReader { queryBuilder.query(filters, it) }
+    suspend fun <T : Event> query(filter: Filter): List<T> = pool.useReader { queryBuilder.query(filter.strippingSearchExtensions(), it) }
+
+    suspend fun <T : Event> query(filters: List<Filter>): List<T> = pool.useReader { queryBuilder.query(filters.strippingSearchExtensions(), it) }
 
     suspend fun <T : Event> query(
         filter: Filter,
         onEach: (T) -> Unit,
-    ) = pool.useReader { queryBuilder.query(filter, it, onEach) }
+    ) = pool.useReader { queryBuilder.query(filter.strippingSearchExtensions(), it, onEach) }
 
     suspend fun <T : Event> query(
         filters: List<Filter>,
         onEach: (T) -> Unit,
-    ) = pool.useReader { queryBuilder.query(filters, it, onEach) }
+    ) = pool.useReader { queryBuilder.query(filters.strippingSearchExtensions(), it, onEach) }
 
-    suspend fun rawQuery(filter: Filter): List<RawEvent> = pool.useReader { queryBuilder.rawQuery(filter, it) }
+    suspend fun rawQuery(filter: Filter): List<RawEvent> = pool.useReader { queryBuilder.rawQuery(filter.strippingSearchExtensions(), it) }
 
-    suspend fun rawQuery(filters: List<Filter>): List<RawEvent> = pool.useReader { queryBuilder.rawQuery(filters, it) }
+    suspend fun rawQuery(filters: List<Filter>): List<RawEvent> = pool.useReader { queryBuilder.rawQuery(filters.strippingSearchExtensions(), it) }
 
     suspend fun rawQuery(
         filter: Filter,
         onEach: (RawEvent) -> Unit,
-    ) = pool.useReader { queryBuilder.rawQuery(filter, it, onEach) }
+    ) = pool.useReader { queryBuilder.rawQuery(filter.strippingSearchExtensions(), it, onEach) }
 
     suspend fun rawQuery(
         filters: List<Filter>,
         onEach: (RawEvent) -> Unit,
-    ) = pool.useReader { queryBuilder.rawQuery(filters, it, onEach) }
+    ) = pool.useReader { queryBuilder.rawQuery(filters.strippingSearchExtensions(), it, onEach) }
 
-    suspend fun planQuery(filter: Filter) = pool.useReader { queryBuilder.planQuery(filter, seedModule.hasher(it), it) }
+    suspend fun planQuery(filter: Filter) = pool.useReader { queryBuilder.planQuery(filter.strippingSearchExtensions(), seedModule.hasher(it), it) }
 
-    suspend fun planQuery(filters: List<Filter>) = pool.useReader { queryBuilder.planQuery(filters, seedModule.hasher(it), it) }
+    suspend fun planQuery(filters: List<Filter>) = pool.useReader { queryBuilder.planQuery(filters.strippingSearchExtensions(), seedModule.hasher(it), it) }
 
-    suspend fun count(filter: Filter): Int = pool.useReader { queryBuilder.count(filter, it) }
+    suspend fun count(filter: Filter): Int = pool.useReader { queryBuilder.count(filter.strippingSearchExtensions(), it) }
 
-    suspend fun count(filters: List<Filter>): Int = pool.useReader { queryBuilder.count(filters, it) }
+    suspend fun count(filters: List<Filter>): Int = pool.useReader { queryBuilder.count(filters.strippingSearchExtensions(), it) }
 
     suspend fun authorsMissingOutbox(): List<HexKey> = pool.useReader { queryBuilder.authorsMissingKind(AdvertisedRelayListEvent.KIND, it) }
 
     suspend fun snapshotIdsForNegentropy(
         filters: List<Filter>,
         maxEntries: Int? = null,
-    ): List<IdAndTime> = pool.useReader { queryBuilder.snapshotIdsForNegentropy(filters, it, maxEntries) }
+        onProgress: ((collected: Int) -> Unit)? = null,
+    ): List<IdAndTime> = pool.useReader { queryBuilder.snapshotIdsForNegentropy(filters.strippingSearchExtensions(), it, maxEntries, onProgress) }
 
     // The invalidate() calls below run INSIDE useWriter: dropping the
     // index while still holding the mutex means no NEG-OPEN can seal a
@@ -588,7 +599,7 @@ class SQLiteEventStore(
 
     suspend fun delete(filter: Filter) {
         pool.useWriter {
-            queryBuilder.delete(filter, it)
+            queryBuilder.delete(filter.strippingSearchExtensions(), it)
             // Delete-by-filter can't itemize what it removed; drop the
             // live index and let the next NEG-OPEN rebuild from one scan.
             liveNegentropyIndex?.invalidate()
@@ -597,7 +608,7 @@ class SQLiteEventStore(
 
     suspend fun delete(filters: List<Filter>) {
         pool.useWriter {
-            queryBuilder.delete(filters, it)
+            queryBuilder.delete(filters.strippingSearchExtensions(), it)
             liveNegentropyIndex?.invalidate()
         }
     }

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayObserver.kt

@@ -73,10 +73,6 @@ class RelayObserver : RelayConnectionListener {
     class Observation(
         val url: NormalizedRelayUrl,
     ) {
-        // Monotonic marks, not wall clock: these measure durations, and a clock
-        // step mid-connection must not produce a negative or wild latency.
-        @Volatile var connectingAt: TimeSource.Monotonic.ValueTimeMark? = null
-
         @Volatile var rttOpenMs: Long? = null
 
         @Volatile var firstReqAt: TimeSource.Monotonic.ValueTimeMark? = null
@@ -124,7 +120,6 @@ class RelayObserver : RelayConnectionListener {
 
     override fun onConnecting(relay: IRelayClient) {
         val o = of(relay)
-        o.connectingAt = TimeSource.Monotonic.markNow()
         // Cleared, not kept: a reconnect is a fresh attempt, and carrying an old
         // error forward would report a working relay as broken for as long as the
         // process lives after one bad minute.
@@ -140,7 +135,20 @@ class RelayObserver : RelayConnectionListener {
         val o = of(relay)
         o.reachable = true
         o.error = null
-        o.connectingAt?.let { o.rttOpenMs = it.elapsedNow().inWholeMilliseconds.coerceAtLeast(0) }
+        // The TRANSPORT's number, not ours. pingMillis is
+        // receivedResponseAtMillis - sentRequestAtMillis, so it starts when the
+        // upgrade request actually goes out and excludes everything before it.
+        //
+        // Timing onConnecting -> onConnected instead measures our own dispatcher
+        // queue as if it were the relay's latency. Under a 16,507-relay fan-out
+        // that queue dominates: published records showed a median rtt-open of
+        // 33.5 SECONDS and a max of 90, against a true minimum of 140ms. That is
+        // the field aggregators rank relays by, so it was worse than publishing
+        // nothing — a slow-looking relay that is not slow.
+        //
+        // Zero or negative means the transport could not time it; no timing is
+        // published rather than a fabricated one.
+        o.rttOpenMs = pingMillis.toLong().takeIf { it > 0 }
         o.touch()
     }
 

quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/StoreQueryContextTest.kt

@@ -0,0 +1,152 @@
+/*
+ * 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.server
+
+import com.vitorpamplona.quartz.nip01Core.core.Event
+import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper
+import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.AuthMessage
+import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
+import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
+import com.vitorpamplona.quartz.nip01Core.relay.server.policies.EmptyPolicy
+import com.vitorpamplona.quartz.nip01Core.relay.server.policies.FullAuthPolicy
+import com.vitorpamplona.quartz.nip01Core.relay.server.policies.IRelayPolicy
+import com.vitorpamplona.quartz.nip01Core.store.IEventStore
+import com.vitorpamplona.quartz.nip01Core.store.RawEvent
+import com.vitorpamplona.quartz.nip01Core.store.StoreQueryContext
+import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore
+import com.vitorpamplona.quartz.nip42RelayAuth.RelayAuthEvent
+import com.vitorpamplona.quartz.utils.TimeUtils
+import kotlinx.coroutines.CoroutineDispatcher
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.test.UnconfinedTestDispatcher
+import kotlinx.coroutines.test.runTest
+import kotlin.coroutines.coroutineContext
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertNull
+import kotlin.test.assertTrue
+
+/**
+ * The caller-identity seam: `LiveEventStore` must install a
+ * [StoreQueryContext] with the connection's NIP-42-authenticated pubkeys
+ * around every REQ-driven store call — and must NOT install one for
+ * unauthenticated connections — so observer-relative stores can read who
+ * is asking straight off the coroutine context.
+ */
+@OptIn(ExperimentalCoroutinesApi::class)
+class StoreQueryContextTest {
+    private val pubkey = "46fcbe3065eaf1ae7811465924e48923363ff3f526bd6f73d7c184b16bd8ce4d"
+    private val sig = "4aa5264965018fa12a326686ad3d3bd8beae3218dcc83689b19ca1e6baeb791531943c15363aa6707c7c0c8b2d601deca1f20c32078b2872d356cdca03b04cce"
+    private val relayUrl = NormalizedRelayUrl("wss://relay.example.com/")
+
+    private fun hexId(n: Int): String = n.toString().padStart(64, '0')
+
+    /**
+     * Delegates everything to a real SQLite store but records the
+     * [StoreQueryContext] visible during each query — both the decoding
+     * and the zero-decode replay path, since the session may use either.
+     */
+    private class ContextRecordingStore(
+        private val inner: IEventStore,
+    ) : IEventStore by inner {
+        val contexts = mutableListOf<StoreQueryContext?>()
+
+        override suspend fun <T : Event> query(
+            filters: List<Filter>,
+            onEach: (T) -> Unit,
+        ) {
+            contexts.add(coroutineContext[StoreQueryContext])
+            inner.query(filters, onEach)
+        }
+
+        override suspend fun rawQuery(
+            filters: List<Filter>,
+            onEach: (RawEvent) -> Unit,
+        ) {
+            contexts.add(coroutineContext[StoreQueryContext])
+            inner.rawQuery(filters, onEach)
+        }
+    }
+
+    private fun createServer(
+        dispatcher: CoroutineDispatcher,
+        store: IEventStore,
+        policyBuilder: () -> IRelayPolicy,
+    ): NostrServer =
+        NostrServer(
+            store = store,
+            policyBuilder = policyBuilder,
+            parentContext = dispatcher,
+        )
+
+    @Test
+    fun unauthenticatedReqSeesNoStoreQueryContext() =
+        runTest {
+            val dispatcher = UnconfinedTestDispatcher(testScheduler)
+            val store = ContextRecordingStore(EventStore(null))
+            val server = createServer(dispatcher, store) { EmptyPolicy }
+            val session = server.connect { }
+
+            session.receive("""["REQ","sub1",{"kinds":[1]}]""")
+
+            assertEquals(1, store.contexts.size, "the REQ must have reached the store")
+            assertNull(store.contexts[0], "no NIP-42 auth → no StoreQueryContext element")
+
+            server.close()
+        }
+
+    @Test
+    fun authenticatedReqCarriesTheObserverToTheStore() =
+        runTest {
+            val dispatcher = UnconfinedTestDispatcher(testScheduler)
+            val store = ContextRecordingStore(EventStore(null))
+            val server = createServer(dispatcher, store) { FullAuthPolicy(relayUrl) }
+            val messages = mutableListOf<String>()
+            val session = server.connect { messages.add(it) }
+
+            // NIP-42 handshake, mirroring NostrServerAuthTest.
+            val challenge = (OptimizedJsonMapper.fromJsonToMessage(messages[0]) as AuthMessage).challenge
+            val auth =
+                RelayAuthEvent(
+                    id = hexId(99),
+                    pubKey = pubkey,
+                    createdAt = TimeUtils.now(),
+                    tags =
+                        arrayOf(
+                            arrayOf("relay", relayUrl.url),
+                            arrayOf("challenge", challenge),
+                        ),
+                    content = "",
+                    sig = sig,
+                )
+            session.receive("""["AUTH",${auth.toJson()}]""")
+            assertTrue(session.requestContext.authenticatedUsers.contains(pubkey))
+
+            session.receive("""["REQ","sub1",{"kinds":[1]}]""")
+
+            assertEquals(1, store.contexts.size, "the REQ must have reached the store")
+            val seen = store.contexts[0]
+            assertEquals(setOf(pubkey), seen?.authenticatedUsers)
+            assertEquals(pubkey, seen?.observer)
+
+            server.close()
+        }
+}

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

@@ -98,6 +98,29 @@ class SearchTest : BaseDBTest() {
             db.assertQuery(comment, Filter(kinds = listOf(CommentEvent.KIND), search = "testing"))
         }
 
+    @Test
+    fun testSearchExtensionTokensAreIgnoredByTheStore() =
+        forEachDB { db ->
+            db.store.insertEvent(comment)
+
+            // The store — not the relay layer — owns the NIP-50 extension
+            // decision now: a raw `include:spam` reaching FTS5 MATCH would
+            // raise "no such column: include", so the store strips the
+            // token at its own boundary and matches on the terms alone.
+            db.assertQuery(comment, Filter(search = "testing include:spam"))
+
+            // Extensions-only search: unsupported extensions are ignored
+            // (NIP-50), so the constraint drops and the query is
+            // unconstrained — not match-nothing, not an FTS error.
+            db.assertQuery(comment, Filter(search = "include:spam language:en"))
+
+            // Delete-by-filter must strip the same way, and an
+            // extensions-only search collapses to an empty filter, which
+            // deletes nothing (safe-by-default contract).
+            db.store.delete(Filter(search = "language:en"))
+            db.assertQuery(comment, Filter(search = "testing"))
+        }
+
     @Test
     fun testFtsCleanedUpAfterReplaceableRotation() =
         forEachDB { db ->

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

@@ -20,9 +20,13 @@
  */
 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.nip01Core.store.FtsReindexProgress
+import com.vitorpamplona.quartz.nip01Core.store.IEventStore
 import com.vitorpamplona.quartz.nip10Notes.TextNoteEvent
+import kotlinx.coroutines.runBlocking
 import kotlin.test.Test
 import kotlin.test.assertEquals
 
@@ -101,4 +105,119 @@ class SnapshotIdsForNegentropyTest : BaseDBTest() {
             val whole = db.snapshotIdsForNegentropy(listOf(filter), maxEntries = 100)
             assertEquals(30, whole.size)
         }
+
+    @Test
+    fun reportsProgressWhileCollecting() =
+        runBlocking {
+            // Single store (not forEachDB): this exercises the row-loop
+            // cadence, which is indexing-strategy independent, and needs
+            // enough rows to cross the reporting interval twice.
+            val db = EventStore(dbName = null)
+            try {
+                val total = IEventStore.NEGENTROPY_PROGRESS_EVERY * 2 + 500
+                db.batchInsert(
+                    List(total) { i ->
+                        Event(
+                            id = i.toString(16).padStart(64, '0'),
+                            pubKey = PUBKEY,
+                            createdAt = 1_700_000_000L + i,
+                            kind = 1,
+                            tags = emptyArray(),
+                            content = "e$i",
+                            sig = SIG,
+                        )
+                    },
+                )
+
+                val ticks = mutableListOf<Int>()
+                val all =
+                    db.snapshotIdsForNegentropy(
+                        listOf(Filter(kinds = listOf(1))),
+                        onProgress = { ticks.add(it) },
+                    )
+                assertEquals(total, all.size)
+                assertEquals(
+                    listOf(IEventStore.NEGENTROPY_PROGRESS_EVERY, IEventStore.NEGENTROPY_PROGRESS_EVERY * 2),
+                    ticks,
+                    "onProgress must tick the running count once per interval",
+                )
+
+                // Omitting the callback stays the zero-cost path.
+                assertEquals(total, db.snapshotIdsForNegentropy(listOf(Filter(kinds = listOf(1)))).size)
+            } finally {
+                db.close()
+            }
+        }
+
+    @Test
+    fun interfaceDefaultReportsProgressForStoresWithoutAnOverride() =
+        runBlocking {
+            // A store that only streams events, so snapshotIdsForNegentropy
+            // resolves to the IEventStore default implementation.
+            val total = IEventStore.NEGENTROPY_PROGRESS_EVERY + 500
+            val store = StreamOnlyStore(total)
+
+            val ticks = mutableListOf<Int>()
+            val all = store.snapshotIdsForNegentropy(listOf(Filter()), onProgress = { ticks.add(it) })
+            assertEquals(total, all.size)
+            assertEquals(listOf(IEventStore.NEGENTROPY_PROGRESS_EVERY), ticks)
+
+            // maxEntries truncation still applies on the default path.
+            val capped = store.snapshotIdsForNegentropy(listOf(Filter()), maxEntries = 10)
+            assertEquals(11, capped.size)
+        }
+
+    private companion object {
+        const val PUBKEY = "46fcbe3065eaf1ae7811465924e48923363ff3f526bd6f73d7c184b16bd8ce4d"
+        const val SIG = "4aa5264965018fa12a326686ad3d3bd8beae3218dcc83689b19ca1e6baeb791531943c15363aa6707c7c0c8b2d601deca1f20c32078b2872d356cdca03b04cce"
+    }
+
+    /** Minimal [IEventStore]: streams [total] synthetic events, nothing else. */
+    private class StreamOnlyStore(
+        private val total: Int,
+    ) : IEventStore {
+        override val relay = null
+
+        @Suppress("UNCHECKED_CAST")
+        override suspend fun <T : Event> query(
+            filters: List<Filter>,
+            onEach: (T) -> Unit,
+        ) {
+            repeat(total) { i ->
+                onEach(Event(i.toString(16).padStart(64, '0'), PUBKEY, 1_700_000_000L + i, 1, emptyArray(), "", SIG) as T)
+            }
+        }
+
+        override suspend fun insert(event: Event) = error("unused")
+
+        override suspend fun transaction(body: IEventStore.ITransaction.() -> Unit) = error("unused")
+
+        override suspend fun <T : Event> query(filter: Filter): List<T> = error("unused")
+
+        override suspend fun <T : Event> query(filters: List<Filter>): List<T> = error("unused")
+
+        override suspend fun <T : Event> query(
+            filter: Filter,
+            onEach: (T) -> Unit,
+        ) = error("unused")
+
+        override suspend fun count(filter: Filter): Int = error("unused")
+
+        override suspend fun count(filters: List<Filter>): Int = error("unused")
+
+        override suspend fun delete(filter: Filter) = error("unused")
+
+        override suspend fun delete(filters: List<Filter>) = error("unused")
+
+        override suspend fun deleteExpiredEvents() = error("unused")
+
+        override suspend fun reindexFullTextSearch() = error("unused")
+
+        override suspend fun reindexFullTextSearch(
+            resumeFrom: String?,
+            batchSize: Int,
+        ): FtsReindexProgress = error("unused")
+
+        override fun close() {}
+    }
 }

quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayObserverTest.kt

@@ -72,6 +72,29 @@ class RelayObserverTest {
 
     // ---- what we measured ---------------------------------------------------
 
+    @Test
+    fun `rtt-open is the transport handshake rather than our own queueing`() {
+        // pingMillis is receivedResponseAtMillis - sentRequestAtMillis: it starts
+        // when the upgrade request goes out, so it excludes time the call spent
+        // queued in the client's dispatcher. Timing the enqueue instead published
+        // our own backlog as the relay's latency — a median of 33.5 SECONDS on a
+        // 16,507-relay fan-out, against a true minimum of 140ms — into the field
+        // aggregators rank relays by.
+        val o = RelayObserver()
+        o.onConnected(client(url), 140, false)
+        assertEquals(140L, o.only().rttOpenMs)
+    }
+
+    @Test
+    fun `a handshake the transport could not time publishes no time`() {
+        val o = RelayObserver()
+        o.onConnected(client(url), 0, false)
+
+        val obs = o.only()
+        assertTrue(obs.reachable, "it opened, and that much is known")
+        assertNull(obs.rttOpenMs, "unmeasurable is not zero")
+    }
+
     @Test
     fun `an opened connection is timed rather than assumed`() {
         val o = RelayObserver()

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

@@ -36,6 +36,7 @@ import com.vitorpamplona.quartz.nip01Core.store.sqlite.TagNameValueHasher
 import com.vitorpamplona.quartz.nip09Deletions.DeletionEvent
 import com.vitorpamplona.quartz.nip40Expiration.expiration
 import com.vitorpamplona.quartz.nip50Search.SearchableEvent
+import com.vitorpamplona.quartz.nip50Search.strippingSearchExtensions
 import com.vitorpamplona.quartz.nip62RequestToVanish.RequestToVanishEvent
 import com.vitorpamplona.quartz.utils.EventFactory
 import com.vitorpamplona.quartz.utils.TimeUtils
@@ -299,15 +300,22 @@ open class FsEventStore(
         filter: Filter,
         onEach: (T) -> Unit,
     ) {
-        val limit = filter.limit ?: Int.MAX_VALUE
+        // Every read path (list/streaming query, count, delete-by-filter,
+        // negentropy snapshot) funnels through here, so this is the one
+        // place this store strips NIP-50 extension tokens: the FTS token
+        // index would otherwise require `include` / `spam` as literal AND
+        // terms. NIP-50 says unsupported extensions are ignored, so an
+        // extensions-only search collapses to an unconstrained query.
+        val stripped = filter.strippingSearchExtensions()
+        val limit = stripped.limit ?: Int.MAX_VALUE
         if (limit <= 0) return
         var emitted = 0
         val seenIds = HashSet<HexKey>()
-        for (candidate in planner.plan(filter)) {
+        for (candidate in planner.plan(stripped)) {
             if (emitted >= limit) break
             if (!seenIds.add(candidate.id)) continue
             val event = readEvent(candidate.id) ?: continue
-            if (!filter.match(event)) continue
+            if (!stripped.match(event)) continue
             @Suppress("UNCHECKED_CAST")
             onEach(event as T)
             emitted++
@@ -348,7 +356,10 @@ open class FsEventStore(
      */
     override suspend fun delete(filter: Filter) =
         lockManager.withWriteLock {
-            if (filter.isEmpty()) return@withWriteLock
+            // The emptiness check runs on the STRIPPED filter: an
+            // extensions-only search (`search = "language:en"`) strips to
+            // an empty filter, which must delete nothing — not everything.
+            if (filter.strippingSearchExtensions().isEmpty()) return@withWriteLock
             val ids = ArrayList<HexKey>()
             query<Event>(filter) { ids.add(it.id) }
             ids.forEach { deleteLocked(it) }
@@ -357,7 +368,7 @@ open class FsEventStore(
     /** See [delete] for the empty-filter contract. */
     override suspend fun delete(filters: List<Filter>) =
         lockManager.withWriteLock {
-            val nonEmpty = filters.filterNot { it.isEmpty() }
+            val nonEmpty = filters.filterNot { it.strippingSearchExtensions().isEmpty() }
             if (nonEmpty.isEmpty()) return@withWriteLock
             val ids = HashSet<HexKey>()
             query<Event>(nonEmpty) { ids.add(it.id) }

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

@@ -359,4 +359,45 @@ class FsSearchTest {
             // And a search no longer finds it.
             assertEquals(emptyList(), store.query<TextNoteEvent>(Filter(search = "zzz")).map { it.id })
         }
+
+    // ------------------------------------------------------------------
+    // NIP-50 extension tokens
+    // ------------------------------------------------------------------
+
+    @Test
+    fun `extension tokens are ignored, not matched as literal AND terms`() =
+        runBlocking {
+            val n = note("bitcoin rocks", ts = 100)
+            store.insert(n)
+
+            // The store owns the extension decision: `include:spam` must be
+            // stripped at the store boundary, not required as the literal
+            // tokens `include` + `spam` (which would match nothing here).
+            assertEquals(
+                listOf(n.id),
+                store.query<TextNoteEvent>(Filter(search = "bitcoin include:spam")).map { it.id },
+            )
+            assertEquals(1, store.count(Filter(search = "bitcoin include:spam")))
+
+            // Extensions-only search collapses to an unconstrained query
+            // (NIP-50: unsupported extensions are ignored, not match-nothing).
+            assertEquals(
+                listOf(n.id),
+                store.query<TextNoteEvent>(Filter(search = "language:en")).map { it.id },
+            )
+        }
+
+    @Test
+    fun `delete with an extensions-only search deletes nothing`() =
+        runBlocking {
+            val n = note("bitcoin rocks", ts = 100)
+            store.insert(n)
+
+            // Stripped, `language:en` is an empty filter — and an empty
+            // filter must fall under delete's safe-by-default contract
+            // instead of matching (and wiping) the whole store.
+            store.delete(Filter(search = "language:en"))
+
+            assertEquals(1, store.count(Filter()))
+        }
 }