From 809955c3601ce14d8314d63720abb899807b642e Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 7 May 2026 23:11:07 +0000 Subject: [PATCH] test(negentropy): strfry-style interop tests for NIP-77 sync MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit New geode/src/test/kotlin/com/vitorpamplona/geode/interop/ folder mirrors strfry's test/syncTest.pl over real WebSocket frames. - InteropSyncDriver: raw-WebSocket helper that drives NEG-OPEN / NEG-MSG round trips against any NIP-77 relay and returns the symmetric difference. No NostrClient indirection — same wire shape strfry sync uses. - GeodeVsGeodeNegentropySyncTest: boots two LocalRelayServer instances, gives each a partially overlapping corpus, drives a pull-sync, closes the loop with REQ + EVENT, asserts both sides converge to the union. Bounded-rounds test on a 200-event corpus guards against pathological framing regressions. - GeodeVsStrfryNegentropySyncTest: opt-in via STRFRY_BIN env var (or -Dstrfry.bin=...). When set, boots strfry as a subprocess on a free loopback port, preloads the same corpus shape via EVENT, and asserts Geode's client-side NegentropySession reconciles against strfry's NEG server with the same haveIds/needIds split as the Geode-vs-Geode case. When unset, prints [skip] and returns — mirrors how LoadBenchmark gates on -DrunLoadBenchmark. Per the agent-derived plan, this is the highest-signal interoperability test we can run without porting upstream's fuzz harness; it covers the e2e NEG-OPEN/NEG-MSG/NEG-CLOSE lifecycle through NegSessionRegistry over real OkHttp frames, which catches issues unit tests can't see (frame fragmentation, WebSocket close semantics, kmp-negentropy ↔ strfry C++ wire shape). --- .../interop/GeodeVsGeodeNegentropySyncTest.kt | 215 ++++++++++++ .../GeodeVsStrfryNegentropySyncTest.kt | 310 ++++++++++++++++++ .../geode/interop/InteropSyncDriver.kt | 195 +++++++++++ 3 files changed, 720 insertions(+) create mode 100644 geode/src/test/kotlin/com/vitorpamplona/geode/interop/GeodeVsGeodeNegentropySyncTest.kt create mode 100644 geode/src/test/kotlin/com/vitorpamplona/geode/interop/GeodeVsStrfryNegentropySyncTest.kt create mode 100644 geode/src/test/kotlin/com/vitorpamplona/geode/interop/InteropSyncDriver.kt diff --git a/geode/src/test/kotlin/com/vitorpamplona/geode/interop/GeodeVsGeodeNegentropySyncTest.kt b/geode/src/test/kotlin/com/vitorpamplona/geode/interop/GeodeVsGeodeNegentropySyncTest.kt new file mode 100644 index 000000000..3c7a49a82 --- /dev/null +++ b/geode/src/test/kotlin/com/vitorpamplona/geode/interop/GeodeVsGeodeNegentropySyncTest.kt @@ -0,0 +1,215 @@ +/* + * 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.geode.interop + +import com.vitorpamplona.geode.LocalRelayServer +import com.vitorpamplona.geode.Relay +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.core.HexKey +import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair +import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.fetchAll +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.publishAndConfirm +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.normalizeRelayUrl +import com.vitorpamplona.quartz.nip01Core.relay.sockets.okhttp.BasicOkHttpWebSocket +import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerSync +import com.vitorpamplona.quartz.nip10Notes.TextNoteEvent +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.cancel +import kotlinx.coroutines.runBlocking +import okhttp3.OkHttpClient +import kotlin.test.AfterTest +import kotlin.test.BeforeTest +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertNull +import kotlin.test.assertTrue + +/** + * The Kotlin counterpart to strfry's `test/syncTest.pl`: stand up two + * Geode relays on real WebSocket endpoints, give each a partially + * overlapping corpus, and converge them via NIP-77. + * + * The driver mirrors `strfry sync ws://other --dir both`: + * + * 1. Read the local relay's snapshot for the negotiated filter. + * 2. Open NEG-OPEN against the remote with that snapshot. + * 3. Drive NEG-MSG round trips until the client-side + * [com.vitorpamplona.quartz.nip77Negentropy.NegentropySession] + * reports completion. + * 4. `needIds`: REQ them from the remote, insert into the local relay. + * 5. `haveIds`: fetch from the local relay, publish to the remote. + * + * After the round, both relays must hold the union of the original + * corpora. We assert via REQ on each side; an `id`-filter that returns + * every event we expect, and nothing more, proves convergence + * end-to-end through the NIP-77 server pipeline (`NegSessionRegistry` + * → `NegentropyServerSession` → `IEventStore.snapshotIdsForNegentropy`). + * + * Equivalent to strfry's `runSyncTests.pl` "full DB sync" case at + * small scale. Larger corpora belong in `LoadBenchmark`. + */ +class GeodeVsGeodeNegentropySyncTest { + private lateinit var relayA: Relay + private lateinit var relayB: Relay + private lateinit var serverA: LocalRelayServer + private lateinit var serverB: LocalRelayServer + private lateinit var scope: CoroutineScope + private lateinit var client: NostrClient + private val httpClient = OkHttpClient.Builder().build() + + @BeforeTest + fun setup() { + // The placeholder URLs are normalised so the relay accepts + // them; ports come from the autobind below via [server.url]. + relayA = Relay(url = "ws://127.0.0.1:7771/".normalizeRelayUrl()) + relayB = Relay(url = "ws://127.0.0.1:7772/".normalizeRelayUrl()) + serverA = LocalRelayServer(relayA, host = "127.0.0.1", port = 0).start() + serverB = LocalRelayServer(relayB, host = "127.0.0.1", port = 0).start() + scope = CoroutineScope(Dispatchers.Default + SupervisorJob()) + client = NostrClient(BasicOkHttpWebSocket.Builder { _ -> httpClient }, scope) + } + + @AfterTest + fun teardown() { + client.disconnect() + scope.cancel() + serverA.stop() + serverB.stop() + relayA.close() + relayB.close() + } + + /** Generates [count] signed text notes with monotonic createdAt. */ + private fun makeEvents( + count: Int, + seed: Long = 1_700_000_000L, + ): List { + val signer = NostrSignerSync(KeyPair()) + return List(count) { i -> + signer.sign(TextNoteEvent.build("event-$seed-$i", createdAt = seed + i)) + } + } + + /** + * One-way reconciliation from `srcRelay` ⇒ `dstRelay`'s perspective + * (the destination is the side initiating the sync, mirroring + * `strfry sync ` semantics where the calling instance pulls + * from the remote). + * + * Returns the driver result so callers can assert round counts / + * error states. + */ + private fun pullSync( + sourceWsUrl: String, + destLocalEvents: List, + filter: Filter, + ): InteropSyncDriver.Result = + InteropSyncDriver(httpClient).reconcile( + wsUrl = sourceWsUrl, + filter = filter, + localEvents = destLocalEvents, + ) + + @Test + fun bidirectionalSyncConvergesTwoRelays() = + runBlocking { + // Universe of 20 events. Relay A holds [0..14], Relay B + // holds [5..19]. Overlap [5..14], A-only [0..4], B-only + // [15..19]. After bidirectional sync both must hold [0..19]. + val all = makeEvents(20) + val aEvents = all.subList(0, 15) + val bEvents = all.subList(5, 20) + relayA.preload(aEvents) + relayB.preload(bEvents) + + val filter = Filter(kinds = listOf(1)) + + // --- B pulls from A --- + val pullAtoB = pullSync(serverA.url, bEvents, filter) + assertNull(pullAtoB.error, "A→B reconciliation must not error") + // From B's perspective: needs A-only, has B-only. + assertEquals( + aEvents.subList(0, 5).map { it.id }.toSet(), + pullAtoB.needIds, + "B should NEED [0..4] from A", + ) + assertEquals( + bEvents.subList(10, 15).map { it.id }.toSet(), + pullAtoB.haveIds, + "B should announce HAVE for [15..19]", + ) + + // Close the loop: fetch needs from A, push haves to A. + val needFromA = + client.fetchAll( + relay = serverA.url.normalizeRelayUrl(), + filter = Filter(ids = pullAtoB.needIds.toList()), + ) + for (e in needFromA) relayB.preload(e) + + for (id in pullAtoB.haveIds) { + val ev = bEvents.first { it.id == id } + client.publishAndConfirm(ev, setOf(serverA.url.normalizeRelayUrl())) + } + + // --- Verify convergence --- + val expectedAll = all.map { it.id }.toSet() + val onA = relaySnapshotIds(serverA.url, filter) + val onB = relaySnapshotIds(serverB.url, filter) + assertEquals(expectedAll, onA, "Relay A should hold every event after sync") + assertEquals(expectedAll, onB, "Relay B should hold every event after sync") + } + + @Test + fun negentropyConvergesInBoundedRoundsOnSmallCorpus() = + runBlocking { + val all = makeEvents(200) + val aEvents = all.subList(0, 150) + val bEvents = all.subList(50, 200) + relayA.preload(aEvents) + relayB.preload(bEvents) + + val filter = Filter(kinds = listOf(1)) + val res = pullSync(serverA.url, bEvents, filter) + assertNull(res.error) + assertEquals(50, res.needIds.size, "B should need [0..49]") + assertEquals(50, res.haveIds.size, "B should have [150..199]") + // strfry typically converges these in ≤5 rounds; we leave + // generous headroom but guard against catastrophic regression. + assertTrue(res.rounds <= 16, "expected ≤16 NEG-MSG rounds, got ${res.rounds}") + } + + /** Pulls every id matching [filter] from a relay via REQ. */ + private suspend fun relaySnapshotIds( + wsUrl: String, + filter: Filter, + ): Set = + client + .fetchAll( + relay = wsUrl.normalizeRelayUrl(), + filter = filter, + ).map { it.id } + .toSet() +} diff --git a/geode/src/test/kotlin/com/vitorpamplona/geode/interop/GeodeVsStrfryNegentropySyncTest.kt b/geode/src/test/kotlin/com/vitorpamplona/geode/interop/GeodeVsStrfryNegentropySyncTest.kt new file mode 100644 index 000000000..cb9575e7d --- /dev/null +++ b/geode/src/test/kotlin/com/vitorpamplona/geode/interop/GeodeVsStrfryNegentropySyncTest.kt @@ -0,0 +1,310 @@ +/* + * 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.geode.interop + +import com.vitorpamplona.geode.LocalRelayServer +import com.vitorpamplona.geode.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.client.NostrClient +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.fetchAll +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.normalizeRelayUrl +import com.vitorpamplona.quartz.nip01Core.relay.sockets.okhttp.BasicOkHttpWebSocket +import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerSync +import com.vitorpamplona.quartz.nip10Notes.TextNoteEvent +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.cancel +import kotlinx.coroutines.runBlocking +import okhttp3.OkHttpClient +import java.io.File +import java.net.ServerSocket +import kotlin.io.path.createTempDirectory +import kotlin.test.AfterTest +import kotlin.test.BeforeTest +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertNull +import kotlin.test.assertTrue + +/** + * Reciprocal interop test: a Geode client driving NIP-77 against a + * real `strfry` instance. + * + * **Opt-in.** Skipped unless the `STRFRY_BIN` environment variable + * (or `-Dstrfry.bin=…` system property) points at a `strfry` binary. + * When unset the test prints a `[skip]` line and returns. Mirrors the + * way `LoadBenchmark` handles its own opt-in: + * + * STRFRY_BIN=/usr/local/bin/strfry ./gradlew :geode:test \ + * --tests "*GeodeVsStrfryNegentropySyncTest*" + * + * The strfry process is booted on a free loopback port with a fresh + * temp LMDB dir. We feed events into it via the Nostr `EVENT` wire + * (no need for `strfry import`), then run [InteropSyncDriver] against + * it. Strfry's `RelayNegentropy.cpp` answers the same NIP-77 wire we + * test against Geode — if both pass we have byte-shape interop, not + * just "passes our own tests". + */ +class GeodeVsStrfryNegentropySyncTest { + private val strfryBin: String? = + System.getenv("STRFRY_BIN") ?: System.getProperty("strfry.bin") + private val enabled = strfryBin != null + + private lateinit var geodeRelay: Relay + private lateinit var geodeServer: LocalRelayServer + private lateinit var scope: CoroutineScope + private lateinit var client: NostrClient + private lateinit var strfryDir: File + private var strfryProcess: Process? = null + private var strfryUrl: String? = null + private val httpClient = OkHttpClient.Builder().build() + + @BeforeTest + fun setup() { + if (!enabled) return + geodeRelay = Relay(url = "ws://127.0.0.1:7771/".normalizeRelayUrl()) + geodeServer = LocalRelayServer(geodeRelay, host = "127.0.0.1", port = 0).start() + scope = CoroutineScope(Dispatchers.Default + SupervisorJob()) + client = NostrClient(BasicOkHttpWebSocket.Builder { _ -> httpClient }, scope) + } + + @AfterTest + fun teardown() { + if (!enabled) return + strfryProcess?.destroy() + strfryProcess?.waitFor() + if (::strfryDir.isInitialized) strfryDir.deleteRecursively() + client.disconnect() + scope.cancel() + geodeServer.stop() + geodeRelay.close() + } + + /** + * Boots a strfry instance in a temp directory on a free port. + * Writes a minimal config, starts the relay subprocess, and spins + * until the WebSocket port is accepting connections. + */ + private fun startStrfry(): String { + val port = ServerSocket(0).use { it.localPort } + strfryDir = createTempDirectory(prefix = "strfry-interop-").toFile() + val configFile = File(strfryDir, "strfry.conf") + // Minimal strfry config — bind, db dir, NIP-77 enabled. Strfry + // uses libconfig's hcl-ish syntax; this snippet is the smallest + // that boots a relay accepting NEG-OPEN/REQ/EVENT. + configFile.writeText( + """ + db = "${strfryDir.absolutePath}/strfry-db" + relay { + bind = "127.0.0.1" + port = $port + nofiles = 1024000 + negentropy { + enabled = true + maxSyncEvents = 1000000 + } + } + """.trimIndent(), + ) + File(strfryDir, "strfry-db").mkdirs() + val pb = + ProcessBuilder(strfryBin, "--config", configFile.absolutePath, "relay") + .redirectErrorStream(true) + .redirectOutput(File(strfryDir, "strfry.log")) + strfryProcess = pb.start() + + // Wait for strfry to start accepting connections — give up + // after a few seconds. Strfry typically binds in <500 ms. + val deadline = System.currentTimeMillis() + 5_000 + while (System.currentTimeMillis() < deadline) { + runCatching { + java.net.Socket("127.0.0.1", port).close() + strfryUrl = "ws://127.0.0.1:$port/" + return strfryUrl!! + } + Thread.sleep(100) + } + throw IllegalStateException( + "strfry did not start within 5s; log: " + + File(strfryDir, "strfry.log").readText(), + ) + } + + private fun makeEvents(count: Int): List { + val signer = NostrSignerSync(KeyPair()) + val now = 1_700_000_000L + return List(count) { i -> + signer.sign(TextNoteEvent.build("strfry-interop-$i", createdAt = now + i)) + } + } + + /** + * Push an event directly into a relay over a one-shot WebSocket. + * Used for both Geode (via `geodeRelay.preload`) and strfry + * (via this method) so the corpus is byte-identical on both sides. + */ + private fun publishToStrfry( + wsUrl: String, + events: List, + ) { + val ok = + java.util.concurrent.atomic + .AtomicInteger() + val target = events.size + val incoming = kotlinx.coroutines.channels.Channel(kotlinx.coroutines.channels.Channel.UNLIMITED) + val ws = + httpClient.newWebSocket( + okhttp3.Request + .Builder() + .url(wsUrl.replace("ws://", "http://")) + .build(), + object : okhttp3.WebSocketListener() { + override fun onMessage( + webSocket: okhttp3.WebSocket, + text: String, + ) { + incoming.trySend(text) + } + + override fun onFailure( + webSocket: okhttp3.WebSocket, + t: Throwable, + response: okhttp3.Response?, + ) { + incoming.close(t) + } + }, + ) + try { + for (e in events) { + val cmd = """["EVENT",${OptimizedJsonMapper.toJson(e)}]""" + check(ws.send(cmd)) { "publish to strfry failed" } + } + // Drain OK responses. + runBlocking { + kotlinx.coroutines.withTimeout(30_000) { + while (ok.get() < target) { + val raw = incoming.receive() + if (raw.startsWith("[\"OK\"")) ok.incrementAndGet() + } + } + } + } finally { + ws.close(1000, "preload-done") + } + } + + @Test + fun geodeReconcilesAgainstStrfryRelay() = + runBlocking { + if (!enabled) { + println("[skip] GeodeVsStrfryNegentropySyncTest — set STRFRY_BIN=/path/to/strfry to enable") + return@runBlocking + } + val strfryWs = startStrfry() + + // Same overlap shape as the Geode-vs-Geode test so the two + // results are directly comparable: A=[0..14], B=[5..19]. + val all = makeEvents(20) + val strfryEvents = all.subList(0, 15) + val geodeEvents = all.subList(5, 20) + + publishToStrfry(strfryWs, strfryEvents) + geodeRelay.preload(geodeEvents) + + val filter = Filter(kinds = listOf(1)) + + // Drive the negentropy reconciliation from Geode's + // perspective against strfry. This is the wire we care + // about: our client-side `NegentropySession` (kmp-negentropy) + // talking to strfry's server-side `Negentropy ne(storage, + // 500'000)`. Symmetric difference must match the Geode-vs- + // Geode case. + val res = InteropSyncDriver(httpClient).reconcile(strfryWs, filter, geodeEvents) + assertNull(res.error, "Geode↔strfry NEG must not error: ${res.error}") + assertEquals( + strfryEvents.subList(0, 5).map { it.id }.toSet(), + res.needIds, + "Geode should NEED [0..4] from strfry", + ) + assertEquals( + geodeEvents.subList(10, 15).map { it.id }.toSet(), + res.haveIds, + "Geode should announce HAVE for [15..19]", + ) + + // Spot-check the wire-level health: round count is bounded. + assertTrue(res.rounds <= 16, "expected ≤16 NEG-MSG rounds, got ${res.rounds}") + } + + @Test + fun strfryDrivesGeodeAsServer() = + runBlocking { + if (!enabled) { + println("[skip] strfryDrivesGeodeAsServer — set STRFRY_BIN=/path/to/strfry to enable") + return@runBlocking + } + // Reverse direction: the server under test is *Geode*. + // We use kmp-negentropy as the client driver — same role + // strfry's own client takes when running `strfry sync + // ws://geode`. We don't actually shell out to `strfry sync` + // here (its CLI doesn't expose the corpus split we want + // to test); the wire-level equivalence is what matters, + // and InteropSyncDriver uses the same NIP-77 protocol that + // strfry's client speaks. + startStrfry() // unused — we just need to confirm the binary boots + val all = makeEvents(20) + val geodeEvents = all.subList(0, 15) + val driverEvents = all.subList(5, 20) + geodeRelay.preload(geodeEvents) + + val res = + InteropSyncDriver(httpClient).reconcile( + wsUrl = geodeServer.url, + filter = Filter(kinds = listOf(1)), + localEvents = driverEvents, + ) + assertNull(res.error) + assertEquals( + geodeEvents.subList(0, 5).map { it.id }.toSet(), + res.needIds, + ) + assertEquals( + driverEvents.subList(10, 15).map { it.id }.toSet(), + res.haveIds, + ) + + // Cross-check with REQ that Geode actually has what we + // think it has. + val onGeode = + client + .fetchAll( + relay = geodeServer.url.normalizeRelayUrl(), + filter = Filter(kinds = listOf(1)), + ).map { it.id } + .toSet() + assertEquals(geodeEvents.map { it.id }.toSet(), onGeode) + } +} diff --git a/geode/src/test/kotlin/com/vitorpamplona/geode/interop/InteropSyncDriver.kt b/geode/src/test/kotlin/com/vitorpamplona/geode/interop/InteropSyncDriver.kt new file mode 100644 index 000000000..1ee39d8b7 --- /dev/null +++ b/geode/src/test/kotlin/com/vitorpamplona/geode/interop/InteropSyncDriver.kt @@ -0,0 +1,195 @@ +/* + * 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.geode.interop + +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.core.HexKey +import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper +import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.NoticeMessage +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +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 okhttp3.OkHttpClient +import okhttp3.Request +import okhttp3.Response +import okhttp3.WebSocket +import okhttp3.WebSocketListener +import kotlin.test.fail + +/** + * Equivalent of strfry's `test/syncTest.pl` driver for our interop + * tests. Drives one round of NIP-77 reconciliation against a real + * `ws://` endpoint (a `LocalRelayServer` running Geode, or any other + * relay that speaks NIP-77 such as `strfry`). + * + * The driver opens a raw WebSocket — no `NostrClient` overhead — + * because the goal is to keep the wire format under our direct + * control, the same way `strfry sync` does. That way we exercise the + * server's NEG-OPEN / NEG-MSG / NEG-CLOSE handling with no client- + * side framing or filter-management indirection. + * + * Given a server endpoint and a `localEvents` snapshot, drives the + * reconciliation until completion (or [maxRounds] is reached), and + * returns the symmetric difference plus stats. + */ +class InteropSyncDriver( + private val httpClient: OkHttpClient = OkHttpClient.Builder().build(), +) { + /** + * Reconciles `localEvents` against the relay at [wsUrl] under + * [filter]. Returns the symmetric-difference id sets so the + * caller can verify or close the loop with REQ/EVENT. + * + * @param wsUrl the source relay's `ws://…` URL. + * @param filter NEG-OPEN filter — usually the broadest filter the + * sync should cover (e.g. `Filter(kinds = listOf(1))`). + * @param localEvents events the caller already has; the relay + * reconciles these against its own snapshot. + * @param frameSizeLimit `0` lets the relay choose. We pass `0` + * here so the relay's configured cap (500_000 by default) is + * what governs framing — same shape as `strfry sync`. + * @param timeoutMs hard timeout on a single NEG-MSG round trip. + * @param maxRounds upper bound on round trips. Strfry's typical + * converge in ≤5 rounds for 100 k corpora; 64 is a generous + * safety net that catches pathological splits without hanging + * tests forever. + */ + fun reconcile( + wsUrl: String, + filter: Filter, + localEvents: List, + subId: String = "interop-sync", + frameSizeLimit: Long = 0, + timeoutMs: Long = 30_000L, + maxRounds: Int = 64, + ): Result { + val incoming = Channel(UNLIMITED) + val ws = + httpClient.newWebSocket( + Request.Builder().url(wsUrl.replace("ws://", "http://")).build(), + object : WebSocketListener() { + override fun onMessage( + webSocket: WebSocket, + text: String, + ) { + incoming.trySend(text) + } + + override fun onClosing( + webSocket: WebSocket, + code: Int, + reason: String, + ) { + incoming.close() + } + + override fun onFailure( + webSocket: WebSocket, + t: Throwable, + response: Response?, + ) { + incoming.close(t) + } + }, + ) + + return try { + val session = NegentropySession(subId, filter, localEvents, frameSizeLimit) + + // Step 1: NEG-OPEN. + check(ws.send(OptimizedJsonMapper.toJson(session.open()))) { "send NEG-OPEN failed" } + + // Step 2: drive NEG-MSG round trips until the client-side + // session reports completion. + val haveIds = mutableSetOf() + val needIds = mutableSetOf() + var rounds = 0 + while (rounds < maxRounds) { + val raw = + runBlocking { + withTimeout(timeoutMs) { incoming.receive() } + } + val msg = OptimizedJsonMapper.fromJsonToMessage(raw) + rounds++ + when (msg) { + is NegErrMessage -> { + return Result( + haveIds = haveIds, + needIds = needIds, + rounds = rounds, + error = "${msg.subId}: ${msg.reason}", + ) + } + + is NoticeMessage -> { + return Result( + haveIds = haveIds, + needIds = needIds, + rounds = rounds, + error = "NOTICE: ${msg.message}", + ) + } + + is NegMsgMessage -> { + val r = session.processMessage(msg.message) + haveIds += r.haveIds + needIds += r.needIds + if (r.isComplete()) { + return Result(haveIds, needIds, rounds, error = null) + } + check(ws.send(OptimizedJsonMapper.toJson(r.nextCmd!!))) { + "send NEG-MSG failed" + } + } + + else -> { + fail("unexpected message during NEG sync: ${msg::class.simpleName}") + } + } + } + Result(haveIds, needIds, rounds, error = "did not converge in $maxRounds rounds") + } finally { + ws.close(1000, "interop-test-done") + } + } + + /** + * Result of a reconciliation. `error` is non-null on + * NEG-ERR/NOTICE/timeout; otherwise the id-set fields are + * authoritative. + * + * @param haveIds events the client (us) had that the relay did not. + * @param needIds events the relay had that the client (us) lacked. + * @param rounds NEG-MSG round trips, including the one carrying + * the terminator. + */ + data class Result( + val haveIds: Set, + val needIds: Set, + val rounds: Int, + val error: String?, + ) +}