From 9f7deee8f3ec3a465fde223ec7b9415247b3369d Mon Sep 17 00:00:00 2001 From: Vitor Pamplona Date: Tue, 12 May 2026 14:44:33 -0400 Subject: [PATCH] refactor(geode): rename relay classes for clarity, move test helpers off public API MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Renames in the geode module to make naming honest: - Relay -> RelayEngine (the transport-agnostic core) - LocalRelayServer -> KtorRelay (Ktor/CIO transport; defaults to 127.0.0.1 but supports public deployment, so "Local" was misleading) - RelayHub -> InProcessRelays (URL-keyed registry that doubles as a WebsocketBuilder for in-JVM tests; "Hub"/"Pool" implied substitutable resources, but each URL maps to a distinct relay) Also moves the test-only helpers `preload` and `publish` out of RelayEngine and into a testFixtures extension file. As part of that, `publish` now waits for the relay's OK reply before returning so that publish-then-subscribe is deterministic — previously the fire-and-forget submit let a subscription register after publish returned but before the ingest queue fanned out, causing ephemeral kinds (not persisted) to leak to late subscribers. Fixes the flaky `Nip01ComplianceTest.ephemeralEventIsNotStoredAndDoesNotShowOnFollowupReq`. Co-Authored-By: Claude Opus 4.7 (1M context) --- geode/plans/2026-05-07-connection-scaling.md | 2 +- .../2026-05-07-event-ingestion-batching.md | 2 +- .../geode/{RelayHub.kt => InProcessRelays.kt} | 18 ++--- .../{LocalRelayServer.kt => KtorRelay.kt} | 12 ++-- .../kotlin/com/vitorpamplona/geode/Main.kt | 4 +- .../geode/{Relay.kt => RelayEngine.kt} | 37 +---------- .../com/vitorpamplona/geode/RelayInfo.kt | 2 +- .../vitorpamplona/geode/config/RelayConfig.kt | 2 +- .../geode/server/WebSocketSessionPump.kt | 2 +- .../geode/GracefulShutdownTest.kt | 10 +-- ...calRelayServerTest.kt => KtorRelayTest.kt} | 29 +++++---- .../geode/Nip01ComplianceTest.kt | 2 + .../vitorpamplona/geode/Nip09DeletionTest.kt | 6 +- .../geode/Nip40ExpirationTest.kt | 4 +- .../vitorpamplona/geode/Nip62VanishTest.kt | 4 +- .../geode/Nip77NegentropyTest.kt | 13 ++-- .../geode/admin/Nip86EndToEndTest.kt | 18 ++--- .../interop/GeodeVsGeodeNegentropySyncTest.kt | 21 +++--- .../geode/interop/InteropSyncDriver.kt | 2 +- .../vitorpamplona/geode/perf/LoadBenchmark.kt | 10 +-- .../geode/persistence/PersistenceTest.kt | 22 +++---- .../geode/policies/PoliciesIntegrationTest.kt | 8 +-- .../geode/policies/PoliciesTest.kt | 2 +- .../geode/testing/RelayClientTest.kt | 14 ++-- .../testing/RelayEngineTestExtensions.kt | 65 +++++++++++++++++++ 25 files changed, 176 insertions(+), 135 deletions(-) rename geode/src/main/kotlin/com/vitorpamplona/geode/{RelayHub.kt => InProcessRelays.kt} (90%) rename geode/src/main/kotlin/com/vitorpamplona/geode/{LocalRelayServer.kt => KtorRelay.kt} (97%) rename geode/src/main/kotlin/com/vitorpamplona/geode/{Relay.kt => RelayEngine.kt} (85%) rename geode/src/test/kotlin/com/vitorpamplona/geode/{LocalRelayServerTest.kt => KtorRelayTest.kt} (94%) create mode 100644 geode/src/testFixtures/kotlin/com/vitorpamplona/geode/testing/RelayEngineTestExtensions.kt diff --git a/geode/plans/2026-05-07-connection-scaling.md b/geode/plans/2026-05-07-connection-scaling.md index e16ab3f11..c8c8c521d 100644 --- a/geode/plans/2026-05-07-connection-scaling.md +++ b/geode/plans/2026-05-07-connection-scaling.md @@ -95,7 +95,7 @@ path = "/" Default is **`null` (Ktor default)** — no behavior change unless an operator explicitly tunes them. The values are wired through -`LocalRelayServer` into the new `embeddedServer(factory = CIO, +`KtorRelay` into the new `embeddedServer(factory = CIO, rootConfig = serverConfig {…}, configure = {…})` overload (the short-form `embeddedServer(factory, host, port) {…}` overload doesn't expose CIO config). The auto-connector that the short form created diff --git a/geode/plans/2026-05-07-event-ingestion-batching.md b/geode/plans/2026-05-07-event-ingestion-batching.md index eb7d8d9d1..7ec6d4225 100644 --- a/geode/plans/2026-05-07-event-ingestion-batching.md +++ b/geode/plans/2026-05-07-event-ingestion-batching.md @@ -93,7 +93,7 @@ on each batch before opening the SQLite transaction. Failed verifies pre-mark `Rejected` and skip the insert. Wired through `NostrServer(parallelVerify = ...)` and -`geode.Relay(parallelVerify = ...)`, controlled by +`geode.RelayEngine(parallelVerify = ...)`, controlled by `[options].parallel_verify` in the relay config (default `true`) and `--no-parallel-verify` on the CLI. Internal direct callers of `NostrServer` (tests, library users) are opt-in: the flag defaults diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/RelayHub.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/InProcessRelays.kt similarity index 90% rename from geode/src/main/kotlin/com/vitorpamplona/geode/RelayHub.kt rename to geode/src/main/kotlin/com/vitorpamplona/geode/InProcessRelays.kt index 12d97451a..860238688 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/RelayHub.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/InProcessRelays.kt @@ -32,7 +32,7 @@ import com.vitorpamplona.quartz.nip77Negentropy.NegentropySettings import java.util.concurrent.ConcurrentHashMap /** - * Registry of [Relay] instances keyed by relay URL. Implements + * Registry of [RelayEngine] instances keyed by relay URL. Implements * [WebsocketBuilder] so it can be plugged into * [com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient] in place of * `BasicOkHttpWebSocket.Builder` to redirect every outbound connection to an @@ -40,7 +40,7 @@ import java.util.concurrent.ConcurrentHashMap * * Usage: * ``` - * val hub = RelayHub() + * val hub = InProcessRelays() * val relay = hub.getOrCreate("ws://test.relay/") * runBlocking { relay.preload(listOf(event1, event2)) } * val client = NostrClient(hub, scope) @@ -49,20 +49,20 @@ import java.util.concurrent.ConcurrentHashMap * Unknown URLs auto-create an empty relay so a single hub can transparently * back any number of test endpoints. */ -class RelayHub( +class InProcessRelays( private val defaultPolicy: () -> IRelayPolicy = { EmptyPolicy }, private val negentropySettings: NegentropySettings = NegentropySettings.Default, ) : WebsocketBuilder, AutoCloseable { - private val relays = ConcurrentHashMap() + private val relays = ConcurrentHashMap() @Volatile private var closed = false - fun getOrCreate(url: NormalizedRelayUrl): Relay { - check(!closed) { "RelayHub has been closed" } + fun getOrCreate(url: NormalizedRelayUrl): RelayEngine { + check(!closed) { "InProcessRelays has been closed" } return relays.getOrPut(url) { - Relay( + RelayEngine( url = url, policyBuilder = defaultPolicy, negentropySettings = negentropySettings, @@ -70,9 +70,9 @@ class RelayHub( } } - fun getOrCreate(url: String): Relay = getOrCreate(RelayUrlNormalizer.normalize(url)) + fun getOrCreate(url: String): RelayEngine = getOrCreate(RelayUrlNormalizer.normalize(url)) - fun get(url: NormalizedRelayUrl): Relay? = relays[url] + fun get(url: NormalizedRelayUrl): RelayEngine? = relays[url] fun urls(): Set = relays.keys.toSet() diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/LocalRelayServer.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/KtorRelay.kt similarity index 97% rename from geode/src/main/kotlin/com/vitorpamplona/geode/LocalRelayServer.kt rename to geode/src/main/kotlin/com/vitorpamplona/geode/KtorRelay.kt index 56ef2f6da..93a5685e7 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/LocalRelayServer.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/KtorRelay.kt @@ -48,7 +48,7 @@ import kotlinx.coroutines.runBlocking import java.util.concurrent.ConcurrentHashMap /** - * Hosts a [Relay] over a real `ws://` endpoint backed by Ktor + CIO. + * Hosts a [RelayEngine] over a real `ws://` endpoint backed by Ktor + CIO. * * Use this when something other than the in-process * [com.vitorpamplona.quartz.nip01Core.relay.server.inprocess.InProcessWebSocket] @@ -56,19 +56,19 @@ import java.util.concurrent.ConcurrentHashMap * tooling, external clients, or a standalone "run a Nostr relay" * process. * - * For unit-test wiring inside a single JVM, prefer [RelayHub] + the + * For unit-test wiring inside a single JVM, prefer [InProcessRelays] + the * in-process socket — same protocol, no socket overhead. * * Lifecycle: * ``` - * val server = LocalRelayServer(Relay(url = ...)).start() + * val server = KtorRelay(RelayEngine(url = ...)).start() * println("listening on ${server.url}") * // ... do stuff ... * server.stop() * ``` */ -class LocalRelayServer( - val relay: Relay, +class KtorRelay( + val relay: RelayEngine, val host: String = "127.0.0.1", /** Pass 0 to let the OS pick a free port. Read [url] after [start] to learn it. */ val port: Int = 0, @@ -178,7 +178,7 @@ class LocalRelayServer( * Binds the Ktor engine. Returns once the engine reports ready, so * [url] is safe to read on the very next line. */ - fun start(): LocalRelayServer { + fun start(): KtorRelay { // Snapshot the constructor-supplied overrides into locals so // the `configure` lambda below can assign to its receiver // without the names colliding with outer properties. diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt index 9290fef8e..021de8d95 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt @@ -117,7 +117,7 @@ fun main(args: Array) { maxSessionsPerConnection = config.negentropy.max_sessions_per_connection, ) val relay = - Relay( + RelayEngine( advertisedUrl, store, info, @@ -132,7 +132,7 @@ fun main(args: Array) { val frameLimit = (config.limits.max_ws_frame_bytes ?: config.limits.max_ws_message_bytes)?.toLong() val server = - LocalRelayServer( + KtorRelay( relay, host = host, port = port, diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/Relay.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/RelayEngine.kt similarity index 85% rename from geode/src/main/kotlin/com/vitorpamplona/geode/Relay.kt rename to geode/src/main/kotlin/com/vitorpamplona/geode/RelayEngine.kt index cccdb7090..1d8c67d6e 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/Relay.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/RelayEngine.kt @@ -23,9 +23,6 @@ package com.vitorpamplona.geode import com.vitorpamplona.geode.persistence.BannedEntry import com.vitorpamplona.geode.persistence.RelayPersistedState import com.vitorpamplona.geode.persistence.RelayStateStore -import com.vitorpamplona.quartz.nip01Core.core.Event -import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper -import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.EventCmd import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl import com.vitorpamplona.quartz.nip01Core.relay.server.IRelayPolicy import com.vitorpamplona.quartz.nip01Core.relay.server.NostrServer @@ -51,13 +48,13 @@ import kotlin.coroutines.CoroutineContext * * Two transports: * - [com.vitorpamplona.quartz.nip01Core.relay.server.inprocess.InProcessWebSocket] / - * [RelayHub] — no socket, fastest path, ideal + * [InProcessRelays] — no socket, fastest path, ideal * for unit tests inside one JVM. - * - [LocalRelayServer] — Ktor `embeddedServer` listening on a real port. + * - [KtorRelay] — Ktor `embeddedServer` listening on a real port. * Use when external clients need to connect (`cli`, instrumented tests, * standalone deployment). */ -class Relay( +class RelayEngine( val url: NormalizedRelayUrl, val store: IEventStore = EventStore(dbName = null, relay = url), info: RelayInfo = RelayInfo.default(url), @@ -179,33 +176,5 @@ class Relay( negentropySettings = negentropySettings, ) - /** - * Inserts events directly into the underlying store, bypassing the wire protocol. - * - * Use this for **pre-test setup** — events that exist before any client connects. - * It does NOT broadcast to active subscriptions. For sending events that should - * fan out to live subscribers (post-EOSE), use [publish] instead. - */ - suspend fun preload(events: Iterable) { - events.forEach { store.insert(it) } - } - - /** @see preload(Iterable) */ - suspend fun preload(vararg events: Event) = preload(events.toList()) - - /** - * Publishes an event through the relay's session machinery so it both lands - * in the store and fans out to active subscriptions matching its filters - * (mirrors what a real client would do via an `EVENT` command). - */ - suspend fun publish(event: Event) { - val session = server.connect { /* ignore OK echo */ } - try { - session.receive(OptimizedJsonMapper.toJson(EventCmd(event))) - } finally { - session.close() - } - } - override fun close() = server.close() } diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/RelayInfo.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/RelayInfo.kt index bd2d9c55e..ebf41edbe 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/RelayInfo.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/RelayInfo.kt @@ -65,7 +65,7 @@ data class RelayInfo( val SUPPORTED_NIPS: List = listOf("1", "9", "11", "40", "42", "45", "50", "62", "77", "86") - /** Pre-built default for `Relay(url = ...)` — advertises the supported NIPs. */ + /** Pre-built default for `RelayEngine(url = ...)` — advertises the supported NIPs. */ fun default(url: NormalizedRelayUrl): RelayInfo = RelayInfo( Nip11RelayInformation( 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 b8f162ffb..c900c4869 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/config/RelayConfig.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/config/RelayConfig.kt @@ -198,7 +198,7 @@ data class RelayConfig( /** * NIP-86 relay management API. When [pubkeys] is non-empty, - * `LocalRelayServer` exposes a POST endpoint at the relay path + * `KtorRelay` exposes a POST endpoint at the relay path * that accepts JSON-RPC admin requests authenticated via NIP-98 * HTTP-Auth. Only requests signed by one of these pubkeys are * dispatched. diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/server/WebSocketSessionPump.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/server/WebSocketSessionPump.kt index f5319779d..66fb609cd 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/server/WebSocketSessionPump.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/server/WebSocketSessionPump.kt @@ -33,7 +33,7 @@ import java.util.concurrent.atomic.AtomicInteger /** * Per-WebSocket pump that owns the bounded outbound queue and the - * writer coroutine. Pulled out of `LocalRelayServer` so that file + * writer coroutine. Pulled out of `KtorRelay` so that file * stays focused on Ktor wiring; the slow-client / backpressure * policy now lives next to the data structures it manages. * diff --git a/geode/src/test/kotlin/com/vitorpamplona/geode/GracefulShutdownTest.kt b/geode/src/test/kotlin/com/vitorpamplona/geode/GracefulShutdownTest.kt index 04d55a660..f6677d7de 100644 --- a/geode/src/test/kotlin/com/vitorpamplona/geode/GracefulShutdownTest.kt +++ b/geode/src/test/kotlin/com/vitorpamplona/geode/GracefulShutdownTest.kt @@ -48,14 +48,14 @@ import kotlin.test.assertNotNull import kotlin.test.assertTrue /** - * Tests [LocalRelayServer.stop] honours the graceful-shutdown contract: + * Tests [KtorRelay.stop] honours the graceful-shutdown contract: * 1. Active clients receive a `NOTICE` warning of imminent shutdown. * 2. The active session counter accurately tracks open WS sessions. * 3. After `stop()` returns, no sessions remain registered. */ class GracefulShutdownTest { - private lateinit var relay: Relay - private lateinit var server: LocalRelayServer + private lateinit var relay: RelayEngine + private lateinit var server: KtorRelay private lateinit var scope: CoroutineScope private lateinit var client: NostrClient @@ -64,8 +64,8 @@ class GracefulShutdownTest { @BeforeTest fun setup() { val placeholder = "ws://127.0.0.1:7771/".normalizeRelayUrl() - relay = Relay(url = placeholder) - server = LocalRelayServer(relay, host = "127.0.0.1", port = 0).start() + relay = RelayEngine(url = placeholder) + server = KtorRelay(relay, host = "127.0.0.1", port = 0).start() scope = CoroutineScope(Dispatchers.Default + SupervisorJob()) val builder = BasicOkHttpWebSocket.Builder { _ -> httpClient } client = NostrClient(builder, scope) diff --git a/geode/src/test/kotlin/com/vitorpamplona/geode/LocalRelayServerTest.kt b/geode/src/test/kotlin/com/vitorpamplona/geode/KtorRelayTest.kt similarity index 94% rename from geode/src/test/kotlin/com/vitorpamplona/geode/LocalRelayServerTest.kt rename to geode/src/test/kotlin/com/vitorpamplona/geode/KtorRelayTest.kt index 6489e794f..682757df3 100644 --- a/geode/src/test/kotlin/com/vitorpamplona/geode/LocalRelayServerTest.kt +++ b/geode/src/test/kotlin/com/vitorpamplona/geode/KtorRelayTest.kt @@ -21,6 +21,8 @@ package com.vitorpamplona.geode import com.vitorpamplona.geode.fixtures.SyntheticEvents +import com.vitorpamplona.geode.testing.preload +import com.vitorpamplona.geode.testing.publish import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair import com.vitorpamplona.quartz.nip01Core.metadata.MetadataEvent import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient @@ -38,6 +40,7 @@ import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.cancel import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withTimeoutOrNull import okhttp3.OkHttpClient import okhttp3.Request import kotlin.test.AfterTest @@ -49,7 +52,7 @@ import kotlin.test.assertTrue /** * End-to-end tests that drive a real `ws://` connection between the - * production [NostrClient] (over OkHttp) and the [LocalRelayServer] + * production [NostrClient] (over OkHttp) and the [KtorRelay] * (Ktor + CIO). These prove the relay implements: * * - NIP-01 wire protocol (REQ/EVENT/EOSE) over real WebSockets @@ -62,9 +65,9 @@ import kotlin.test.assertTrue * Tests use port 0 for autobind to avoid conflicts when multiple suites * run in parallel. */ -class LocalRelayServerTest { - private lateinit var relay: Relay - private lateinit var server: LocalRelayServer +class KtorRelayTest { + private lateinit var relay: RelayEngine + private lateinit var server: KtorRelay private lateinit var scope: CoroutineScope private lateinit var client: NostrClient @@ -76,8 +79,8 @@ class LocalRelayServerTest { // must be resolvable by the Nostr URL normalizer, which only // accepts loopback addresses. 127.0.0.1 qualifies. val placeholderUrl = "ws://127.0.0.1:7771/".normalizeRelayUrl() - relay = Relay(url = placeholderUrl) - server = LocalRelayServer(relay, host = "127.0.0.1", port = 0).start() + relay = RelayEngine(url = placeholderUrl) + server = KtorRelay(relay, host = "127.0.0.1", port = 0).start() scope = CoroutineScope(Dispatchers.Default + SupervisorJob()) val builder = BasicOkHttpWebSocket.Builder { _ -> httpClient } client = NostrClient(builder, scope) @@ -191,8 +194,8 @@ class LocalRelayServerTest { // Spin up a second relay that requires AUTH. Bind on a // separate port so it doesn't collide with [setup]'s server. val authUrl = "ws://127.0.0.1:7772/".normalizeRelayUrl() - val authRelay = Relay(authUrl, policyBuilder = { FullAuthPolicy(authUrl) }) - val authServer = LocalRelayServer(authRelay, host = "127.0.0.1", port = 0).start() + val authRelay = RelayEngine(authUrl, policyBuilder = { FullAuthPolicy(authUrl) }) + val authServer = KtorRelay(authRelay, host = "127.0.0.1", port = 0).start() try { val signer = com.vitorpamplona.quartz.nip01Core.signers @@ -228,8 +231,8 @@ class LocalRelayServerTest { val freePort = java.net.ServerSocket(0).use { it.localPort } val authUrl = "ws://127.0.0.1:$freePort/".normalizeRelayUrl() - val authRelay = Relay(authUrl, policyBuilder = { FullAuthPolicy(authUrl) }) - val authServer = LocalRelayServer(authRelay, host = "127.0.0.1", port = freePort).start() + val authRelay = RelayEngine(authUrl, policyBuilder = { FullAuthPolicy(authUrl) }) + val authServer = KtorRelay(authRelay, host = "127.0.0.1", port = freePort).start() try { val signer = com.vitorpamplona.quartz.nip01Core.signers @@ -334,9 +337,9 @@ class LocalRelayServerTest { supported_nips = listOf("1", "11", "42"), ), ) - val customRelay = Relay(customUrl, info = customInfo) + val customRelay = RelayEngine(customUrl, info = customInfo) val customServer = - LocalRelayServer(customRelay, host = "127.0.0.1", port = freePort).start() + KtorRelay(customRelay, host = "127.0.0.1", port = freePort).start() try { val httpUrl = customServer.url.replace("ws://", "http://") val response = @@ -405,7 +408,7 @@ class LocalRelayServerTest { val late = signer.sign(TextNoteEvent.build("post-close")) relay.publish(late) - val seen = kotlinx.coroutines.withTimeoutOrNull(500) { ch.receive() } + val seen = withTimeoutOrNull(500) { ch.receive() } assertEquals(null, seen, "events arriving after CLOSE must not reach the unsubscribed client") } } diff --git a/geode/src/test/kotlin/com/vitorpamplona/geode/Nip01ComplianceTest.kt b/geode/src/test/kotlin/com/vitorpamplona/geode/Nip01ComplianceTest.kt index 7a5a865cc..41fc1b218 100644 --- a/geode/src/test/kotlin/com/vitorpamplona/geode/Nip01ComplianceTest.kt +++ b/geode/src/test/kotlin/com/vitorpamplona/geode/Nip01ComplianceTest.kt @@ -24,6 +24,8 @@ import com.vitorpamplona.geode.fixtures.SyntheticEvents import com.vitorpamplona.geode.testing.RelayClientTest import com.vitorpamplona.geode.testing.collectUntilEose import com.vitorpamplona.geode.testing.collectUntilEoseMulti +import com.vitorpamplona.geode.testing.preload +import com.vitorpamplona.geode.testing.publish import com.vitorpamplona.quartz.nip01Core.core.Event import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener diff --git a/geode/src/test/kotlin/com/vitorpamplona/geode/Nip09DeletionTest.kt b/geode/src/test/kotlin/com/vitorpamplona/geode/Nip09DeletionTest.kt index d721d6686..c4e801665 100644 --- a/geode/src/test/kotlin/com/vitorpamplona/geode/Nip09DeletionTest.kt +++ b/geode/src/test/kotlin/com/vitorpamplona/geode/Nip09DeletionTest.kt @@ -42,7 +42,7 @@ import kotlin.test.assertEquals /** * Verifies NIP-09 deletion request behavior end-to-end through - * `NostrClient` → `RelayHub`. The relay's + * `NostrClient` → `InProcessRelays`. The relay's * [com.vitorpamplona.quartz.nip01Core.store.sqlite.DeletionRequestModule] * is responsible for honouring kind-5 events: * @@ -54,14 +54,14 @@ import kotlin.test.assertEquals * pubkey X cannot delete pubkey Y's events. */ class Nip09DeletionTest { - private lateinit var hub: RelayHub + private lateinit var hub: InProcessRelays private lateinit var scope: CoroutineScope private lateinit var client: NostrClient private val relayUrl: NormalizedRelayUrl = RelayUrlNormalizer.normalize("ws://127.0.0.1:7770/") @BeforeTest fun setup() { - hub = RelayHub() + hub = InProcessRelays() scope = CoroutineScope(Dispatchers.Default + SupervisorJob()) client = NostrClient(hub, scope) } diff --git a/geode/src/test/kotlin/com/vitorpamplona/geode/Nip40ExpirationTest.kt b/geode/src/test/kotlin/com/vitorpamplona/geode/Nip40ExpirationTest.kt index a4dc6c3ef..9df1e8c04 100644 --- a/geode/src/test/kotlin/com/vitorpamplona/geode/Nip40ExpirationTest.kt +++ b/geode/src/test/kotlin/com/vitorpamplona/geode/Nip40ExpirationTest.kt @@ -51,14 +51,14 @@ import kotlin.test.assertNotNull * passes that `isExpired()` returns true on read). */ class Nip40ExpirationTest { - private lateinit var hub: RelayHub + private lateinit var hub: InProcessRelays private lateinit var scope: CoroutineScope private lateinit var client: NostrClient private val relayUrl: NormalizedRelayUrl = RelayUrlNormalizer.normalize("ws://127.0.0.1:7770/") @BeforeTest fun setup() { - hub = RelayHub() + hub = InProcessRelays() scope = CoroutineScope(Dispatchers.Default + SupervisorJob()) client = NostrClient(hub, scope) } diff --git a/geode/src/test/kotlin/com/vitorpamplona/geode/Nip62VanishTest.kt b/geode/src/test/kotlin/com/vitorpamplona/geode/Nip62VanishTest.kt index 620c88a91..a4f9e1447 100644 --- a/geode/src/test/kotlin/com/vitorpamplona/geode/Nip62VanishTest.kt +++ b/geode/src/test/kotlin/com/vitorpamplona/geode/Nip62VanishTest.kt @@ -55,14 +55,14 @@ import kotlin.test.assertNull * - A vanish from author A does not affect author B's events. */ class Nip62VanishTest { - private lateinit var hub: RelayHub + private lateinit var hub: InProcessRelays private lateinit var scope: CoroutineScope private lateinit var client: NostrClient private val relayUrl: NormalizedRelayUrl = RelayUrlNormalizer.normalize("ws://127.0.0.1:7770/") @BeforeTest fun setup() { - hub = RelayHub() + hub = InProcessRelays() scope = CoroutineScope(Dispatchers.Default + SupervisorJob()) client = NostrClient(hub, scope) } diff --git a/geode/src/test/kotlin/com/vitorpamplona/geode/Nip77NegentropyTest.kt b/geode/src/test/kotlin/com/vitorpamplona/geode/Nip77NegentropyTest.kt index a205d8759..aedb80a8c 100644 --- a/geode/src/test/kotlin/com/vitorpamplona/geode/Nip77NegentropyTest.kt +++ b/geode/src/test/kotlin/com/vitorpamplona/geode/Nip77NegentropyTest.kt @@ -20,6 +20,7 @@ */ package com.vitorpamplona.geode +import com.vitorpamplona.geode.testing.preload import com.vitorpamplona.quartz.nip01Core.core.Event import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair @@ -46,7 +47,7 @@ import kotlin.test.assertEquals import kotlin.test.assertTrue /** - * End-to-end NIP-77 reconciliation through `RelayHub` + the + * End-to-end NIP-77 reconciliation through `InProcessRelays` + the * in-process WebSocket bridge. * * - The relay is preloaded with a known set of events. @@ -62,12 +63,12 @@ import kotlin.test.assertTrue * must surface a NEG-ERR from the relay. */ class Nip77NegentropyTest { - private lateinit var hub: RelayHub + private lateinit var hub: InProcessRelays private val relayUrl: NormalizedRelayUrl = RelayUrlNormalizer.normalize("ws://127.0.0.1:7770/") @BeforeTest fun setup() { - hub = RelayHub() + hub = InProcessRelays() } @AfterTest @@ -82,7 +83,7 @@ class Nip77NegentropyTest { * tests — we drive the wire. */ private class WireClient( - hub: RelayHub, + hub: InProcessRelays, url: NormalizedRelayUrl, ) { val incoming: Channel = Channel(UNLIMITED) @@ -252,7 +253,7 @@ class Nip77NegentropyTest { // 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)) + val capped = InProcessRelays(negentropySettings = NegentropySettings(maxSyncEvents = 5)) try { val capUrl = RelayUrlNormalizer.normalize("ws://127.0.0.1:7771/") val events = makeEvents(20) @@ -286,7 +287,7 @@ class Nip77NegentropyTest { 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)) + val capped = InProcessRelays(negentropySettings = NegentropySettings(maxSessionsPerConnection = 2)) try { val capUrl = RelayUrlNormalizer.normalize("ws://127.0.0.1:7772/") capped.getOrCreate(capUrl).preload(makeEvents(3)) diff --git a/geode/src/test/kotlin/com/vitorpamplona/geode/admin/Nip86EndToEndTest.kt b/geode/src/test/kotlin/com/vitorpamplona/geode/admin/Nip86EndToEndTest.kt index f83ab0d85..9ced9c136 100644 --- a/geode/src/test/kotlin/com/vitorpamplona/geode/admin/Nip86EndToEndTest.kt +++ b/geode/src/test/kotlin/com/vitorpamplona/geode/admin/Nip86EndToEndTest.kt @@ -20,8 +20,8 @@ */ package com.vitorpamplona.geode.admin -import com.vitorpamplona.geode.LocalRelayServer -import com.vitorpamplona.geode.Relay +import com.vitorpamplona.geode.KtorRelay +import com.vitorpamplona.geode.RelayEngine import com.vitorpamplona.quartz.nip01Core.core.JsonMapper import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient @@ -53,14 +53,14 @@ import kotlin.test.assertEquals import kotlin.test.assertTrue /** - * Drives a real `LocalRelayServer` over HTTP and proves the NIP-86 + * Drives a real `KtorRelay` over HTTP and proves the NIP-86 * admin RPC flow works end-to-end: NIP-98 auth, admin allow-list * gate, ban mutation, and the resulting policy effect on a follow-up * EVENT publish. */ class Nip86EndToEndTest { - private lateinit var relay: Relay - private lateinit var server: LocalRelayServer + private lateinit var relay: RelayEngine + private lateinit var server: KtorRelay private lateinit var scope: CoroutineScope private lateinit var nostrClient: NostrClient @@ -73,9 +73,9 @@ class Nip86EndToEndTest { @BeforeTest fun setup() { val placeholder = "ws://127.0.0.1:7771/".normalizeRelayUrl() - relay = Relay(url = placeholder) + relay = RelayEngine(url = placeholder) server = - LocalRelayServer( + KtorRelay( relay = relay, host = "127.0.0.1", port = 0, @@ -200,9 +200,9 @@ class Nip86EndToEndTest { runBlocking { // Spin up a *separate* server with no admin pubkeys. val placeholder = "ws://127.0.0.1:7771/".normalizeRelayUrl() - val openRelay = Relay(url = placeholder) + val openRelay = RelayEngine(url = placeholder) val openServer = - LocalRelayServer(openRelay, host = "127.0.0.1", port = 0).start() + KtorRelay(openRelay, host = "127.0.0.1", port = 0).start() try { val openHttpUrl = openServer.url.replace("ws://", "http://") val body = diff --git a/geode/src/test/kotlin/com/vitorpamplona/geode/interop/GeodeVsGeodeNegentropySyncTest.kt b/geode/src/test/kotlin/com/vitorpamplona/geode/interop/GeodeVsGeodeNegentropySyncTest.kt index dd79a9475..b0ad5a863 100644 --- a/geode/src/test/kotlin/com/vitorpamplona/geode/interop/GeodeVsGeodeNegentropySyncTest.kt +++ b/geode/src/test/kotlin/com/vitorpamplona/geode/interop/GeodeVsGeodeNegentropySyncTest.kt @@ -20,8 +20,9 @@ */ package com.vitorpamplona.geode.interop -import com.vitorpamplona.geode.LocalRelayServer -import com.vitorpamplona.geode.Relay +import com.vitorpamplona.geode.KtorRelay +import com.vitorpamplona.geode.RelayEngine +import com.vitorpamplona.geode.testing.preload import com.vitorpamplona.quartz.nip01Core.core.Event import com.vitorpamplona.quartz.nip01Core.core.HexKey import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair @@ -72,10 +73,10 @@ import kotlin.test.assertTrue * 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 relayA: RelayEngine + private lateinit var relayB: RelayEngine + private lateinit var serverA: KtorRelay + private lateinit var serverB: KtorRelay private lateinit var scope: CoroutineScope private lateinit var client: NostrClient private val httpClient = OkHttpClient.Builder().build() @@ -84,10 +85,10 @@ class GeodeVsGeodeNegentropySyncTest { 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() + relayA = RelayEngine(url = "ws://127.0.0.1:7771/".normalizeRelayUrl()) + relayB = RelayEngine(url = "ws://127.0.0.1:7772/".normalizeRelayUrl()) + serverA = KtorRelay(relayA, host = "127.0.0.1", port = 0).start() + serverB = KtorRelay(relayB, host = "127.0.0.1", port = 0).start() scope = CoroutineScope(Dispatchers.Default + SupervisorJob()) client = NostrClient(BasicOkHttpWebSocket.Builder { _ -> httpClient }, scope) } diff --git a/geode/src/test/kotlin/com/vitorpamplona/geode/interop/InteropSyncDriver.kt b/geode/src/test/kotlin/com/vitorpamplona/geode/interop/InteropSyncDriver.kt index 1768a0f96..a154ba94b 100644 --- a/geode/src/test/kotlin/com/vitorpamplona/geode/interop/InteropSyncDriver.kt +++ b/geode/src/test/kotlin/com/vitorpamplona/geode/interop/InteropSyncDriver.kt @@ -42,7 +42,7 @@ import kotlin.test.fail * Equivalent of strfry's `test/syncTest.pl` driver for our interop * tests. Drives one round of NIP-77 *negotiation* (NEG-OPEN / * NEG-MSG / NEG-CLOSE) against a real `ws://` endpoint — a Geode - * `LocalRelayServer`, a strfry process, or any other NIP-77 relay. + * `KtorRelay`, a strfry process, or any other NIP-77 relay. * * The driver opens a raw WebSocket — no `NostrClient` overhead — * to keep the wire format under direct control, the same way diff --git a/geode/src/test/kotlin/com/vitorpamplona/geode/perf/LoadBenchmark.kt b/geode/src/test/kotlin/com/vitorpamplona/geode/perf/LoadBenchmark.kt index ced9cfc54..771a12784 100644 --- a/geode/src/test/kotlin/com/vitorpamplona/geode/perf/LoadBenchmark.kt +++ b/geode/src/test/kotlin/com/vitorpamplona/geode/perf/LoadBenchmark.kt @@ -20,8 +20,8 @@ */ package com.vitorpamplona.geode.perf -import com.vitorpamplona.geode.LocalRelayServer -import com.vitorpamplona.geode.Relay +import com.vitorpamplona.geode.KtorRelay +import com.vitorpamplona.geode.RelayEngine import com.vitorpamplona.quartz.nip01Core.core.toHexKey import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient @@ -801,10 +801,10 @@ class LoadBenchmark { } /** Spin up an isolated relay + http client per scenario. */ - private inline fun runBenchmarkServer(block: (LocalRelayServer, OkHttpClient) -> Unit) { + private inline fun runBenchmarkServer(block: (KtorRelay, OkHttpClient) -> Unit) { val placeholder = "ws://127.0.0.1:7771/".normalizeRelayUrl() - val relay = Relay(url = placeholder) - val server = LocalRelayServer(relay, host = "127.0.0.1", port = 0).start() + val relay = RelayEngine(url = placeholder) + val server = KtorRelay(relay, host = "127.0.0.1", port = 0).start() val http = OkHttpClient .Builder() diff --git a/geode/src/test/kotlin/com/vitorpamplona/geode/persistence/PersistenceTest.kt b/geode/src/test/kotlin/com/vitorpamplona/geode/persistence/PersistenceTest.kt index cf6eacdaa..0dd60cea4 100644 --- a/geode/src/test/kotlin/com/vitorpamplona/geode/persistence/PersistenceTest.kt +++ b/geode/src/test/kotlin/com/vitorpamplona/geode/persistence/PersistenceTest.kt @@ -20,7 +20,7 @@ */ package com.vitorpamplona.geode.persistence -import com.vitorpamplona.geode.Relay +import com.vitorpamplona.geode.RelayEngine import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer import com.vitorpamplona.quartz.nip11RelayInfo.Nip11RelayInformation import java.io.File @@ -50,7 +50,7 @@ class PersistenceTest { @Test fun firstBootWritesNothingUntilFirstMutation() { - val relay = Relay(url = url, stateFile = stateFile) + val relay = RelayEngine(url = url, stateFile = stateFile) try { // No mutation yet → file does not exist. assertTrue(!stateFile.exists(), "fresh relay must not eagerly write a snapshot") @@ -62,7 +62,7 @@ class PersistenceTest { @Test fun banPubkeyTriggersSnapshotAndSurvivesRestart() { val pk = "a".repeat(64) - val r1 = Relay(url = url, stateFile = stateFile) + val r1 = RelayEngine(url = url, stateFile = stateFile) try { r1.banStore.banPubkey(pk, "spam") } finally { @@ -71,7 +71,7 @@ class PersistenceTest { assertTrue(stateFile.exists(), "snapshot must be written after a mutation") // Fresh relay reads the snapshot and sees the ban. - val r2 = Relay(url = url, stateFile = stateFile) + val r2 = RelayEngine(url = url, stateFile = stateFile) try { assertTrue(r2.banStore.isBanned(pk)) assertEquals("spam", r2.banStore.listBannedPubkeys()[0].second) @@ -82,14 +82,14 @@ class PersistenceTest { @Test fun updateInfoSurvivesRestart() { - val r1 = Relay(url = url, stateFile = stateFile) + val r1 = RelayEngine(url = url, stateFile = stateFile) try { r1.updateInfo { it.copy(name = "renamed") } } finally { r1.close() } - val r2 = Relay(url = url, stateFile = stateFile) + val r2 = RelayEngine(url = url, stateFile = stateFile) try { assertEquals("renamed", r2.info.document.name) } finally { @@ -99,7 +99,7 @@ class PersistenceTest { @Test fun allowKindRoundTripsAcrossRestart() { - val r1 = Relay(url = url, stateFile = stateFile) + val r1 = RelayEngine(url = url, stateFile = stateFile) try { r1.banStore.allowKind(1) r1.banStore.allowKind(7) @@ -108,7 +108,7 @@ class PersistenceTest { r1.close() } - val r2 = Relay(url = url, stateFile = stateFile) + val r2 = RelayEngine(url = url, stateFile = stateFile) try { assertEquals(listOf(1, 7), r2.banStore.listAllowedKinds()) assertEquals(listOf(4), r2.banStore.listDisallowedKinds()) @@ -121,7 +121,7 @@ class PersistenceTest { fun corruptStateFileIsTolerated() { stateFile.writeText("not valid json {") // Should not throw — just log and start fresh. - val r = Relay(url = url, stateFile = stateFile) + val r = RelayEngine(url = url, stateFile = stateFile) try { assertTrue(r.banStore.listBannedPubkeys().isEmpty()) assertTrue(r.banStore.listAllowedKinds().isEmpty()) @@ -133,7 +133,7 @@ class PersistenceTest { @Test fun snapshotWriteIsAtomicViaTempFile() { // After a mutation completes, no `.tmp` file should remain. - val r = Relay(url = url, stateFile = stateFile) + val r = RelayEngine(url = url, stateFile = stateFile) try { r.banStore.banPubkey("b".repeat(64)) val tmp = File(dir, "admin.json.tmp") @@ -145,7 +145,7 @@ class PersistenceTest { @Test fun missingStateFileMeansInMemoryOnly() { - val r = Relay(url = url) // no stateFile + val r = RelayEngine(url = url) // no stateFile try { r.banStore.banPubkey("c".repeat(64)) // No snapshot path → nothing on disk in our temp dir. diff --git a/geode/src/test/kotlin/com/vitorpamplona/geode/policies/PoliciesIntegrationTest.kt b/geode/src/test/kotlin/com/vitorpamplona/geode/policies/PoliciesIntegrationTest.kt index 0236e2a21..9bdb8e0bc 100644 --- a/geode/src/test/kotlin/com/vitorpamplona/geode/policies/PoliciesIntegrationTest.kt +++ b/geode/src/test/kotlin/com/vitorpamplona/geode/policies/PoliciesIntegrationTest.kt @@ -20,7 +20,7 @@ */ package com.vitorpamplona.geode.policies -import com.vitorpamplona.geode.RelayHub +import com.vitorpamplona.geode.InProcessRelays import com.vitorpamplona.geode.fixtures.SyntheticEvents import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient @@ -44,7 +44,7 @@ import kotlin.test.Test import kotlin.test.assertEquals /** - * End-to-end through `NostrClient → RelayHub → Relay` with the policies + * End-to-end through `NostrClient → InProcessRelays → Relay` with the policies * actually wired into the relay. Proves an EVENT command sent on the * wire surfaces an OK false response when the policy rejects. */ @@ -63,8 +63,8 @@ class PoliciesIntegrationTest { } /** Spin up a hub whose only relay uses the supplied policy factory. */ - private fun hubWith(policyFactory: () -> IRelayPolicy): Pair { - val hub = RelayHub(defaultPolicy = policyFactory) + private fun hubWith(policyFactory: () -> IRelayPolicy): Pair { + val hub = InProcessRelays(defaultPolicy = policyFactory) // Materialise the relay so the URL resolves in the hub. hub.getOrCreate(relayUrl) return NostrClient(hub, scope) to hub diff --git a/geode/src/test/kotlin/com/vitorpamplona/geode/policies/PoliciesTest.kt b/geode/src/test/kotlin/com/vitorpamplona/geode/policies/PoliciesTest.kt index 5e0721b9f..dd7cca7c2 100644 --- a/geode/src/test/kotlin/com/vitorpamplona/geode/policies/PoliciesTest.kt +++ b/geode/src/test/kotlin/com/vitorpamplona/geode/policies/PoliciesTest.kt @@ -36,7 +36,7 @@ import kotlin.test.fail * single hit, collision between allow + deny, etc.). * * The end-to-end "policy is applied through the Ktor server" coverage - * lives in `LocalRelayServerTest` / `Nip01ComplianceTest` — these tests + * lives in `KtorRelayTest` / `Nip01ComplianceTest` — these tests * just exercise the policy in isolation. */ class PoliciesTest { diff --git a/geode/src/testFixtures/kotlin/com/vitorpamplona/geode/testing/RelayClientTest.kt b/geode/src/testFixtures/kotlin/com/vitorpamplona/geode/testing/RelayClientTest.kt index 76112f2b3..1c67e4734 100644 --- a/geode/src/testFixtures/kotlin/com/vitorpamplona/geode/testing/RelayClientTest.kt +++ b/geode/src/testFixtures/kotlin/com/vitorpamplona/geode/testing/RelayClientTest.kt @@ -20,8 +20,8 @@ */ package com.vitorpamplona.geode.testing -import com.vitorpamplona.geode.Relay -import com.vitorpamplona.geode.RelayHub +import com.vitorpamplona.geode.InProcessRelays +import com.vitorpamplona.geode.RelayEngine import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl import kotlinx.coroutines.CoroutineScope @@ -32,7 +32,7 @@ import org.junit.After /** * Base class for tests that drive a real [NostrClient] against an - * in-process [RelayHub]. Owns the lifecycle of the four pieces every + * in-process [InProcessRelays]. Owns the lifecycle of the four pieces every * such test needs: * * - [hub] — the registry of in-process relays (also serves as @@ -59,15 +59,15 @@ import org.junit.After * ``` */ open class RelayClientTest { - val hub: RelayHub = RelayHub() + val hub: InProcessRelays = InProcessRelays() val scope: CoroutineScope = CoroutineScope(Dispatchers.Default + SupervisorJob()) val client: NostrClient = NostrClient(hub, scope) - /** Stable URL for the single-relay case — see [RelayHub.DEFAULT_URL]. */ - val defaultRelayUrl: NormalizedRelayUrl get() = RelayHub.DEFAULT_URL + /** Stable URL for the single-relay case — see [InProcessRelays.DEFAULT_URL]. */ + val defaultRelayUrl: NormalizedRelayUrl get() = InProcessRelays.DEFAULT_URL /** Lazy handle to the relay at [defaultRelayUrl]. Auto-created on first read. */ - val defaultRelay: Relay get() = hub.getOrCreate(defaultRelayUrl) + val defaultRelay: RelayEngine get() = hub.getOrCreate(defaultRelayUrl) @After fun tearDownRelayClientTest() { diff --git a/geode/src/testFixtures/kotlin/com/vitorpamplona/geode/testing/RelayEngineTestExtensions.kt b/geode/src/testFixtures/kotlin/com/vitorpamplona/geode/testing/RelayEngineTestExtensions.kt new file mode 100644 index 000000000..0e100496a --- /dev/null +++ b/geode/src/testFixtures/kotlin/com/vitorpamplona/geode/testing/RelayEngineTestExtensions.kt @@ -0,0 +1,65 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.geode.testing + +import com.vitorpamplona.geode.RelayEngine +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper +import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.EventCmd +import kotlinx.coroutines.CompletableDeferred + +/** + * Inserts events directly into the underlying store, bypassing the wire protocol. + * + * Use this for **pre-test setup** — events that exist before any client connects. + * It does NOT broadcast to active subscriptions. For sending events that should + * fan out to live subscribers (post-EOSE), use [publish] instead. + */ +suspend fun RelayEngine.preload(events: Iterable) { + events.forEach { store.insert(it) } +} + +/** @see preload */ +suspend fun RelayEngine.preload(vararg events: Event) = preload(events.toList()) + +/** + * Publishes an event through the relay's session machinery so it both lands + * in the store and fans out to active subscriptions matching its filters + * (mirrors what a real client would do via an `EVENT` command). + * + * Suspends until the relay's `OK` (or `NOTICE`) reply lands, i.e. until + * the [com.vitorpamplona.quartz.nip01Core.relay.server.IngestQueue] drain + * loop has processed the event and fanned it out to active subscribers. + * Tests that publish-then-subscribe rely on this ordering: otherwise the + * fire-and-forget submit lets a subscription register *after* publish + * returns but *before* fanout runs, and ephemeral kinds (not persisted) + * still leak to the late subscriber. + */ +suspend fun RelayEngine.publish(event: Event) { + val replied = CompletableDeferred() + val session = server.connect { replied.complete(Unit) } + try { + session.receive(OptimizedJsonMapper.toJson(EventCmd(event))) + replied.await() + } finally { + session.close() + } +}