From df88cab92df6985effb822958552e188c19dc3f7 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 30 Mar 2026 12:35:27 +0000 Subject: [PATCH] refactor: simplify relay API with AutoCloseable and serve() helper - NostrServer and IEventStore implement AutoCloseable for .use {} support - Add NostrServer.serve() to handle session lifecycle automatically - IRelayPolicy + operator now returns PolicyStack instead of List - Deprecate shutdown() in favor of close() - Update RELAY.md guide to use simplified API patterns https://claude.ai/code/session_013oL9PkQaFyNQHKVg2vw9qs --- quartz/RELAY.md | 150 +++++++++--------- .../nip01Core/relay/server/IRelayPolicy.kt | 3 +- .../nip01Core/relay/server/NostrServer.kt | 33 +++- .../quartz/nip01Core/store/IEventStore.kt | 4 +- .../relay/server/NostrServerAuthTest.kt | 28 ++-- .../nip01Core/relay/server/NostrServerTest.kt | 22 +-- 6 files changed, 130 insertions(+), 110 deletions(-) diff --git a/quartz/RELAY.md b/quartz/RELAY.md index 88233d26f..dcad2bc25 100644 --- a/quartz/RELAY.md +++ b/quartz/RELAY.md @@ -15,6 +15,8 @@ Quartz provides a complete, transport-agnostic relay engine: `NostrServer` doesn't know about HTTP or WebSockets. You provide a `send` callback per connection, and it gives you a `RelaySession` that accepts raw JSON strings. This makes it trivial to plug into Ktor (or any other transport). +Both `NostrServer` and `EventStore` implement `AutoCloseable`, so you can use Kotlin's `.use {}` for automatic resource cleanup. + ## Quick Start ### 1. Add Dependencies @@ -59,21 +61,15 @@ fun main() { routing { webSocket("/") { - // 4. Register this WebSocket as a relay connection - val session = server.connect { json -> - launch { send(Frame.Text(json)) } - } - - try { - // 5. Forward incoming frames to the relay session + // 4. Serve this WebSocket connection (auto-cleanup on disconnect) + server.serve( + send = { json -> launch { send(Frame.Text(json)) } }, + ) { session -> for (frame in incoming) { if (frame is Frame.Text) { session.receive(frame.readText()) } } - } finally { - // 6. Clean up when the client disconnects - session.close() } } } @@ -83,6 +79,17 @@ fun main() { That's it. You now have a NIP-01 compliant relay running on `ws://localhost:7777`. +The `serve` method creates a `RelaySession`, passes it to your block, and automatically closes the session when the block completes (normally or via exception). No `try`/`finally` needed. + +> **Tip:** If you need direct control over the session lifecycle, use `connect()` with `.use {}`: +> ```kotlin +> server.connect { json -> launch { send(Frame.Text(json)) } }.use { session -> +> for (frame in incoming) { +> if (frame is Frame.Text) session.receive(frame.readText()) +> } +> } +> ``` + ## How It Works ### Connection Lifecycle @@ -91,7 +98,7 @@ That's it. You now have a NIP-01 compliant relay running on `ws://localhost:7777 Client connects via WebSocket │ ▼ -server.connect { json -> send(json) } +server.serve(send) { session -> ... } │ ▼ RelaySession created @@ -106,7 +113,7 @@ server.connect { json -> send(json) } └───────────────────────────────┘ │ ▼ -Client disconnects → session.close() +Client disconnects → session auto-closed │ ▼ All subscriptions cancelled @@ -233,11 +240,20 @@ Validates: - Timestamp is within 10 minutes - Event signature is valid -### Composing Policies with `PolicyStack` +### Composing Policies -Chain multiple policies together. All must approve; the first rejection wins. Policies run in order and can rewrite commands for downstream policies. +Chain multiple policies using the `+` operator or `PolicyStack`. All must approve; the first rejection wins. Policies run in order and can rewrite commands for downstream policies. ```kotlin +// Using the + operator +val server = NostrServer( + store = store, + policyBuilder = { + VerifyPolicy + FullAuthPolicy(relay = "wss://myrelay.example.com/".normalizeRelayUrl()!!) + }, +) + +// Or using PolicyStack for three or more policies val server = NostrServer( store = store, policyBuilder = { @@ -283,10 +299,7 @@ Use it: val server = NostrServer( store = store, policyBuilder = { - PolicyStack( - VerifyPolicy, - KindWhitelistPolicy(allowedKinds = setOf(0, 1, 3, 7, 30023)), - ) + VerifyPolicy + KindWhitelistPolicy(allowedKinds = setOf(0, 1, 3, 7, 30023)) }, ) ``` @@ -307,7 +320,6 @@ val server = NostrServer( ```kotlin import com.vitorpamplona.quartz.nip01Core.relay.normalizer.normalizeRelayUrl import com.vitorpamplona.quartz.nip01Core.relay.server.NostrServer -import com.vitorpamplona.quartz.nip01Core.relay.server.policies.PolicyStack import com.vitorpamplona.quartz.nip01Core.relay.server.policies.VerifyPolicy import com.vitorpamplona.quartz.nip01Core.store.sqlite.DefaultIndexingStrategy import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore @@ -318,61 +330,43 @@ import io.ktor.server.routing.* import io.ktor.server.websocket.* import io.ktor.websocket.* import kotlinx.coroutines.launch -import java.time.Duration fun main() { - val relayUrl = "wss://myrelay.example.com/" - val store = EventStore( dbName = "relay-events.db", - relay = relayUrl.normalizeRelayUrl(), + relay = "wss://myrelay.example.com/".normalizeRelayUrl(), indexStrategy = DefaultIndexingStrategy( indexEventsByCreatedAtAlone = true, ), ) - val server = NostrServer( + // NostrServer implements AutoCloseable — .use {} ensures clean shutdown + NostrServer( store = store, - policyBuilder = { - PolicyStack( - VerifyPolicy, - // Add your custom policies here - ) - }, - ) + policyBuilder = { VerifyPolicy }, + ).use { server -> + embeddedServer(Netty, port = 7777) { + install(WebSockets) { + pingPeriodMillis = 30_000 + timeoutMillis = 60_000 + maxFrameSize = Long.MAX_VALUE + } - embeddedServer(Netty, port = 7777) { - install(WebSockets) { - pingPeriodMillis = 30_000 - timeoutMillis = 60_000 - maxFrameSize = Long.MAX_VALUE - } - - routing { - webSocket("/") { - val session = server.connect { json -> - launch { send(Frame.Text(json)) } - } - - try { - for (frame in incoming) { - when (frame) { - is Frame.Text -> session.receive(frame.readText()) - else -> { /* ignore binary, ping, pong */ } + routing { + webSocket("/") { + server.serve( + send = { json -> launch { send(Frame.Text(json)) } }, + ) { session -> + for (frame in incoming) { + if (frame is Frame.Text) { + session.receive(frame.readText()) + } } } - } finally { - session.close() } } - } - }.start(wait = true) - - // Clean shutdown - Runtime.getRuntime().addShutdownHook(Thread { - server.shutdown() - store.close() - }) + }.start(wait = true) + } } ``` @@ -388,7 +382,7 @@ fun main() { │ ┌────────────────────────────────────────┐ │ │ │ NostrServer │ │ │ │ │ │ -│ │ connect(send) ──► RelaySession │ │ +│ │ serve(send) { session -> ... } │ │ │ │ │ │ │ │ │ ├─ IRelayPolicy │ │ │ │ │ (validate) │ │ @@ -433,31 +427,29 @@ class MyRelayTest { @Test fun clientCanPublishAndSubscribe() = runTest { val dispatcher = UnconfinedTestDispatcher(testScheduler) - val store = EventStore(null) // in-memory - val server = NostrServer( - store = store, + + NostrServer( + store = EventStore(null), policyBuilder = { EmptyPolicy }, parentContext = dispatcher, - ) + ).use { server -> + // Collect messages sent to this client + val messages = mutableListOf() + val session = server.connect { messages.add(it) } - // Collect messages sent to this client - val messages = mutableListOf() - val session = server.connect { messages.add(it) } + // Publish an event + session.receive("""["EVENT",{"id":"${"0".repeat(64)}","pubkey":"${"a".repeat(64)}","created_at":1000,"kind":1,"tags":[],"content":"hello","sig":"${"b".repeat(128)}"}]""") - // Publish an event - session.receive("""["EVENT",{"id":"${"0".repeat(64)}","pubkey":"${"a".repeat(64)}","created_at":1000,"kind":1,"tags":[],"content":"hello","sig":"${"b".repeat(128)}"}]""") + // Verify OK response + assertTrue(messages.any { it.contains("OK") }) - // Verify OK response - assertTrue(messages.any { it.contains("OK") }) + // Subscribe to kind 1 + session.receive("""["REQ","sub1",{"kinds":[1]}]""") - // Subscribe to kind 1 - session.receive("""["REQ","sub1",{"kinds":[1]}]""") - - // Verify we get the event back + EOSE - assertTrue(messages.any { it.contains("EVENT") }) - assertTrue(messages.any { it.contains("EOSE") }) - - server.shutdown() + // Verify we get the event back + EOSE + assertTrue(messages.any { it.contains("EVENT") }) + assertTrue(messages.any { it.contains("EOSE") }) + } } } ``` diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/IRelayPolicy.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/IRelayPolicy.kt index 113824d62..249ac7c75 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/IRelayPolicy.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/IRelayPolicy.kt @@ -27,6 +27,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.Command import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.CountCmd import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.EventCmd import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.ReqCmd +import com.vitorpamplona.quartz.nip01Core.relay.server.policies.PolicyStack /** * Defines custom behavior for this relay. @@ -76,7 +77,7 @@ interface IRelayPolicy { */ fun canSendToSession(event: Event): Boolean = true - operator fun plus(other: IRelayPolicy) = listOf(this, other) + operator fun plus(other: IRelayPolicy): IRelayPolicy = PolicyStack(this, other) } sealed interface PolicyResult { diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServer.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServer.kt index 78651df5e..f72c9e7b2 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServer.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServer.kt @@ -41,7 +41,7 @@ class NostrServer( private val store: IEventStore, private val policyBuilder: () -> IRelayPolicy = { VerifyPolicy }, private val parentContext: CoroutineContext = SupervisorJob(), -) { +) : AutoCloseable { private val subStore = LiveEventStore(store) /** Scope for all subscriptions. */ @@ -70,11 +70,38 @@ class NostrServer( } /** - * Shuts down the server, cancelling all subscriptions and sessions. + * Registers a new client connection and serves it for the duration of + * [incoming]. The session is automatically closed when [incoming] returns. + * + * @param send Callback the server uses to send JSON messages to this client. + * @param incoming Suspend block that yields raw JSON strings from the client + * (e.g., reading WebSocket text frames in a loop). */ - fun shutdown() { + suspend fun serve( + send: (String) -> Unit, + incoming: suspend (RelaySession) -> Unit, + ) { + val session = connect(send) + try { + incoming(session) + } finally { + session.close() + } + } + + /** + * Shuts down the server, cancelling all subscriptions and closing the store. + */ + override fun close() { connections.forEach { _, session -> session.cancelAllSubscriptions() } connections.clear() scope.cancel() + store.close() } + + /** + * Shuts down the server, cancelling all subscriptions and closing the store. + */ + @Deprecated("Use close() instead", replaceWith = ReplaceWith("close()")) + fun shutdown() = close() } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/IEventStore.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/IEventStore.kt index 1d3d74efd..e9268922d 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/IEventStore.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/IEventStore.kt @@ -23,7 +23,7 @@ package com.vitorpamplona.quartz.nip01Core.store import com.vitorpamplona.quartz.nip01Core.core.Event import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter -interface IEventStore { +interface IEventStore : AutoCloseable { fun insert(event: Event) interface ITransaction { @@ -56,5 +56,5 @@ interface IEventStore { fun deleteExpiredEvents() - fun close() + override fun close() } diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServerAuthTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServerAuthTest.kt index bc89df39f..4e7200643 100644 --- a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServerAuthTest.kt +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServerAuthTest.kt @@ -142,7 +142,7 @@ class NostrServerAuthTest { assertEquals(1, collector.messages.size) assertTrue(collector.messages[0].contains("\"AUTH\"")) - server.shutdown() + server.close() } @Test @@ -167,7 +167,7 @@ class NostrServerAuthTest { assertTrue((session.policy as FullAuthPolicy).isAuthenticated()) assertTrue(session.policy.authenticatedUsers.contains(pubkey)) - server.shutdown() + server.close() } @Test @@ -188,7 +188,7 @@ class NostrServerAuthTest { assertTrue(okMessages[0].contains("challenge")) assertFalse((session.policy as FullAuthPolicy).isAuthenticated()) - server.shutdown() + server.close() } @Test @@ -213,7 +213,7 @@ class NostrServerAuthTest { assertTrue(okMessages[0].contains("relay url")) assertFalse((session.policy as FullAuthPolicy).isAuthenticated()) - server.shutdown() + server.close() } @Test @@ -242,7 +242,7 @@ class NostrServerAuthTest { assertTrue(okMessages[0].contains("created_at")) assertFalse((session.policy as FullAuthPolicy).isAuthenticated()) - server.shutdown() + server.close() } @Test @@ -281,7 +281,7 @@ class NostrServerAuthTest { assertTrue(okMessages[0].contains("could not parse message")) assertFalse((session.policy as FullAuthPolicy).isAuthenticated()) - server.shutdown() + server.close() } @Test @@ -315,7 +315,7 @@ class NostrServerAuthTest { assertTrue(authedPubkeys.contains(pubkey)) assertTrue(authedPubkeys.contains(pubkey2)) - server.shutdown() + server.close() } // -- NIP-42: requireAuth --------------------------------------------------- @@ -337,7 +337,7 @@ class NostrServerAuthTest { assertTrue(okMessages[0].contains("\"false\"")) assertTrue(okMessages[0].contains("auth-required:")) - server.shutdown() + server.close() } @Test @@ -354,7 +354,7 @@ class NostrServerAuthTest { assertEquals(1, closedMessages.size) assertTrue(closedMessages[0].contains("auth-required:")) - server.shutdown() + server.close() } @Test @@ -371,7 +371,7 @@ class NostrServerAuthTest { assertEquals(1, closedMessages.size) assertTrue(closedMessages[0].contains("auth-required:")) - server.shutdown() + server.close() } @Test @@ -410,7 +410,7 @@ class NostrServerAuthTest { val eoseMessages = collector.rawMessagesContaining("EOSE") assertTrue(eoseMessages.isNotEmpty()) - server.shutdown() + server.close() } @Test @@ -430,7 +430,7 @@ class NostrServerAuthTest { assertEquals(1, okMessages.size) assertTrue(okMessages[0].contains("\"true\"")) - server.shutdown() + server.close() } // -- Custom AuthPolicy tests ----------------------------------------------- @@ -471,7 +471,7 @@ class NostrServerAuthTest { assertTrue(okMessages[1].contains("\"false\"")) assertTrue(okMessages[1].contains("auth-required:")) - server.shutdown() + server.close() } @Test @@ -531,6 +531,6 @@ class NostrServerAuthTest { assertEquals(1, events.size) assertEquals(pubkey, events[0].event.pubKey) - server.shutdown() + server.close() } } diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServerTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServerTest.kt index 165bfb0e4..3e033cf49 100644 --- a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServerTest.kt +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServerTest.kt @@ -115,7 +115,7 @@ class NostrServerTest { val stored = store.query(Filter(ids = listOf(event.id))) assertEquals(1, stored.size) - server.shutdown() + server.close() } @Test @@ -138,7 +138,7 @@ class NostrServerTest { assertTrue(okMessages[0].contains("\"true\"")) assertTrue(okMessages[1].contains("\"false\"")) - server.shutdown() + server.close() } // -- REQ command ----------------------------------------------------------- @@ -172,7 +172,7 @@ class NostrServerTest { // Events should be newest first assertTrue(events[0].event.createdAt >= events[1].event.createdAt) - server.shutdown() + server.close() } @Test @@ -195,7 +195,7 @@ class NostrServerTest { val events = collector.parsedEventMessages().filterIsInstance() assertEquals(3, events.size) - server.shutdown() + server.close() } // -- Live subscription ----------------------------------------------------- @@ -226,7 +226,7 @@ class NostrServerTest { assertTrue(newMessages.isNotEmpty()) assertTrue(newMessages[0].contains("\"EVENT\"")) - server.shutdown() + server.close() } @Test @@ -251,7 +251,7 @@ class NostrServerTest { assertEquals(countAfterEose, collector1.messages.size) - server.shutdown() + server.close() } // -- CLOSE command --------------------------------------------------------- @@ -282,7 +282,7 @@ class NostrServerTest { assertEquals(countAfterClose, collector1.messages.size) - server.shutdown() + server.close() } @Test @@ -318,7 +318,7 @@ class NostrServerTest { assertTrue(newMessages[0].contains("\"EVENT\"")) assertTrue(newMessages[0].contains(hexId(2))) - server.shutdown() + server.close() } // -- COUNT command (NIP-45) ------------------------------------------------ @@ -344,7 +344,7 @@ class NostrServerTest { assertEquals(1, countMessages.size) assertTrue(countMessages[0].contains("\"count\":2")) - server.shutdown() + server.close() } // -- Disconnect ------------------------------------------------------------ @@ -374,7 +374,7 @@ class NostrServerTest { assertEquals(countAfterDisconnect, collector1.messages.size) assertEquals(2, collector2.messages.size) - server.shutdown() + server.close() } // -- Invalid messages ------------------------------------------------------ @@ -392,6 +392,6 @@ class NostrServerTest { assertEquals(1, collector.messages.size) assertTrue(collector.messages[0].contains("NOTICE")) - server.shutdown() + server.close() } }