nick/amethyst
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

refactor(geode): rename relay classes for clarity, move test helpers off public API

Browse Source 
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) <noreply@anthropic.com>
This commit is contained in:
Vitor Pamplona
2026-05-12 14:44:33 -04:00
parent dd12938a0f
commit 9f7deee8f3
25 changed files with 176 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files
geode/plans/2026-05-07-connection-scaling.md
+1 -1
View File
@@ -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
geode/plans/2026-05-07-event-ingestion-batching.md
+1 -1
View File
@@ -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
geode/src/main/kotlin/com/vitorpamplona/geode/RelayHub.kt → geode/src/main/kotlin/com/vitorpamplona/geode/InProcessRelays.kt
+9 -9
View File
@@ -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<NormalizedRelayUrl, Relay>()
private val relays = ConcurrentHashMap<NormalizedRelayUrl, RelayEngine>()
@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<NormalizedRelayUrl> = relays.keys.toSet()
geode/src/main/kotlin/com/vitorpamplona/geode/LocalRelayServer.kt → geode/src/main/kotlin/com/vitorpamplona/geode/KtorRelay.kt
+6 -6
View File
@@ -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.
geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt
+2 -2
View File
@@ -117,7 +117,7 @@ fun main(args: Array<String>) {
maxSessionsPerConnection = config.negentropy.max_sessions_per_connection,
)
val relay =
Relay(
RelayEngine(
advertisedUrl,
store,
info,
@@ -132,7 +132,7 @@ fun main(args: Array<String>) {
val frameLimit =
(config.limits.max_ws_frame_bytes ?: config.limits.max_ws_message_bytes)?.toLong()
val server =
LocalRelayServer(
KtorRelay(
relay,
host = host,
port = port,
geode/src/main/kotlin/com/vitorpamplona/geode/Relay.kt → geode/src/main/kotlin/com/vitorpamplona/geode/RelayEngine.kt
+3 -34
View File
@@ -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<Event>) {
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()
}
geode/src/main/kotlin/com/vitorpamplona/geode/RelayInfo.kt
+1 -1
View File
@@ -65,7 +65,7 @@ data class RelayInfo(
val SUPPORTED_NIPS: List<String> =
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(
geode/src/main/kotlin/com/vitorpamplona/geode/config/RelayConfig.kt
+1 -1
View File
@@ -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.
geode/src/main/kotlin/com/vitorpamplona/geode/server/WebSocketSessionPump.kt
+1 -1
View File
@@ -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.
*
geode/src/test/kotlin/com/vitorpamplona/geode/GracefulShutdownTest.kt
+5 -5
View File
@@ -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)
geode/src/test/kotlin/com/vitorpamplona/geode/LocalRelayServerTest.kt → geode/src/test/kotlin/com/vitorpamplona/geode/KtorRelayTest.kt
+16 -13
View File
@@ -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")
}
}
geode/src/test/kotlin/com/vitorpamplona/geode/Nip01ComplianceTest.kt
+2
View File
@@ -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
geode/src/test/kotlin/com/vitorpamplona/geode/Nip09DeletionTest.kt
+3 -3
View File
@@ -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)
}
geode/src/test/kotlin/com/vitorpamplona/geode/Nip40ExpirationTest.kt
+2 -2
View File
@@ -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)
}
geode/src/test/kotlin/com/vitorpamplona/geode/Nip62VanishTest.kt
+2 -2
View File
@@ -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)
}
geode/src/test/kotlin/com/vitorpamplona/geode/Nip77NegentropyTest.kt
+7 -6
View File
@@ -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<String> = 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))
geode/src/test/kotlin/com/vitorpamplona/geode/admin/Nip86EndToEndTest.kt
+9 -9
View File
@@ -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 =
geode/src/test/kotlin/com/vitorpamplona/geode/interop/GeodeVsGeodeNegentropySyncTest.kt
+11 -10
View File
@@ -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)
}
geode/src/test/kotlin/com/vitorpamplona/geode/interop/InteropSyncDriver.kt
+1 -1
View File
@@ -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
geode/src/test/kotlin/com/vitorpamplona/geode/perf/LoadBenchmark.kt
+5 -5
View File
@@ -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()
geode/src/test/kotlin/com/vitorpamplona/geode/persistence/PersistenceTest.kt
+11 -11
View File
@@ -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.
geode/src/test/kotlin/com/vitorpamplona/geode/policies/PoliciesIntegrationTest.kt
+4 -4
View File
@@ -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<NostrClient, RelayHub> {
val hub = RelayHub(defaultPolicy = policyFactory)
private fun hubWith(policyFactory: () -> IRelayPolicy): Pair<NostrClient, InProcessRelays> {
val hub = InProcessRelays(defaultPolicy = policyFactory)
// Materialise the relay so the URL resolves in the hub.
hub.getOrCreate(relayUrl)
return NostrClient(hub, scope) to hub
geode/src/test/kotlin/com/vitorpamplona/geode/policies/PoliciesTest.kt
+1 -1
View File
@@ -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 {
geode/src/testFixtures/kotlin/com/vitorpamplona/geode/testing/RelayClientTest.kt
+7 -7
View File
@@ -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() {
geode/src/testFixtures/kotlin/com/vitorpamplona/geode/testing/RelayEngineTestExtensions.kt
+65
View File
@@ -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<Event>) {
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<Unit>()
val session = server.connect { replied.complete(Unit) }
try {
session.receive(OptimizedJsonMapper.toJson(EventCmd(event)))
replied.await()
} finally {
session.close()
}
}