diff --git a/quartz-relay/src/main/kotlin/com/vitorpamplona/quartz/relay/RelayInfo.kt b/quartz-relay/src/main/kotlin/com/vitorpamplona/quartz/relay/RelayInfo.kt index 87884f284..418650de9 100644 --- a/quartz-relay/src/main/kotlin/com/vitorpamplona/quartz/relay/RelayInfo.kt +++ b/quartz-relay/src/main/kotlin/com/vitorpamplona/quartz/relay/RelayInfo.kt @@ -50,8 +50,9 @@ data class RelayInfo( // DeletionRequestModule), NIP-11 (this doc), NIP-40 (expiration // via ExpirationModule), NIP-42 (AUTH — when policy enables), // NIP-45 (COUNT), NIP-50 (search via FTS), NIP-62 (right to vanish), - // NIP-86 (relay management API — when admin pubkeys are configured). - supported_nips = listOf("1", "9", "11", "40", "42", "45", "50", "62", "86"), + // NIP-77 (negentropy reconciliation), NIP-86 (relay management API + // — when admin pubkeys are configured). + supported_nips = listOf("1", "9", "11", "40", "42", "45", "50", "62", "77", "86"), ), ) diff --git a/quartz-relay/src/main/kotlin/com/vitorpamplona/quartz/relay/config/RelayConfig.kt b/quartz-relay/src/main/kotlin/com/vitorpamplona/quartz/relay/config/RelayConfig.kt index 74fc540c6..9762c667a 100644 --- a/quartz-relay/src/main/kotlin/com/vitorpamplona/quartz/relay/config/RelayConfig.kt +++ b/quartz-relay/src/main/kotlin/com/vitorpamplona/quartz/relay/config/RelayConfig.kt @@ -65,7 +65,7 @@ data class RelayConfig( // Keep in sync with `RelayInfo.default()` — // both lists must reflect the NIPs actually // wired into the relay. - ?: listOf("1", "9", "11", "40", "42", "45", "50", "62", "86"), + ?: listOf("1", "9", "11", "40", "42", "45", "50", "62", "77", "86"), privacy_policy = info.privacy_policy, terms_of_service = info.terms_of_service, relay_countries = info.relay_countries, diff --git a/quartz-relay/src/test/kotlin/com/vitorpamplona/quartz/relay/Nip77NegentropyTest.kt b/quartz-relay/src/test/kotlin/com/vitorpamplona/quartz/relay/Nip77NegentropyTest.kt new file mode 100644 index 000000000..a66a30000 --- /dev/null +++ b/quartz-relay/src/test/kotlin/com/vitorpamplona/quartz/relay/Nip77NegentropyTest.kt @@ -0,0 +1,273 @@ +/* + * 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.relay + +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.filters.Filter +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer +import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebSocketListener +import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerSync +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 kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withTimeout +import kotlin.test.AfterTest +import kotlin.test.BeforeTest +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertTrue + +/** + * End-to-end NIP-77 reconciliation through `RelayHub` + the + * in-process WebSocket bridge. + * + * - The relay is preloaded with a known set of events. + * - A [NegentropySession] is initialised with a partially-overlapping + * set on the client side. + * - We drive the NEG-OPEN / NEG-MSG round trips manually until + * `processMessage` reports completion. + * - We assert that `haveIds` (events the client has that the relay + * doesn't) and `needIds` (events the relay has that the client + * doesn't) cover exactly the symmetric difference. + * + * NEG-CLOSE is exercised separately — sending NEG-MSG after CLOSE + * must surface a NEG-ERR from the relay. + */ +class Nip77NegentropyTest { + private lateinit var hub: RelayHub + private val relayUrl: NormalizedRelayUrl = RelayUrlNormalizer.normalize("ws://127.0.0.1:7770/") + + @BeforeTest + fun setup() { + hub = RelayHub() + } + + @AfterTest + fun teardown() { + hub.close() + } + + /** + * Bare-bones WS client that captures every server message into a + * channel and exposes a `send` that goes straight at the in-process + * bridge. We don't need NostrClient's filter-management for these + * tests — we drive the wire. + */ + private class WireClient( + hub: RelayHub, + url: NormalizedRelayUrl, + ) { + val incoming: Channel = Channel(UNLIMITED) + private val ws = + hub.build( + url, + object : WebSocketListener { + override fun onOpen( + pingMillis: Int, + compression: Boolean, + ) {} + + override fun onMessage(text: String) { + incoming.trySend(text) + } + + override fun onClosed( + code: Int, + reason: String, + ) { + incoming.close() + } + + override fun onFailure( + t: Throwable, + code: Int?, + response: String?, + ) { + incoming.close(t) + } + }, + ) + + init { + ws.connect() + } + + fun send(json: String) { + check(ws.send(json)) { "send returned false" } + } + + fun close() { + ws.disconnect() + } + } + + private suspend fun WireClient.nextMessage(timeoutMs: Long = 5_000): Message { + val raw = withTimeout(timeoutMs) { incoming.receive() } + return OptimizedJsonMapper.fromJsonToMessage(raw) + } + + /** Generates [count] signed text notes with monotonic createdAt. */ + private fun makeEvents(count: Int): List { + val signer = NostrSignerSync(KeyPair()) + val now = 1_700_000_000L + return List(count) { i -> + signer.sign(TextNoteEvent.build("event-$i", createdAt = now + i)) + } + } + + @Test + fun negentropyComputesSymmetricDifference() = + runBlocking { + // Universe of 10 events. Relay has events [0..7], client has [3..9] — + // overlap [3..7], relay-only [0..2], client-only [8..9]. + val all = makeEvents(10) + val relayEvents = all.subList(0, 8) + val clientEvents = all.subList(3, 10) + + hub.getOrCreate(relayUrl).preload(relayEvents) + + val client = WireClient(hub, relayUrl) + try { + val session = + NegentropySession( + subId = "neg-1", + filter = Filter(kinds = listOf(1)), + localEvents = clientEvents, + ) + + // Step 1: send NEG-OPEN. + val openCmd = session.open() + client.send(OptimizedJsonMapper.toJson(openCmd)) + + // Step 2: drive NEG-MSG round trips until reconciliation completes. + val haveIds = mutableSetOf() + val needIds = mutableSetOf() + var safety = 32 + while (safety-- > 0) { + val response = client.nextMessage() + if (response is NegErrMessage) { + kotlin.test.fail("relay sent NEG-ERR: ${response.reason}") + } + response as NegMsgMessage + val result = session.processMessage(response.message) + haveIds += result.haveIds + needIds += result.needIds + if (result.isComplete()) break + client.send(OptimizedJsonMapper.toJson(result.nextCmd!!)) + } + assertTrue(safety > 0, "reconciliation did not converge in 32 rounds") + + // Step 3: verify the symmetric difference. + val expectedNeed = relayEvents.subList(0, 3).map { it.id }.toSet() // [0..2] + val expectedHave = clientEvents.subList(5, 7).map { it.id }.toSet() // [8..9] + assertEquals(expectedNeed, needIds, "client should NEED events 0..2 from relay") + assertEquals(expectedHave, haveIds, "client should HAVE events 8..9 to send to relay") + } finally { + client.close() + } + } + + @Test + fun negCloseFreesServerStateAndReopenWorks() = + runBlocking { + val all = makeEvents(5) + hub.getOrCreate(relayUrl).preload(all) + + val client = WireClient(hub, relayUrl) + try { + // First session. + val s1 = NegentropySession("neg", Filter(kinds = listOf(1)), localEvents = emptyList()) + client.send(OptimizedJsonMapper.toJson(s1.open())) + val response = client.nextMessage() as NegMsgMessage + val r1 = s1.processMessage(response.message) + // Client had nothing, so it needs all 5 from the relay. + assertEquals(5, r1.needIds.size) + + // Close. + client.send(OptimizedJsonMapper.toJson(s1.close())) + + // Re-OPEN with the same subId and an empty client store — + // server must build a new session and respond. If the + // close didn't free state, this would either error or + // continue the previous reconciliation. + val s2 = NegentropySession("neg", Filter(kinds = listOf(1)), localEvents = emptyList()) + client.send(OptimizedJsonMapper.toJson(s2.open())) + val resp2 = client.nextMessage() as NegMsgMessage + val r2 = s2.processMessage(resp2.message) + assertEquals(5, r2.needIds.size) + } finally { + client.close() + } + } + + @Test + fun negMsgWithoutOpenReturnsNegErr() = + runBlocking { + val client = WireClient(hub, relayUrl) + try { + // Synthesise a stray NEG-MSG for a sub-id that was never opened. + val raw = """["NEG-MSG","ghost-sub","00"]""" + client.send(raw) + 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")) + } finally { + client.close() + } + } + + @Test + fun negOpenWithSameSubIdReplacesPriorSession() = + runBlocking { + val a = makeEvents(3) + val b = makeEvents(2) + hub.getOrCreate(relayUrl).preload(a + b) + + val client = WireClient(hub, relayUrl) + try { + // First open with localEvents = a; next we'll re-open + // and confirm the new session sees a fresh state. + val first = NegentropySession("dup", Filter(kinds = listOf(1)), localEvents = a) + client.send(OptimizedJsonMapper.toJson(first.open())) + client.nextMessage() as NegMsgMessage // discard + + // Re-OPEN with same subId, different localEvents. + val second = NegentropySession("dup", Filter(kinds = listOf(1)), localEvents = a + b) + client.send(OptimizedJsonMapper.toJson(second.open())) + val resp = client.nextMessage() as NegMsgMessage + val r = second.processMessage(resp.message) + // Client now has every event the relay has → nothing to need. + assertEquals(0, r.needIds.size) + assertEquals(0, r.haveIds.size) + } finally { + client.close() + } + } +} 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 d53525f31..43543b1e5 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 @@ -96,4 +96,11 @@ class LiveEventStore( } suspend fun count(filters: List) = store.count(filters) + + /** + * One-shot snapshot query. Used by NIP-77 negentropy: the server + * needs the full set of event ids matching the filter at the + * moment the NEG-OPEN arrives, not a streamed/live result. + */ + suspend fun snapshotQuery(filter: Filter): List = store.query(filter) } 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 dd43f03d4..9baf769a9 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 @@ -34,6 +34,11 @@ import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.CloseCmd import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.CountCmd import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.EventCmd 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.utils.Log import com.vitorpamplona.quartz.utils.cache.LargeCache import kotlinx.coroutines.CoroutineScope @@ -53,6 +58,14 @@ class RelaySession( ) : AutoCloseable { private val subscriptions = LargeCache() + /** + * NIP-77 negentropy reconciliation sessions, keyed by NEG-OPEN + * subId. Plain hash map here (not [LargeCache]) because it's + * mutated only from the single-threaded `receive()` path — + * RelaySession.receive is serialised by the WebSocket handler. + */ + private val negSessions = HashMap() + private fun addSubscription( subId: String, job: Job, @@ -67,6 +80,7 @@ class RelaySession( fun cancelAllSubscriptions() { subscriptions.forEach { _, job -> job.cancel() } subscriptions.clear() + negSessions.clear() } fun send(message: Message) { @@ -107,6 +121,9 @@ class RelaySession( is ReqCmd -> handleReq(cmd) is CloseCmd -> handleClose(cmd) is CountCmd -> handleCount(cmd) + is NegOpenCmd -> handleNegOpen(cmd) + is NegMsgCmd -> handleNegMsg(cmd) + is NegCloseCmd -> handleNegClose(cmd) else -> send(NoticeMessage("error: unsupported command ${cmd.label()}")) } } @@ -194,6 +211,82 @@ class RelaySession( } } + // -- NIP-77: NEG-OPEN ----------------------------------------------------- + + /** + * Open a negentropy 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. + * + * 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. + */ + private suspend fun handleNegOpen(cmd: NegOpenCmd) { + // Run the same access controls as REQ would. + val asReq = ReqCmd(cmd.subId, listOf(cmd.filter)) + val gate = policy.accept(asReq) + if (gate is PolicyResult.Rejected) { + send(NegErrMessage(cmd.subId, gate.reason)) + return + } + val filters = (gate as PolicyResult.Accepted).cmd.filters + + // Drop any prior session at this subId (NIP-77: same-subId + // OPEN replaces). + negSessions.remove(cmd.subId) + + val events = + if (filters.size == 1) { + store.snapshotQuery(filters[0]) + } else { + // Multiple filters: union the snapshots and dedupe by id. + val seen = HashSet() + val merged = mutableListOf() + for (f in filters) { + for (e in store.snapshotQuery(f)) { + if (seen.add(e.id)) merged += e + } + } + merged + } + + val neg = NegentropyServerSession(cmd.subId, events) + negSessions[cmd.subId] = neg + + try { + val response = neg.processMessage(cmd.initialMessage) + if (response != null) send(response) + } catch (e: Exception) { + negSessions.remove(cmd.subId) + send(NegErrMessage(cmd.subId, "error: ${e.message ?: e::class.simpleName}")) + } + } + + // -- NIP-77: NEG-MSG ------------------------------------------------------ + private fun handleNegMsg(cmd: NegMsgCmd) { + val neg = negSessions[cmd.subId] + if (neg == null) { + send(NegErrMessage(cmd.subId, "error: no negentropy session for ${cmd.subId}")) + return + } + try { + val response = neg.processMessage(cmd.message) + if (response != null) send(response) + } catch (e: Exception) { + negSessions.remove(cmd.subId) + send(NegErrMessage(cmd.subId, "error: ${e.message ?: e::class.simpleName}")) + } + } + + // -- NIP-77: NEG-CLOSE ---------------------------------------------------- + private fun handleNegClose(cmd: NegCloseCmd) { + // Spec: clients send NEG-CLOSE to free server-side state. + // Silent no-op if the session is unknown — there's no authoritative + // error response in NIP-77 for an unknown close. + negSessions.remove(cmd.subId) + } + init { policy.onConnect(::send) }