From 2e90cc24ba14f35a01b7b4613761e1109b02c9ee Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 1 Aug 2026 16:44:31 +0000 Subject: [PATCH 1/2] feat(quartz): move NIP-50 extension handling into the stores; add IEventStore seams MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The IEventStore contract now defines the seams a store implementation needs instead of having the relay layer decide for it: - NIP-50 search extensions: LiveEventStore no longer strips key:value tokens before the store sees them. Filter.search reaches every store verbatim, and each implementation decides which extensions it supports — the store, not the middleware, is the only component that knows whether sort:rank is a directive or noise. The built-in SQLite and filesystem stores strip at their own boundary (their FTS engines would otherwise error on / literally match the tokens), preserving the NIP-50 'ignore unsupported extensions' behavior end to end. Extension-aware stores need no side channel to recover the raw string anymore. Fs delete now checks emptiness on the stripped filter so an extensions-only search cannot wipe the store. - Caller identity: new StoreQueryContext coroutine-context element, installed by LiveEventStore around every REQ/COUNT store call when the connection has NIP-42-authenticated pubkeys. Observer-relative stores (web-of-trust ranking, for-you relevance) read it off the coroutine context; ranking context only, never match-set changes. - Negentropy liveness: snapshotIdsForNegentropy gains an optional onProgress hook so mirrors syncing huge corpora get a running count through the interface type instead of a concrete-class overload. SQLite reports from its row loop; the interface default streams and reports too. - FtsReindexProgress.cursor documented as opaque and store-defined so resumable-reindex callers never assume id semantics. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_011hH4RY2AUwfMZ54RkMiT45 --- quartz/RELAY.md | 28 ++-- .../cache/interning/InterningEventStore.kt | 3 +- .../relay/server/backend/LiveEventStore.kt | 62 ++++--- .../quartz/nip01Core/store/IEventStore.kt | 54 ++++++- .../nip01Core/store/ObservableEventStore.kt | 3 +- .../nip01Core/store/StoreQueryContext.kt | 65 ++++++++ .../nip01Core/store/sqlite/EventStore.kt | 3 +- .../nip01Core/store/sqlite/QueryBuilder.kt | 10 +- .../store/sqlite/SQLiteEventStore.kt | 41 +++-- .../relay/server/StoreQueryContextTest.kt | 152 ++++++++++++++++++ .../nip01Core/store/sqlite/SearchTest.kt | 23 +++ .../sqlite/SnapshotIdsForNegentropyTest.kt | 119 ++++++++++++++ .../quartz/nip01Core/store/fs/FsEventStore.kt | 21 ++- .../quartz/nip01Core/store/fs/FsSearchTest.kt | 41 +++++ 14 files changed, 571 insertions(+), 54 deletions(-) create mode 100644 quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/StoreQueryContext.kt create mode 100644 quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/StoreQueryContextTest.kt diff --git a/quartz/RELAY.md b/quartz/RELAY.md index 01bc5fc736..e5010ed069 100644 --- a/quartz/RELAY.md +++ b/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: diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/cache/interning/InterningEventStore.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/cache/interning/InterningEventStore.kt index 2f87070657..c84544c3cf 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/cache/interning/InterningEventStore.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/cache/interning/InterningEventStore.kt @@ -116,7 +116,8 @@ class InterningEventStore( override suspend fun snapshotIdsForNegentropy( filters: List, maxEntries: Int?, - ): List = inner.snapshotIdsForNegentropy(filters, maxEntries) + onProgress: ((collected: Int) -> Unit)?, + ): List = inner.snapshotIdsForNegentropy(filters, maxEntries, onProgress) override suspend fun delete(filter: Filter) = inner.delete(filter) diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/LiveEventStore.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/LiveEventStore.kt index 9f5f8c4ecc..345a4337a5 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/LiveEventStore.kt +++ b/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 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(filters.strippingSearchExtensions()) { event -> - seen.record(event.id) - onEach(event) + withCallerIdentity(ctx) { + store.query(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, ): 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 = store.query(filter.strippingSearchExtensions()) + suspend fun snapshotQuery(filter: Filter): List = store.query(filter) /** * Multi-filter snapshot. Unions the per-filter results and @@ -333,7 +357,7 @@ class LiveEventStore( val seen = HashSet() val merged = ArrayList() for (f in filters) { - for (e in store.query(f.strippingSearchExtensions())) { + for (e in store.query(f)) { if (seen.add(e.id)) merged += e } } @@ -353,7 +377,7 @@ class LiveEventStore( override suspend fun snapshotIdsForNegentropy( filters: List, maxEntries: Int?, - ): List = store.snapshotIdsForNegentropy(filters.strippingSearchExtensions(), maxEntries) + ): List = store.snapshotIdsForNegentropy(filters, maxEntries) // ------------------------------------------------------------------ // NIP-77 snapshot cache diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/IEventStore.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/IEventStore.kt index 690106500b..8f77817522 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/IEventStore.kt +++ b/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, maxEntries: Int? = null, + onProgress: ((collected: Int) -> Unit)? = null, ): List { - val all = query(filters).map { IdAndTime(it.createdAt, it.id) } + val all = ArrayList() + query(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`: * diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/ObservableEventStore.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/ObservableEventStore.kt index ca0d372d6d..c314c4e3b2 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/ObservableEventStore.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/ObservableEventStore.kt @@ -192,7 +192,8 @@ class ObservableEventStore( override suspend fun snapshotIdsForNegentropy( filters: List, maxEntries: Int?, - ): List = inner.snapshotIdsForNegentropy(filters, maxEntries) + onProgress: ((collected: Int) -> Unit)?, + ): List = inner.snapshotIdsForNegentropy(filters, maxEntries, onProgress) override suspend fun liveNegentropySnapshot(maxEntries: Int) = inner.liveNegentropySnapshot(maxEntries) diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/StoreQueryContext.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/StoreQueryContext.kt new file mode 100644 index 0000000000..37d4524442 --- /dev/null +++ b/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, +) : AbstractCoroutineContextElement(StoreQueryContext) { + companion object Key : CoroutineContext.Key + + /** + * Convenience for the common single-identity case: one of + * [authenticatedUsers], or `null` when the set is empty. + */ + val observer: HexKey? get() = authenticatedUsers.firstOrNull() +} diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/EventStore.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/EventStore.kt index 13545cef2e..50ffdd6e37 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/EventStore.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/EventStore.kt @@ -82,7 +82,8 @@ class EventStore( override suspend fun snapshotIdsForNegentropy( filters: List, maxEntries: Int?, - ): List = store.snapshotIdsForNegentropy(filters, maxEntries) + onProgress: ((collected: Int) -> Unit)?, + ): List = store.snapshotIdsForNegentropy(filters, maxEntries, onProgress) override suspend fun liveNegentropySnapshot(maxEntries: Int) = store.liveNegentropySnapshot(maxEntries) diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/QueryBuilder.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/QueryBuilder.kt index 15c69596f7..4cd73ae9f6 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/QueryBuilder.kt +++ b/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, db: SQLiteConnection, maxEntries: Int? = null, + onProgress: ((collected: Int) -> Unit)? = null, ): List { 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 = + private fun SQLiteConnection.runIdAndTimeQuery( + query: QuerySpec, + onProgress: ((collected: Int) -> Unit)? = null, + ): List = prepare(query.sql).use { stmt -> query.args.forEachIndexed { index, arg -> stmt.bindText(index + 1, arg) @@ -462,6 +467,7 @@ class QueryBuilder( val results = ArrayList() 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 } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/SQLiteEventStore.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/SQLiteEventStore.kt index aee8f0c40f..515ca0b916 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/SQLiteEventStore.kt +++ b/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 query(filter: Filter): List = 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 query(filters: List): List = pool.useReader { queryBuilder.query(filters, it) } + suspend fun query(filter: Filter): List = pool.useReader { queryBuilder.query(filter.strippingSearchExtensions(), it) } + + suspend fun query(filters: List): List = pool.useReader { queryBuilder.query(filters.strippingSearchExtensions(), it) } suspend fun query( filter: Filter, onEach: (T) -> Unit, - ) = pool.useReader { queryBuilder.query(filter, it, onEach) } + ) = pool.useReader { queryBuilder.query(filter.strippingSearchExtensions(), it, onEach) } suspend fun query( filters: List, onEach: (T) -> Unit, - ) = pool.useReader { queryBuilder.query(filters, it, onEach) } + ) = pool.useReader { queryBuilder.query(filters.strippingSearchExtensions(), it, onEach) } - suspend fun rawQuery(filter: Filter): List = pool.useReader { queryBuilder.rawQuery(filter, it) } + suspend fun rawQuery(filter: Filter): List = pool.useReader { queryBuilder.rawQuery(filter.strippingSearchExtensions(), it) } - suspend fun rawQuery(filters: List): List = pool.useReader { queryBuilder.rawQuery(filters, it) } + suspend fun rawQuery(filters: List): List = 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, 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) = pool.useReader { queryBuilder.planQuery(filters, seedModule.hasher(it), it) } + suspend fun planQuery(filters: List) = 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): Int = pool.useReader { queryBuilder.count(filters, it) } + suspend fun count(filters: List): Int = pool.useReader { queryBuilder.count(filters.strippingSearchExtensions(), it) } suspend fun authorsMissingOutbox(): List = pool.useReader { queryBuilder.authorsMissingKind(AdvertisedRelayListEvent.KIND, it) } suspend fun snapshotIdsForNegentropy( filters: List, maxEntries: Int? = null, - ): List = pool.useReader { queryBuilder.snapshotIdsForNegentropy(filters, it, maxEntries) } + onProgress: ((collected: Int) -> Unit)? = null, + ): List = 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) { pool.useWriter { - queryBuilder.delete(filters, it) + queryBuilder.delete(filters.strippingSearchExtensions(), it) liveNegentropyIndex?.invalidate() } } diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/StoreQueryContextTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/StoreQueryContextTest.kt new file mode 100644 index 0000000000..a39cdc5c24 --- /dev/null +++ b/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() + + override suspend fun query( + filters: List, + onEach: (T) -> Unit, + ) { + contexts.add(coroutineContext[StoreQueryContext]) + inner.query(filters, onEach) + } + + override suspend fun rawQuery( + filters: List, + 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() + 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() + } +} diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/SearchTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/SearchTest.kt index 615e6fc47e..689466b310 100644 --- a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/SearchTest.kt +++ b/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 -> diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/SnapshotIdsForNegentropyTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/SnapshotIdsForNegentropyTest.kt index fe1355160f..30d6a29b1a 100644 --- a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/SnapshotIdsForNegentropyTest.kt +++ b/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() + 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() + 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 query( + filters: List, + 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 query(filter: Filter): List = error("unused") + + override suspend fun query(filters: List): List = error("unused") + + override suspend fun query( + filter: Filter, + onEach: (T) -> Unit, + ) = error("unused") + + override suspend fun count(filter: Filter): Int = error("unused") + + override suspend fun count(filters: List): Int = error("unused") + + override suspend fun delete(filter: Filter) = error("unused") + + override suspend fun delete(filters: List) = 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() {} + } } diff --git a/quartz/src/jvmMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/fs/FsEventStore.kt b/quartz/src/jvmMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/fs/FsEventStore.kt index 9062867baf..38cc465830 100644 --- a/quartz/src/jvmMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/fs/FsEventStore.kt +++ b/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() - 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() query(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) = lockManager.withWriteLock { - val nonEmpty = filters.filterNot { it.isEmpty() } + val nonEmpty = filters.filterNot { it.strippingSearchExtensions().isEmpty() } if (nonEmpty.isEmpty()) return@withWriteLock val ids = HashSet() query(nonEmpty) { ids.add(it.id) } diff --git a/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/fs/FsSearchTest.kt b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/fs/FsSearchTest.kt index 1dde055e7f..ccd9400ae9 100644 --- a/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/fs/FsSearchTest.kt +++ b/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(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(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(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())) + } } From f2e7a07a97474b53aad3bea2e3e774319057adf8 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 1 Aug 2026 16:58:36 +0000 Subject: [PATCH 2/2] fix(relay): make the per-connection auth set a copy-on-write snapshot MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Audit finding on the StoreQueryContext seam: RelaySession kept authenticatedUsers as a plain mutable LinkedHashSet and handed out a live view through RequestContext. A store (or EventSource) reading the set during a long REQ replay — exactly what StoreQueryContext invites — could race a concurrent NIP-42 AUTH commit on another coroutine: iteration vs. add on an unsynchronized set is a ConcurrentModificationException or a torn read. The engine now swaps an immutable Set behind a @Volatile field on each AUTH (the only writer), so every read is a consistent snapshot and holding one across a replay is safe. AUTH is rare; the copy is off the hot path. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_011hH4RY2AUwfMZ54RkMiT45 --- .../quartz/nip01Core/relay/server/RelaySession.kt | 15 +++++++++++++-- .../relay/server/backend/RequestContext.kt | 5 ++++- 2 files changed, 17 insertions(+), 3 deletions(-) diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt index 7aa1e208d6..630ddec74a 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt +++ b/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() + @Volatile + private var authenticatedUsers = setOf() /** * 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, "")) } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/RequestContext.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/RequestContext.kt index b492e80017..7948ce61cd 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/RequestContext.kt +++ b/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 }