mirror of
https://github.com/vitorpamplona/amethyst.git
synced 2026-08-08 23:54:39 +00:00
Merge pull request #3842 from vitorpamplona/claude/liveventstore-searchextensions-3ya4ir
Add StoreQueryContext for observer-relative ranking in stores
This commit is contained in:
Vitor Pamplona
+19
-9
@@ -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:
|
||||
|
||||
+2
-1
@@ -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)
|
||||
|
||||
|
||||
+13
-2
@@ -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, ""))
|
||||
}
|
||||
|
||||
+43
-19
@@ -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
|
||||
|
||||
+4
-1
@@ -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>
|
||||
}
|
||||
|
||||
+53
-1
@@ -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`:
|
||||
*
|
||||
|
||||
+2
-1
@@ -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)
|
||||
|
||||
|
||||
+65
@@ -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()
|
||||
}
|
||||
+2
-1
@@ -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)
|
||||
|
||||
|
||||
+8
-2
@@ -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
|
||||
}
|
||||
|
||||
+26
-15
@@ -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()
|
||||
}
|
||||
}
|
||||
|
||||
+152
@@ -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()
|
||||
}
|
||||
}
|
||||
+23
@@ -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 ->
|
||||
|
||||
+119
@@ -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() {}
|
||||
}
|
||||
}
|
||||
|
||||
+16
-5
@@ -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) }
|
||||
|
||||
@@ -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()))
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user