diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt index a731890e88..9290fef8e9 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt @@ -32,6 +32,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.server.policies.VerifyAuthOnlyPo import com.vitorpamplona.quartz.nip01Core.relay.server.policies.VerifyPolicy import com.vitorpamplona.quartz.nip01Core.store.IEventStore import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore +import com.vitorpamplona.quartz.nip77Negentropy.NegentropySettings import java.io.File /** @@ -109,6 +110,12 @@ fun main(args: Array) { } val stateFile = config.admin.state_file?.let { File(it) } + val negentropySettings = + NegentropySettings( + frameSizeLimit = config.negentropy.frame_size_limit, + maxSyncEvents = config.negentropy.max_sync_events, + maxSessionsPerConnection = config.negentropy.max_sessions_per_connection, + ) val relay = Relay( advertisedUrl, @@ -117,6 +124,7 @@ fun main(args: Array) { policyBuilder, stateFile = stateFile, parallelVerify = parallelVerify, + negentropySettings = negentropySettings, ) // Frame cap honors max_ws_frame_bytes when set; max_ws_message_bytes // is treated as the same cap (Ktor's WebSockets plugin only exposes diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/Relay.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/Relay.kt index 6a8436103b..cccdb70909 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/Relay.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/Relay.kt @@ -33,6 +33,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.server.policies.EmptyPolicy import com.vitorpamplona.quartz.nip01Core.store.IEventStore import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore import com.vitorpamplona.quartz.nip11RelayInfo.Nip11RelayInformation +import com.vitorpamplona.quartz.nip77Negentropy.NegentropySettings import com.vitorpamplona.quartz.nip86RelayManagement.server.BanListPolicy import com.vitorpamplona.quartz.nip86RelayManagement.server.BanStore import kotlinx.coroutines.SupervisorJob @@ -84,6 +85,11 @@ class Relay( * `Main.kt` skips `VerifyPolicy` when this flag is on. */ parallelVerify: Boolean = false, + /** + * NIP-77 server-side tuning (frame cap, snapshot cap, + * per-connection session cap). Defaults to strfry-parity values. + */ + negentropySettings: NegentropySettings = NegentropySettings.Default, ) : AutoCloseable { private val stateStore: RelayStateStore? = stateFile?.let { RelayStateStore(it) } @@ -168,8 +174,9 @@ class Relay( val user = policyBuilder() if (user === EmptyPolicy) BanListPolicy(banStore) else user + BanListPolicy(banStore) }, - parentContext, + parentContext = parentContext, parallelVerify = parallelVerify, + negentropySettings = negentropySettings, ) /** diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/RelayHub.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/RelayHub.kt index 6a6866306e..12d97451a6 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/RelayHub.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/RelayHub.kt @@ -28,6 +28,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.server.policies.EmptyPolicy import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebSocket import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebSocketListener import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebsocketBuilder +import com.vitorpamplona.quartz.nip77Negentropy.NegentropySettings import java.util.concurrent.ConcurrentHashMap /** @@ -50,6 +51,7 @@ import java.util.concurrent.ConcurrentHashMap */ class RelayHub( private val defaultPolicy: () -> IRelayPolicy = { EmptyPolicy }, + private val negentropySettings: NegentropySettings = NegentropySettings.Default, ) : WebsocketBuilder, AutoCloseable { private val relays = ConcurrentHashMap() @@ -60,7 +62,11 @@ class RelayHub( fun getOrCreate(url: NormalizedRelayUrl): Relay { check(!closed) { "RelayHub has been closed" } return relays.getOrPut(url) { - Relay(url = url, policyBuilder = defaultPolicy) + Relay( + url = url, + policyBuilder = defaultPolicy, + negentropySettings = negentropySettings, + ) } } diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/config/RelayConfig.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/config/RelayConfig.kt index 1f4ff44415..b8f162ffbe 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/config/RelayConfig.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/config/RelayConfig.kt @@ -43,6 +43,7 @@ data class RelayConfig( val limits: LimitsSection = LimitsSection(), val authorization: AuthorizationSection = AuthorizationSection(), val admin: AdminSection = AdminSection(), + val negentropy: NegentropySection = NegentropySection(), ) { /** * Maps the `[info]` section into a [RelayInfo] used by the NIP-11 @@ -160,6 +161,34 @@ data class RelayConfig( val max_ws_frame_bytes: Int? = null, ) + /** + * NIP-77 negentropy tuning. Defaults track strfry + * (`hoytech/strfry`) so a Geode relay accepts the same workload + * shape and exchanges the same NEG-MSG round-trip size as + * strfry — the de-facto reference implementation. + * + * - [frame_size_limit] mirrors strfry's hard-coded + * `Negentropy ne(storage, 500'000)` in `RelayNegentropy.cpp`. + * Hex-encoded that's ~1 MB on the wire per NEG-MSG; ensure + * `[limits].max_ws_frame_bytes` (when set) is at least double + * this or NEG-MSGs get truncated by the WS layer. + * - [max_sync_events] mirrors strfry's + * `relay__negentropy__maxSyncEvents`. NEG-OPEN whose snapshot + * exceeds this returns + * `["NEG-ERR", "", "blocked: too many query results"]`. + * - [max_sessions_per_connection] caps concurrent NEG-OPEN + * sessions held by a single connection. strfry shares its + * 200-cap with REQ subs via `relay__maxSubsPerConnection`; + * Geode counts NEG independently for now (REQ has no cap yet). + * Overflow returns NOTICE + * `"too many concurrent NEG requests"` (matches strfry). + */ + data class NegentropySection( + val frame_size_limit: Long = 500_000L, + val max_sync_events: Int = 1_000_000, + val max_sessions_per_connection: Int = 200, + ) + data class AuthorizationSection( val pubkey_whitelist: List = emptyList(), val pubkey_blacklist: List = emptyList(), diff --git a/geode/src/test/kotlin/com/vitorpamplona/geode/Nip77NegentropyTest.kt b/geode/src/test/kotlin/com/vitorpamplona/geode/Nip77NegentropyTest.kt index d399457b02..a205d8759d 100644 --- a/geode/src/test/kotlin/com/vitorpamplona/geode/Nip77NegentropyTest.kt +++ b/geode/src/test/kotlin/com/vitorpamplona/geode/Nip77NegentropyTest.kt @@ -24,6 +24,7 @@ import com.vitorpamplona.quartz.nip01Core.core.Event import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message +import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.NoticeMessage import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer @@ -33,6 +34,7 @@ import com.vitorpamplona.quartz.nip10Notes.TextNoteEvent import com.vitorpamplona.quartz.nip77Negentropy.NegErrMessage import com.vitorpamplona.quartz.nip77Negentropy.NegMsgMessage import com.vitorpamplona.quartz.nip77Negentropy.NegentropySession +import com.vitorpamplona.quartz.nip77Negentropy.NegentropySettings import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED import kotlinx.coroutines.runBlocking @@ -237,12 +239,82 @@ class Nip77NegentropyTest { val response = client.nextMessage() assertTrue(response is NegErrMessage, "expected NEG-ERR, got ${response::class.simpleName}") assertEquals("ghost-sub", response.subId) - assertTrue(response.reason.contains("no negentropy session")) + // strfry-parity wording — clients in the wild string-match this. + assertEquals("closed: unknown subscription handle", response.reason) } finally { client.close() } } + @Test + fun negOpenSnapshotOverflowReturnsStrFryNegErr() = + runBlocking { + // Tiny cap so the test is fast. Preload more events than the + // cap so NEG-OPEN must reject — strfry's parity behaviour for + // `relay__negentropy__maxSyncEvents`. + val capped = RelayHub(negentropySettings = NegentropySettings(maxSyncEvents = 5)) + try { + val capUrl = RelayUrlNormalizer.normalize("ws://127.0.0.1:7771/") + val events = makeEvents(20) + capped.getOrCreate(capUrl).preload(events) + + val client = WireClient(capped, capUrl) + try { + val session = + NegentropySession( + subId = "neg-overflow", + filter = Filter(kinds = listOf(1)), + localEvents = emptyList(), + ) + client.send(OptimizedJsonMapper.toJson(session.open())) + + val response = client.nextMessage() + assertTrue(response is NegErrMessage, "expected NEG-ERR, got ${response::class.simpleName}") + assertEquals("neg-overflow", response.subId) + // strfry-parity wording. + assertEquals("blocked: too many query results", response.reason) + } finally { + client.close() + } + } finally { + capped.close() + } + } + + @Test + fun negOpenPerConnectionCapEmitsNotice() = + runBlocking { + // Cap = 2, so the third NEG-OPEN on one connection should + // be rejected with a NOTICE (matching strfry's wording). + val capped = RelayHub(negentropySettings = NegentropySettings(maxSessionsPerConnection = 2)) + try { + val capUrl = RelayUrlNormalizer.normalize("ws://127.0.0.1:7772/") + capped.getOrCreate(capUrl).preload(makeEvents(3)) + + val client = WireClient(capped, capUrl) + try { + repeat(2) { i -> + val s = NegentropySession("ok-$i", Filter(kinds = listOf(1)), localEvents = emptyList()) + client.send(OptimizedJsonMapper.toJson(s.open())) + // Drain the NEG-MSG response so the next OPEN goes + // through cleanly. + client.nextMessage() as NegMsgMessage + } + + // Third OPEN — should be rejected with a NOTICE. + val third = NegentropySession("third", Filter(kinds = listOf(1)), localEvents = emptyList()) + client.send(OptimizedJsonMapper.toJson(third.open())) + val response = client.nextMessage() + assertTrue(response is NoticeMessage, "expected NOTICE, got ${response::class.simpleName}") + assertEquals("too many concurrent NEG requests", response.message) + } finally { + client.close() + } + } finally { + capped.close() + } + } + @Test fun negOpenWithSameSubIdReplacesPriorSession() = runBlocking { 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 d224a17d9a..85fbc1c6fa 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 @@ -24,6 +24,7 @@ import com.vitorpamplona.quartz.nip01Core.core.Event import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl import com.vitorpamplona.quartz.nip01Core.store.IEventStore +import com.vitorpamplona.quartz.nip01Core.store.IdAndTime /** * Decorator that canonicalises every [Event] returned by the inner @@ -111,6 +112,11 @@ class InterningEventStore( override suspend fun count(filters: List): Int = inner.count(filters) + override suspend fun snapshotIdsForNegentropy( + filters: List, + maxEntries: Int?, + ): List = inner.snapshotIdsForNegentropy(filters, maxEntries) + override suspend fun delete(filter: Filter) = inner.delete(filter) override suspend fun delete(filters: List) = inner.delete(filters) diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/LiveEventStore.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/LiveEventStore.kt index 912c1e4d04..6264fc7a82 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/LiveEventStore.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/LiveEventStore.kt @@ -23,6 +23,7 @@ package com.vitorpamplona.quartz.nip01Core.relay.server import com.vitorpamplona.quartz.nip01Core.core.Event import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.store.IEventStore +import com.vitorpamplona.quartz.nip01Core.store.IdAndTime import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.channels.BufferOverflow import kotlinx.coroutines.flow.MutableSharedFlow @@ -161,4 +162,19 @@ class LiveEventStore( } return merged } + + /** + * Lightweight snapshot for NIP-77 negentropy. Returns + * `(created_at, id)` pairs only — no Event materialisation — + * matching strfry's `MemoryView` footprint of ~40 B/entry. + * + * If [maxEntries] is non-null, the underlying store may return + * up to `maxEntries + 1` entries; the +1 sentinel lets the + * caller distinguish "exactly at cap" from "exceeds cap" without + * scanning past the cap. + */ + suspend fun snapshotIdsForNegentropy( + filters: List, + maxEntries: Int? = null, + ): List = store.snapshotIdsForNegentropy(filters, maxEntries) } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NegSessionRegistry.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NegSessionRegistry.kt index f85b723d0c..628895fc3d 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NegSessionRegistry.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NegSessionRegistry.kt @@ -21,12 +21,14 @@ package com.vitorpamplona.quartz.nip01Core.relay.server import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message +import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.NoticeMessage import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.ReqCmd import com.vitorpamplona.quartz.nip77Negentropy.NegCloseCmd import com.vitorpamplona.quartz.nip77Negentropy.NegErrMessage import com.vitorpamplona.quartz.nip77Negentropy.NegMsgCmd import com.vitorpamplona.quartz.nip77Negentropy.NegOpenCmd import com.vitorpamplona.quartz.nip77Negentropy.NegentropyServerSession +import com.vitorpamplona.quartz.nip77Negentropy.NegentropySettings /** * Per-connection NIP-77 negentropy state and dispatch. @@ -39,21 +41,37 @@ import com.vitorpamplona.quartz.nip77Negentropy.NegentropyServerSession * Plain [HashMap] is sufficient because the registry is mutated only * from [RelaySession.receive] — that path is single-threaded per the * WebSocket handler contract. + * + * Defaults match strfry (`hoytech/strfry`) so a Geode relay reconciles + * with the same round-trip shape and the same operator-visible + * protections — see [NegentropySettings]. */ class NegSessionRegistry( private val store: LiveEventStore, private val send: (Message) -> Unit, + private val settings: NegentropySettings = NegentropySettings.Default, ) { private val sessions = HashMap() /** - * Open a reconciliation session. The relay snapshots its matching - * events at this instant — concurrent inserts during the sync are - * not surfaced; clients re-open if they want fresh state. + * Open a reconciliation session. The relay snapshots the matching + * `(created_at, id)` pairs at this instant — concurrent inserts + * during the sync are not surfaced; clients re-open if they want + * fresh state. * * Access control reuses the REQ policy hook: a relay that requires * AUTH or has kind/pubkey allow-deny lists applies the same rules * to NEG-OPEN as it does to subscription REQs. + * + * Two strfry-parity protections fire here: + * - **Per-connection session cap.** If an OPEN would push the + * map past [NegentropySettings.maxSessionsPerConnection], we + * send a NOTICE (matching strfry's + * `"too many concurrent NEG requests"`) and drop the OPEN. + * - **Snapshot size cap.** The store is asked for at most + * `maxSyncEvents + 1` entries; if the +1 sentinel comes back, + * the corpus exceeds the cap and we send NEG-ERR + * `"blocked: too many query results"` (matching strfry). */ suspend fun open( cmd: NegOpenCmd, @@ -66,20 +84,43 @@ class NegSessionRegistry( } val filters = (gate as PolicyResult.Accepted).cmd.filters + // Per-connection cap. Only fires when this is a NEW subId — + // a same-subId re-open replaces the prior session 1-for-1. + val isReopen = sessions.containsKey(cmd.subId) + if (!isReopen && sessions.size >= settings.maxSessionsPerConnection) { + send(NoticeMessage("too many concurrent NEG requests")) + return + } + // NIP-77: same-subId OPEN replaces any prior session. sessions.remove(cmd.subId) - val events = store.snapshotQuery(filters) - val session = NegentropyServerSession(cmd.subId, events) + val cap = settings.maxSyncEvents + val entries = store.snapshotIdsForNegentropy(filters, maxEntries = cap) + if (entries.size > cap) { + send(NegErrMessage(cmd.subId, "blocked: too many query results")) + return + } + + val session = + NegentropyServerSession( + subId = cmd.subId, + localEntries = entries, + frameSizeLimit = settings.frameSizeLimit, + ) sessions[cmd.subId] = session runMessage(cmd.subId, session) { it.processMessage(cmd.initialMessage) } } + /** + * Process a follow-up NEG-MSG. strfry-parity wording for the + * unknown-subId case: `"closed: unknown subscription handle"`. + */ fun msg(cmd: NegMsgCmd) { val session = sessions[cmd.subId] if (session == null) { - send(NegErrMessage(cmd.subId, "error: no negentropy session for ${cmd.subId}")) + send(NegErrMessage(cmd.subId, "closed: unknown subscription handle")) return } runMessage(cmd.subId, session) { it.processMessage(cmd.message) } @@ -99,6 +140,9 @@ class NegSessionRegistry( sessions.clear() } + /** Test/diagnostic accessor. */ + val activeSessionCount: Int get() = sessions.size + private inline fun runMessage( subId: String, session: NegentropyServerSession, @@ -107,9 +151,11 @@ class NegSessionRegistry( try { val response = block(session) if (response != null) send(response) - } catch (e: Exception) { + } catch (_: Exception) { + // strfry sends `PROTOCOL-ERROR` on library reconcile() + // parse failure and tears the session down. sessions.remove(subId) - send(NegErrMessage(subId, "error: ${e.message ?: e::class.simpleName}")) + send(NegErrMessage(subId, "PROTOCOL-ERROR")) } } } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServer.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServer.kt index 0ec8e602be..422d4b1ee2 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServer.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServer.kt @@ -23,6 +23,7 @@ package com.vitorpamplona.quartz.nip01Core.relay.server import com.vitorpamplona.quartz.nip01Core.crypto.verify import com.vitorpamplona.quartz.nip01Core.relay.server.policies.VerifyPolicy import com.vitorpamplona.quartz.nip01Core.store.IEventStore +import com.vitorpamplona.quartz.nip77Negentropy.NegentropySettings import com.vitorpamplona.quartz.utils.cache.LargeCache import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.SupervisorJob @@ -43,12 +44,16 @@ import kotlin.coroutines.CoroutineContext * coroutine inside [VerifyPolicy]. Callers that flip this on should * *omit* `VerifyPolicy` from their [policyBuilder] chain to avoid * double-verifying. + * @param negentropySettings NIP-77 server-side tuning (frame cap, + * snapshot cap, per-connection session cap). Defaults to strfry- + * parity values; see [NegentropySettings]. */ class NostrServer( private val store: IEventStore, private val policyBuilder: () -> IRelayPolicy = { VerifyPolicy }, private val parentContext: CoroutineContext = SupervisorJob(), parallelVerify: Boolean = false, + private val negentropySettings: NegentropySettings = NegentropySettings.Default, ) : AutoCloseable { /** Scope for all subscriptions. */ private val scope = CoroutineScope(parentContext + SupervisorJob()) @@ -87,6 +92,7 @@ class NostrServer( onClose = { session -> connections.remove(session.hashCode()) }, + negentropySettings = negentropySettings, ).also { session -> connections.put(session.hashCode(), session) } 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 c1698492a3..7a4d2d973d 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 @@ -38,6 +38,7 @@ import com.vitorpamplona.quartz.nip01Core.store.IEventStore import com.vitorpamplona.quartz.nip77Negentropy.NegCloseCmd import com.vitorpamplona.quartz.nip77Negentropy.NegMsgCmd import com.vitorpamplona.quartz.nip77Negentropy.NegOpenCmd +import com.vitorpamplona.quartz.nip77Negentropy.NegentropySettings import com.vitorpamplona.quartz.utils.Log import com.vitorpamplona.quartz.utils.cache.LargeCache import kotlinx.coroutines.CancellationException @@ -56,11 +57,12 @@ class RelaySession( private val scope: CoroutineScope, private val onSend: (String) -> Unit, private val onClose: (RelaySession) -> Unit, + negentropySettings: NegentropySettings = NegentropySettings.Default, ) : AutoCloseable { private val subscriptions = LargeCache() /** NIP-77 negentropy state for this connection. */ - private val negentropy = NegSessionRegistry(store, ::send) + private val negentropy = NegSessionRegistry(store, ::send, negentropySettings) private fun addSubscription( subId: String, 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 8984e2fc5c..3300185a43 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 @@ -95,6 +95,38 @@ interface IEventStore : AutoCloseable { suspend fun count(filters: List): Int + /** + * NIP-77 negentropy snapshot. Returns `(created_at, id)` pairs + * for every event matching [filters], with no content/tags/sig + * decode. Used by the server-side reconciliation path to build a + * `StorageVector` without materialising full [Event] objects — + * ~40 B/entry instead of ~1 KB/entry. Order is unspecified; + * negentropy's `seal()` re-sorts. + * + * If [maxEntries] is non-null, the implementation may return up + * to `maxEntries + 1` entries; the caller compares the result + * size to detect overflow (matching strfry's `maxSyncEvents` + * guard). The +1 sentinel lets the caller distinguish "exactly + * capped" from "too many to fit". + * + * 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` + * index. Honors the same filter semantics as [query] including + * any `limit`. + */ + suspend fun snapshotIdsForNegentropy( + filters: List, + maxEntries: Int? = null, + ): List { + val all = query(filters).map { IdAndTime(it.createdAt, it.id) } + return if (maxEntries != null && all.size > maxEntries + 1) { + all.subList(0, maxEntries + 1) + } else { + all + } + } + suspend fun delete(filter: Filter) suspend fun delete(filters: List) diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/IdAndTime.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/IdAndTime.kt new file mode 100644 index 0000000000..0533dffd99 --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/IdAndTime.kt @@ -0,0 +1,39 @@ +/* + * 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 + +/** + * Lightweight projection of an event used by NIP-77 negentropy: just + * the two fields the reconciliation library indexes — `created_at` + * and the 32-byte event id. + * + * Returned by [IEventStore.snapshotIdsForNegentropy] so the relay can + * build a [com.vitorpamplona.negentropy.storage.StorageVector] without + * materialising full [com.vitorpamplona.quartz.nip01Core.core.Event] + * objects (content, tags, sig). For a 1 M-event snapshot this drops + * peak heap from ~1 GB to ~40 MB — strfry's `MemoryView` parity. + */ +data class IdAndTime( + val createdAt: Long, + val id: HexKey, +) 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 cb18c13a96..b11d1b27e4 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 @@ -189,6 +189,11 @@ class ObservableEventStore( override suspend fun count(filters: List): Int = inner.count(filters) + override suspend fun snapshotIdsForNegentropy( + filters: List, + maxEntries: Int?, + ): List = inner.snapshotIdsForNegentropy(filters, maxEntries) + override suspend fun delete(filter: Filter) { inner.delete(filter) _changes.emit(StoreChange.DeleteByFilter(listOf(filter))) 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 097c5672ac..faad3204d3 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 @@ -26,6 +26,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl import com.vitorpamplona.quartz.nip01Core.relay.normalizer.normalizeRelayUrl import com.vitorpamplona.quartz.nip01Core.store.IEventStore +import com.vitorpamplona.quartz.nip01Core.store.IdAndTime /** * SQLite-backed [IEventStore] with default DB-file name and relay @@ -63,6 +64,11 @@ class EventStore( override suspend fun count(filters: List) = store.count(filters) + override suspend fun snapshotIdsForNegentropy( + filters: List, + maxEntries: Int?, + ): List = store.snapshotIdsForNegentropy(filters, maxEntries) + override suspend fun delete(filter: Filter) { store.delete(filter) } 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 6a3412e522..46fa1579a4 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.IdAndTime import com.vitorpamplona.quartz.nip01Core.store.sqlite.sql.where import com.vitorpamplona.quartz.utils.EventFactory @@ -172,6 +173,222 @@ class QueryBuilder( } } + // ----------------------------------------------------------------- + // NIP-77 negentropy snapshot path + // + // Projects only (id, created_at) — no content/tags/sig decode — + // so the relay can build a StorageVector without materialising + // full Event objects. ~40 B/entry instead of ~1 KB/entry. + // No ORDER BY: negentropy's seal() re-sorts. No limit injection: + // the per-session cap is enforced upstream as a count check. + // ----------------------------------------------------------------- + fun snapshotIdsForNegentropy( + filters: List, + db: SQLiteConnection, + maxEntries: Int? = null, + ): List { + val inner = + if (filters.size == 1) { + toSnapshotIdsSql(filters.first(), hasher(db)) + } else { + toSnapshotIdsSql(filters, hasher(db)) + } + // Safety cap: wrap with `LIMIT maxEntries + 1` so we can + // detect overflow without scanning beyond the cap. The +1 + // sentinel lets the caller distinguish "exactly capped" from + // "too many to fit". Matches strfry's `maxSyncEvents` guard. + val query = + if (maxEntries != null) { + QuerySpec( + "SELECT id, created_at FROM (${inner.sql}) LIMIT ${maxEntries + 1}", + inner.args, + ) + } else { + inner + } + return db.runIdAndTimeQuery(query) + } + + private fun toSnapshotIdsSql( + filter: Filter, + hasher: TagNameValueHasher, + ): QuerySpec { + val newFilter = filter.toFilterWithDTags() + + // Simple path — no tag joins, no FTS — collapses to a single + // SELECT against event_headers. + if (newFilter.isSimpleQuery()) { + return makeSimpleIdsQuery( + ids = newFilter.ids, + authors = newFilter.authors, + kinds = newFilter.kinds, + dTags = newFilter.dTags, + since = newFilter.since, + until = newFilter.until, + limit = newFilter.limit, + ) + } + + // Search path — FTS join. Project id+created_at off + // event_headers via the FTS row_id linkage. + if (newFilter.isSimpleSearch()) { + return makeSimpleIdsSearch( + search = newFilter.search!!, + ids = newFilter.ids, + authors = newFilter.authors, + kinds = newFilter.kinds, + dTags = newFilter.dTags, + since = newFilter.since, + until = newFilter.until, + limit = newFilter.limit, + ) + } + + // Tag-join path — reuse the existing row_id subquery and + // join back to event_headers for the projection. + val rowIdSubquery = prepareRowIDSubQueries(filter, hasher) + return if (rowIdSubquery == null) { + QuerySpec("SELECT id, created_at FROM event_headers") + } else { + QuerySpec( + """ + SELECT event_headers.id, event_headers.created_at FROM event_headers + INNER JOIN ( + ${rowIdSubquery.sql} + ) AS filtered + ON event_headers.row_id = filtered.row_id + """.trimIndent(), + rowIdSubquery.args, + ) + } + } + + private fun toSnapshotIdsSql( + filters: List, + hasher: TagNameValueHasher, + ): QuerySpec { + val rowIdSubqueries = unionSubqueriesIfNeeded(filters, hasher) + return if (rowIdSubqueries == null) { + QuerySpec("SELECT id, created_at FROM event_headers") + } else { + QuerySpec( + """ + SELECT DISTINCT event_headers.id, event_headers.created_at FROM event_headers + INNER JOIN ( + ${rowIdSubqueries.sql} + ) AS filtered + ON event_headers.row_id = filtered.row_id + """.trimIndent(), + rowIdSubqueries.args, + ) + } + } + + private fun makeSimpleIdsQuery( + ids: List? = null, + authors: List? = null, + kinds: List? = null, + dTags: List? = null, + since: Long? = null, + until: Long? = null, + limit: Int? = null, + ): QuerySpec { + val clause = + where { + ids?.let { equalsOrIn("id", it) } + kinds?.let { equalsOrIn("kind", it) } + authors?.let { equalsOrIn("pubkey", it) } + dTags?.let { equalsOrIn("d_tag", it) } + since?.let { greaterThanOrEquals("created_at", it) } + until?.let { lessThanOrEquals("created_at", it) } + if (dTags != null && kinds != null) { + if (kinds.all { it.isAddressable() }) { + raw("(kind >= 30000 AND kind < 40000)") + } + } + } + + val sql = + buildString { + append("SELECT id, created_at FROM event_headers") + if (clause.conditions.isNotEmpty()) { + append("\nWHERE ") + append(clause.conditions) + } + // Negentropy honors filter `limit` like REQ does + // (matches strfry's NostrFilterGroup behaviour). + // ORDER BY is required for LIMIT to be meaningful. + if (limit != null) { + append("\nORDER BY created_at DESC") + if (indexStrategy.useAndIndexIdOnOrderBy) { + append(", id ASC") + } + append("\nLIMIT ") + append(limit) + } + } + + return QuerySpec(sql, clause.args) + } + + private fun makeSimpleIdsSearch( + search: String, + ids: List? = null, + authors: List? = null, + kinds: List? = null, + dTags: List? = null, + since: Long? = null, + until: Long? = null, + limit: Int? = null, + ): QuerySpec { + val clause = + where { + ids?.let { equalsOrIn("event_headers.id", it) } + match(fts.tableName, search) + kinds?.let { equalsOrIn("event_headers.kind", it) } + authors?.let { equalsOrIn("event_headers.pubkey", it) } + dTags?.let { equalsOrIn("event_headers.d_tag", it) } + since?.let { greaterThanOrEquals("event_headers.created_at", it) } + until?.let { lessThanOrEquals("event_headers.created_at", it) } + if (dTags != null && kinds != null) { + if (kinds.all { it.isAddressable() }) { + raw("(event_headers.kind >= 30000 AND kind < 40000)") + } + } + } + + val sql = + buildString { + append("SELECT event_headers.id, event_headers.created_at FROM event_headers") + append("\nINNER JOIN ${fts.tableName} ON event_headers.row_id = ${fts.tableName}.${fts.eventHeaderRowIdName}") + if (clause.conditions.isNotEmpty()) { + append("\nWHERE ${clause.conditions}") + } + if (limit != null) { + append("\nORDER BY event_headers.created_at DESC") + if (indexStrategy.useAndIndexIdOnOrderBy) { + append(", event_headers.id ASC") + } + append("\nLIMIT ") + append(limit) + } + } + + return QuerySpec(sql, clause.args) + } + + private fun SQLiteConnection.runIdAndTimeQuery(query: QuerySpec): List = + prepare(query.sql).use { stmt -> + query.args.forEachIndexed { index, arg -> + stmt.bindText(index + 1, arg) + } + val results = ArrayList() + while (stmt.step()) { + results.add(IdAndTime(stmt.getLong(1), stmt.getText(0))) + } + results + } + private fun makeEverythingQuery() = "SELECT id, pubkey, created_at, kind, tags, content, sig FROM event_headers ORDER BY created_at DESC${if (indexStrategy.useAndIndexIdOnOrderBy) ", id ASC" else ""}" private fun makeQueryIn(rowIdQuery: String) = 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 78b29bffdc..ddd98ef7b1 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 @@ -32,6 +32,7 @@ import com.vitorpamplona.quartz.nip01Core.core.isEphemeral import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl import com.vitorpamplona.quartz.nip01Core.store.IEventStore +import com.vitorpamplona.quartz.nip01Core.store.IdAndTime import com.vitorpamplona.quartz.nip40Expiration.isExpired import com.vitorpamplona.quartz.utils.EventFactory @@ -315,6 +316,11 @@ class SQLiteEventStore( suspend fun count(filters: List): Int = pool.useReader { queryBuilder.count(filters, it) } + suspend fun snapshotIdsForNegentropy( + filters: List, + maxEntries: Int? = null, + ): List = pool.useReader { queryBuilder.snapshotIdsForNegentropy(filters, it, maxEntries) } + suspend fun delete(filter: Filter) = pool.useWriter { queryBuilder.delete(filter, it) } suspend fun delete(filters: List) = pool.useWriter { queryBuilder.delete(filters, it) } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip77Negentropy/NegentropyServerSession.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip77Negentropy/NegentropyServerSession.kt index 4fe221d002..62cbb65634 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip77Negentropy/NegentropyServerSession.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip77Negentropy/NegentropyServerSession.kt @@ -23,6 +23,7 @@ package com.vitorpamplona.quartz.nip77Negentropy import com.vitorpamplona.negentropy.Negentropy import com.vitorpamplona.negentropy.storage.StorageVector import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.store.IdAndTime import com.vitorpamplona.quartz.utils.Hex /** @@ -31,28 +32,65 @@ import com.vitorpamplona.quartz.utils.Hex * Used when acting as a relay (or relay-relay sync) to respond to * incoming NEG-OPEN and NEG-MSG from a client. * + * The constructor takes [IdAndTime] entries (just `created_at` and the + * 32-byte event id) to keep the per-session footprint at ~40 B/entry — + * matching strfry's `MemoryView` path. A [List] overload is + * kept for callers (and tests) that already hold full events. + * * Usage: - * 1. On NEG-OPEN: create a [NegentropyServerSession] with the matching local events + * 1. On NEG-OPEN: create a [NegentropyServerSession] with the matching local entries * 2. Call [processMessage] with the initial hex message from NEG-OPEN * 3. Send back the resulting [NegMsgMessage] * 4. On subsequent NEG-MSG: call [processMessage] again and send the response + * + * @param frameSizeLimit max bytes per NEG-MSG response (raw payload, + * before hex). Default `500_000` matches strfry's hard-coded + * `Negentropy ne(storage, 500'000)` so a single round-trip carries + * the same payload as strfry's reconciliation. */ class NegentropyServerSession( val subId: String, - localEvents: List, - frameSizeLimit: Long = 0, + localEntries: List, + frameSizeLimit: Long = DEFAULT_FRAME_SIZE_LIMIT, ) { private val storage = StorageVector() private val negentropy: Negentropy init { - for (event in localEvents) { - storage.insert(event.createdAt, event.id) + for (entry in localEntries) { + storage.insert(entry.createdAt, entry.id) } storage.seal() negentropy = Negentropy(storage, frameSizeLimit) } + companion object { + /** + * strfry parity: `Negentropy ne(storage, 500'000)` in + * `RelayNegentropy.cpp`. Hex-encoded that's ~1 MB on the wire + * per NEG-MSG, the de-facto sync round-trip size. + */ + const val DEFAULT_FRAME_SIZE_LIMIT: Long = 500_000L + + /** + * Convenience for callers that hold full [Event] objects + * (mostly tests + relay-relay sync paths). Production server + * code should call the [IdAndTime] constructor directly via + * `IEventStore.snapshotIdsForNegentropy` to avoid the full + * Event materialisation that this projection collapses. + */ + fun fromEvents( + subId: String, + localEvents: List, + frameSizeLimit: Long = DEFAULT_FRAME_SIZE_LIMIT, + ): NegentropyServerSession = + NegentropyServerSession( + subId = subId, + localEntries = localEvents.map { IdAndTime(it.createdAt, it.id) }, + frameSizeLimit = frameSizeLimit, + ) + } + fun processMessage(hexMessage: String): NegMsgMessage? { val msgBytes = Hex.decode(hexMessage) val result = negentropy.reconcile(msgBytes) diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip77Negentropy/NegentropySettings.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip77Negentropy/NegentropySettings.kt new file mode 100644 index 0000000000..896ec7c472 --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip77Negentropy/NegentropySettings.kt @@ -0,0 +1,50 @@ +/* + * 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.nip77Negentropy + +/** + * Server-side NIP-77 tuning. Defaults track strfry + * (`hoytech/strfry`) so a Quartz-based relay accepts the same + * workload shape and exchanges the same NEG-MSG round-trip size. + * + * @param frameSizeLimit Max bytes per NEG-MSG response payload + * (raw, before hex). 500_000 matches strfry's hard-coded + * `Negentropy ne(storage, 500'000)` in `RelayNegentropy.cpp`. + * The `kmp-negentropy` library enforces `>= 4096` (or `0` for + * unlimited). + * @param maxSyncEvents Hard cap on the snapshot size for a single + * NEG-OPEN. Mirrors strfry's `relay__negentropy__maxSyncEvents`. + * Overflow returns NEG-ERR `"blocked: too many query results"`. + * @param maxSessionsPerConnection Cap on concurrent NEG sessions + * held by one connection. strfry shares 200 with REQ subs; we + * count NEG independently. Overflow sends NOTICE + * `"too many concurrent NEG requests"`. + */ +data class NegentropySettings( + val frameSizeLimit: Long = NegentropyServerSession.DEFAULT_FRAME_SIZE_LIMIT, + val maxSyncEvents: Int = 1_000_000, + val maxSessionsPerConnection: Int = 200, +) { + companion object { + /** strfry-equivalent defaults. */ + val Default = NegentropySettings() + } +} 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 new file mode 100644 index 0000000000..d2cb5e134b --- /dev/null +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/SnapshotIdsForNegentropyTest.kt @@ -0,0 +1,103 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.quartz.nip01Core.store.sqlite + +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerSync +import com.vitorpamplona.quartz.nip10Notes.TextNoteEvent +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertTrue + +/** + * Verifies the NIP-77 negentropy id-and-time projection against the + * full-event query path. Goal: same result set, ~25× lighter + * footprint per row. Run across every indexing strategy via + * [BaseDBTest.forEachDB] so plan changes don't silently break the + * snapshot path. + */ +class SnapshotIdsForNegentropyTest : BaseDBTest() { + private val signer = NostrSignerSync() + + private fun makeEvents(count: Int) = + List(count) { i -> + signer.sign(TextNoteEvent.build("event-$i", createdAt = 1_700_000_000L + i)) + } + + @Test + fun matchesFullQueryForSimpleKindFilter() = + forEachDB { db -> + val events = makeEvents(50) + for (e in events) db.insert(e) + + val filter = Filter(kinds = listOf(1)) + val full = db.query(filter) + val ids = db.snapshotIdsForNegentropy(listOf(filter)) + + assertEquals(full.size, ids.size, "snapshot must cover the same row set") + assertEquals( + full.map { it.id }.toSet(), + ids.map { it.id }.toSet(), + "snapshot ids must match the full-query ids", + ) + // Every (createdAt, id) pair must round-trip. + val byId = full.associate { it.id to it.createdAt } + for (entry in ids) { + assertEquals(byId[entry.id], entry.createdAt, "createdAt mismatch for ${entry.id}") + } + } + + @Test + fun honorsSinceUntilLimit() = + forEachDB { db -> + val events = makeEvents(20) // createdAt 1_700_000_000..1_700_000_019 + for (e in events) db.insert(e) + + // since/until window: [+5, +14] inclusive + val filter = + Filter( + kinds = listOf(1), + since = 1_700_000_005L, + until = 1_700_000_014L, + ) + val ids = db.snapshotIdsForNegentropy(listOf(filter)) + assertEquals(10, ids.size, "since/until window should yield 10 rows") + } + + @Test + fun maxEntriesPlusOneSentinelMarksOverflow() = + forEachDB { db -> + val events = makeEvents(30) + for (e in events) db.insert(e) + + val filter = Filter(kinds = listOf(1)) + // cap = 10; we have 30 rows, so the result must be 11 + // (cap + 1 sentinel) — matches strfry's `maxSyncEvents` + // overflow-detection idiom. + val capped = db.snapshotIdsForNegentropy(listOf(filter), maxEntries = 10) + assertEquals(11, capped.size) + assertTrue(capped.size > 10, "caller relies on size > cap as overflow signal") + + // cap >= total: returns the whole set unchanged. + val whole = db.snapshotIdsForNegentropy(listOf(filter), maxEntries = 100) + assertEquals(30, whole.size) + } +} diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip77Negentropy/NegentropySessionTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip77Negentropy/NegentropySessionTest.kt index 655956239d..3c849e0d27 100644 --- a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip77Negentropy/NegentropySessionTest.kt +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip77Negentropy/NegentropySessionTest.kt @@ -285,7 +285,7 @@ class NegentropySessionTest { val openCmd = clientSession.open() // Server processes via NegentropyServerSession - val serverSession = NegentropyServerSession("sub1", serverEvents) + val serverSession = NegentropyServerSession.fromEvents("sub1", serverEvents) val response = serverSession.processMessage(openCmd.initialMessage) assertNotNull(response)