feat(relay): NIP-77 negentropy reconciliation

Server-side negentropy: NEG-OPEN snapshots the matching event set, NEG-MSG drives the round-trip via the kmp-negentropy library, and NEG-CLOSE frees per-subId state. Same access controls as REQ apply to NEG-OPEN. supported_nips bumped to advertise "77".
2026-05-07 04:02:02 +00:00
parent d49d4a1025
commit c440186575
5 changed files with 377 additions and 3 deletions
@@ -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"),
                ),
            )

@@ -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,
@@ -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<String> = 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<Event> {
+        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<String>()
+                val needIds = mutableSetOf<String>()
+                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()
+            }
+        }
+}
@@ -96,4 +96,11 @@ class LiveEventStore(
    }

    suspend fun count(filters: List<Filter>) = 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<Event> = store.query(filter)
 }
@@ -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<String, Job>()

+    /**
+     * 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<String, NegentropyServerSession>()
+
    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<String>()
+                val merged = mutableListOf<com.vitorpamplona.quartz.nip01Core.core.Event>()
+                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)
    }