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
2026-03-30 12:35:27 +00:00
parent e556f7716c
commit df88cab92d
6 changed files with 130 additions and 110 deletions
@@ -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<String>()
+            val session = server.connect { messages.add(it) }

-        // Collect messages sent to this client
-        val messages = mutableListOf<String>()
-        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") })
+        }
    }
 }
 ```
@@ -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<T : Command> {
@@ -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()
 }
@@ -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()
 }
@@ -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()
        }
 }
@@ -115,7 +115,7 @@ class NostrServerTest {
            val stored = store.query<Event>(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<EventMessage>()
            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()
        }
 }