From e78561af675529ebfc35204a5179f8f5d69c88d4 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 7 May 2026 12:31:27 +0000 Subject: [PATCH] refactor(relay): split overgrown files + reduce duplication MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Audit follow-ups, no behavior change. - Centralize NIPs/name/version constants in RelayInfo so RelayConfig and the default doc share one source of truth. - Extract NegSessionRegistry (NIP-77 state + open/msg/close) out of RelaySession; the connection class now routes commands. - Move multi-filter snapshot union/dedupe onto LiveEventStore. - Pull Nip86HttpRoute and WebSocketSessionPump out of LocalRelayServer (was 485 lines, three responsibilities). - Collapse Nip86Server.dispatch repetition with withHex/withHexAndReason/ withInt/withString helpers; reuse Hex.isHex64 instead of a local regex. - Make Nip11RelayInformation (and nested types) data classes so Nip86Server uses the synthesized copy() directly — drops the hand-rolled field-by-field shim. --- .../quartz/relay/LocalRelayServer.kt | 251 ++---------------- .../vitorpamplona/quartz/relay/RelayInfo.kt | 43 ++- .../quartz/relay/admin/Nip86Server.kt | 166 +++++------- .../quartz/relay/config/RelayConfig.kt | 15 +- .../quartz/relay/server/Nip86HttpRoute.kt | 181 +++++++++++++ .../relay/server/WebSocketSessionPump.kt | 114 ++++++++ .../nip01Core/relay/server/LiveEventStore.kt | 18 ++ .../relay/server/NegSessionRegistry.kt | 115 ++++++++ .../nip01Core/relay/server/RelaySession.kt | 98 +------ .../nip11RelayInfo/Nip11RelayInformation.kt | 10 +- 10 files changed, 565 insertions(+), 446 deletions(-) create mode 100644 quartz-relay/src/main/kotlin/com/vitorpamplona/quartz/relay/server/Nip86HttpRoute.kt create mode 100644 quartz-relay/src/main/kotlin/com/vitorpamplona/quartz/relay/server/WebSocketSessionPump.kt create mode 100644 quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NegSessionRegistry.kt diff --git a/quartz-relay/src/main/kotlin/com/vitorpamplona/quartz/relay/LocalRelayServer.kt b/quartz-relay/src/main/kotlin/com/vitorpamplona/quartz/relay/LocalRelayServer.kt index ae6b989b8..3c54b4c27 100644 --- a/quartz-relay/src/main/kotlin/com/vitorpamplona/quartz/relay/LocalRelayServer.kt +++ b/quartz-relay/src/main/kotlin/com/vitorpamplona/quartz/relay/LocalRelayServer.kt @@ -21,13 +21,12 @@ package com.vitorpamplona.quartz.relay import com.vitorpamplona.quartz.nip01Core.core.HexKey -import com.vitorpamplona.quartz.nip01Core.core.JsonMapper import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.NoticeMessage import com.vitorpamplona.quartz.nip01Core.relay.server.RelaySession -import com.vitorpamplona.quartz.nip86RelayManagement.rpc.Nip86Request -import com.vitorpamplona.quartz.nip86RelayManagement.rpc.Nip86Response import com.vitorpamplona.quartz.relay.admin.Nip86Server import com.vitorpamplona.quartz.relay.admin.Nip98AuthVerifier +import com.vitorpamplona.quartz.relay.server.Nip86HttpRoute +import com.vitorpamplona.quartz.relay.server.WebSocketSessionPump import io.ktor.http.ContentType import io.ktor.http.HttpHeaders import io.ktor.http.HttpStatusCode @@ -36,18 +35,12 @@ import io.ktor.server.cio.CIO import io.ktor.server.cio.CIOApplicationEngine import io.ktor.server.engine.embeddedServer import io.ktor.server.request.header -import io.ktor.server.request.receiveChannel import io.ktor.server.response.respondText import io.ktor.server.routing.get import io.ktor.server.routing.post import io.ktor.server.routing.routing import io.ktor.server.websocket.WebSockets import io.ktor.server.websocket.webSocket -import io.ktor.utils.io.readAvailable -import io.ktor.websocket.Frame -import io.ktor.websocket.readText -import kotlinx.coroutines.channels.consumeEach -import kotlinx.coroutines.launch import kotlinx.coroutines.runBlocking import java.util.concurrent.ConcurrentHashMap @@ -114,10 +107,6 @@ class LocalRelayServer( */ val maxAdminBodyBytes: Int = 1 shl 20, ) { - /** - * Bridges the relay's mutable [RelayInfo] to [Nip86Server.InfoHolder] - * so admin RPCs can rewrite the NIP-11 doc atomically. - */ private val infoHolder = object : Nip86Server.InfoHolder { override fun get() = relay.info @@ -127,10 +116,18 @@ class LocalRelayServer( } } - private val nip86 = Nip86Server(banStore = relay.banStore, infoHolder = infoHolder, store = relay.store) - private val nip98 = Nip98AuthVerifier() + private val nip86Server = Nip86Server(banStore = relay.banStore, infoHolder = infoHolder, store = relay.store) + private val nip86Route = + Nip86HttpRoute( + server = nip86Server, + verifier = Nip98AuthVerifier(), + allowList = adminPubkeys.mapTo(HashSet()) { it.lowercase() }, + maxBodyBytes = maxAdminBodyBytes, + signedUrlFor = { call -> + publicUrl ?: ("http://" + (call.request.header(HttpHeaders.Host) ?: "$host:$resolvedPort") + path) + }, + ) - private val adminAllowList: Set = adminPubkeys.mapTo(HashSet()) { it.lowercase() } private var engine: CIOApplicationEngine? = null private var resolvedPort: Int = -1 @@ -197,67 +194,18 @@ class LocalRelayServer( // NIP-86: POST application/nostr+json+rpc with a NIP-98 // signed Authorization header → JSON-RPC dispatch. post(path) { - handleNip86(call) + nip86Route.handle(call) } webSocket(path) { if (shuttingDown) { // Just return — Ktor closes the WS for us. - // We can't `close(reason)` here because the - // CIO engine's outgoing channel may already - // be torn down during shutdown. return@webSocket } - // Per-session outbound queue. The relay's - // `connect` callback runs on whatever thread - // produced the message — it can't suspend, so - // we hand off to a dedicated writer coroutine - // that does suspend on `outgoing.send` and thus - // applies real backpressure on slow clients. - // When the queue fills, that's a slow consumer - // — drop the connection cleanly so subscribers - // don't silently miss EVENT/EOSE. - val outQueue = - kotlinx.coroutines.channels - .Channel(capacity = SESSION_OUTGOING_BUFFER) - val writerJob = - launch { - try { - for (json in outQueue) { - outgoing.send(Frame.Text(json)) - } - } catch (_: kotlinx.coroutines.channels.ClosedSendChannelException) { - // socket closed — let the handler's - // finally block run normal teardown - } - } - var droppedForBackpressure = false - val session = - relay.server.connect { json -> - val res = outQueue.trySend(json) - if (!res.isSuccess && !res.isClosed) { - // Buffer is full → slow client. - // Mark + close the queue; the - // writer drains, then we let the - // outer handler's finally close - // the WS session. - droppedForBackpressure = true - outQueue.close() - } - } - activeSessions.add(session) - try { - incoming.consumeEach { frame -> - if (droppedForBackpressure) return@consumeEach - if (frame is Frame.Text) { - session.receive(frame.readText()) - } - } - } finally { - outQueue.close() - writerJob.cancel() - activeSessions.remove(session) - session.close() - } + WebSocketSessionPump(this).pump( + server = relay.server, + registerSession = activeSessions::add, + unregisterSession = activeSessions::remove, + ) } } } @@ -309,151 +257,6 @@ class LocalRelayServer( resolvedPort = -1 } - /** - * Handles a NIP-86 admin RPC request: - * 1. 403 if no admin pubkey list is configured (endpoint disabled). - * 2. 401 if the NIP-98 Authorization header is missing/invalid. - * 3. 403 if the verified pubkey isn't in [adminAllowList]. - * 4. 400 if the body isn't a valid Nip86Request. - * 5. 200 with a Nip86Response JSON body otherwise. - * - * `application/nostr+json+rpc` is the wire content type prescribed - * by NIP-86; we send it on responses and accept any body on the - * request (the auth event's payload-hash already binds the body). - */ - private suspend fun handleNip86(call: io.ktor.server.application.ApplicationCall) { - if (adminAllowList.isEmpty()) { - call.respondText( - "NIP-86 management API is not enabled on this relay.", - ContentType.Text.Plain, - HttpStatusCode.Forbidden, - ) - return - } - - // Cap the body BEFORE we read it. We have to read the bytes - // (NIP-98 payload-hash binds them), but unauthenticated - // attackers shouldn't be able to stream gigabytes here. - val declared = call.request.headers[HttpHeaders.ContentLength]?.toLongOrNull() - if (declared != null && declared > maxAdminBodyBytes) { - call.respondText( - "request body exceeds $maxAdminBodyBytes-byte cap", - ContentType.Text.Plain, - HttpStatusCode.PayloadTooLarge, - ) - return - } - val body = readBoundedBody(call, maxAdminBodyBytes) ?: return - - val authHeader = call.request.header(HttpHeaders.Authorization) - // The URL the client signed must match the relay's CANONICAL - // public URL — not whatever `Host` header reaches us. An - // attacker can spoof `Host`, and behind TLS termination the - // verifier would compare against the wrong scheme. Operators - // configure [publicUrl] explicitly. The Host fallback is for - // local loopback unit tests only and is documented as unsafe. - val signedUrl = - publicUrl - ?: ("http://" + (call.request.header(HttpHeaders.Host) ?: "$host:$resolvedPort") + path) - val verification = nip98.verify(authHeader, method = "POST", url = signedUrl, body = body) - - val pubkey = - when (verification) { - is Nip98AuthVerifier.Result.Verified -> { - verification.pubkey - } - - Nip98AuthVerifier.Result.Missing -> { - call.response.headers.append(HttpHeaders.WWWAuthenticate, Nip98AuthVerifier.SCHEME.trim()) - call.respondText( - "missing Authorization header (NIP-98)", - ContentType.Text.Plain, - HttpStatusCode.Unauthorized, - ) - return - } - - is Nip98AuthVerifier.Result.Malformed -> { - call.respondText( - "invalid NIP-98 Authorization: ${verification.reason}", - ContentType.Text.Plain, - HttpStatusCode.Unauthorized, - ) - return - } - } - - if (pubkey.lowercase() !in adminAllowList) { - call.respondText( - "pubkey is not on the admin list", - ContentType.Text.Plain, - HttpStatusCode.Forbidden, - ) - return - } - - val req = - try { - JsonMapper.fromJson(body.decodeToString()) - } catch (e: Exception) { - call.respondText( - "invalid Nip86Request: ${e.message ?: e::class.simpleName}", - ContentType.Text.Plain, - HttpStatusCode.BadRequest, - ) - return - } - - val response: Nip86Response = nip86.dispatch(req) - - // Audit log: structured single line so an operator can grep - // "nip86" / pubkey / method without a logging framework - // dependency. Keep it best-effort — System.err is already what - // the rest of Main.kt uses, and a missing log line shouldn't - // fail the response. - runCatching { - System.err.println( - "nip86 audit pubkey=$pubkey method=${req.method} ok=${response.error == null}" + - (response.error?.let { " error=$it" } ?: ""), - ) - } - - call.respondText( - JsonMapper.toJson(response), - ContentType.parse("application/nostr+json+rpc"), - HttpStatusCode.OK, - ) - } - - /** - * Reads up to [maxBytes] bytes from the request body and returns - * them. If the stream produces more than [maxBytes] (i.e. a - * lying or absent `Content-Length`), responds 413 and returns - * `null` — caller stops handling. - */ - private suspend fun readBoundedBody( - call: io.ktor.server.application.ApplicationCall, - maxBytes: Int, - ): ByteArray? { - val ch = call.receiveChannel() - val buf = ByteArray(maxBytes + 1) - var pos = 0 - while (pos <= maxBytes) { - val read = ch.readAvailable(buf, pos, buf.size - pos) - if (read <= 0) break - pos += read - } - if (pos > maxBytes) { - call.respondText( - "request body exceeds $maxBytes-byte cap", - ContentType.Text.Plain, - HttpStatusCode.PayloadTooLarge, - ) - return null - } - return buf.copyOfRange(0, pos) - } - /** * Best-effort NOTICE to every active client. Failures are * swallowed — a flaky socket on its way out is exactly the case @@ -466,20 +269,4 @@ class LocalRelayServer( runCatching { session.send(notice) } } } - - companion object { - /** - * Per-session outbound buffer size. When a slow client falls - * this many frames behind, we close their connection rather - * than silently dropping further frames (which would corrupt - * NIP-01 by missing EVENT/EOSE messages). - * - * Sized to hold fan-out for a connection holding several - * thousand subscriptions when one event matches all of them - * — the realistic upper bound for a relay client. At ~250B - * per frame this caps per-session memory at ~2 MiB before - * we drop the connection. - */ - const val SESSION_OUTGOING_BUFFER: Int = 8192 - } } diff --git a/quartz-relay/src/main/kotlin/com/vitorpamplona/quartz/relay/RelayInfo.kt b/quartz-relay/src/main/kotlin/com/vitorpamplona/quartz/relay/RelayInfo.kt index 418650de9..34943a7b1 100644 --- a/quartz-relay/src/main/kotlin/com/vitorpamplona/quartz/relay/RelayInfo.kt +++ b/quartz-relay/src/main/kotlin/com/vitorpamplona/quartz/relay/RelayInfo.kt @@ -38,21 +38,42 @@ data class RelayInfo( val json: String by lazy { JsonMapper.toJson(document) } companion object { + const val NAME = "quartz-relay" + const val DESCRIPTION = "Embedded Nostr relay from the Amethyst quartz library." + const val SOFTWARE = "https://github.com/vitorpamplona/amethyst/tree/main/quartz-relay" + const val VERSION = "1.08.0" + + /** + * NIPs this relay implements out of the box. Single source of + * truth — both [default] and [com.vitorpamplona.quartz.relay.config.RelayConfig.resolveInfo] + * consult this list. Add a NIP here when its handler is wired + * into [com.vitorpamplona.quartz.nip01Core.relay.server.RelaySession] + * (or in this module's policy stack). + * + * Currently: + * - 1 NIP-01 basic + * - 9 NIP-09 deletion (DeletionRequestModule) + * - 11 NIP-11 this doc + * - 40 NIP-40 expiration (ExpirationModule) + * - 42 NIP-42 AUTH (when policy enables) + * - 45 NIP-45 COUNT + * - 50 NIP-50 search (SQLite FTS) + * - 62 NIP-62 right to vanish + * - 77 NIP-77 negentropy reconciliation + * - 86 NIP-86 relay management API (when admin pubkeys configured) + */ + 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. */ fun default(url: NormalizedRelayUrl): RelayInfo = RelayInfo( Nip11RelayInformation( - name = "quartz-relay", - description = "Embedded Nostr relay from the Amethyst quartz library.", - software = "https://github.com/vitorpamplona/amethyst/tree/main/quartz-relay", - version = "1.08.0", - // Currently implemented: NIP-01 (basic), NIP-09 (deletion via - // DeletionRequestModule), NIP-11 (this doc), NIP-40 (expiration - // via ExpirationModule), NIP-42 (AUTH — when policy enables), - // NIP-45 (COUNT), NIP-50 (search via FTS), NIP-62 (right to vanish), - // NIP-77 (negentropy reconciliation), NIP-86 (relay management API - // — when admin pubkeys are configured). - supported_nips = listOf("1", "9", "11", "40", "42", "45", "50", "62", "77", "86"), + name = NAME, + description = DESCRIPTION, + software = SOFTWARE, + version = VERSION, + supported_nips = SUPPORTED_NIPS, ), ) diff --git a/quartz-relay/src/main/kotlin/com/vitorpamplona/quartz/relay/admin/Nip86Server.kt b/quartz-relay/src/main/kotlin/com/vitorpamplona/quartz/relay/admin/Nip86Server.kt index 8324f338e..7d03babef 100644 --- a/quartz-relay/src/main/kotlin/com/vitorpamplona/quartz/relay/admin/Nip86Server.kt +++ b/quartz-relay/src/main/kotlin/com/vitorpamplona/quartz/relay/admin/Nip86Server.kt @@ -30,6 +30,8 @@ import com.vitorpamplona.quartz.nip86RelayManagement.rpc.Nip86Method import com.vitorpamplona.quartz.nip86RelayManagement.rpc.Nip86Request import com.vitorpamplona.quartz.nip86RelayManagement.rpc.Nip86Response import com.vitorpamplona.quartz.relay.RelayInfo +import com.vitorpamplona.quartz.utils.Hex +import kotlinx.coroutines.CancellationException import kotlinx.serialization.KSerializer import kotlinx.serialization.builtins.ListSerializer import kotlinx.serialization.json.Json @@ -101,112 +103,71 @@ class Nip86Server( runCatching { when (req.method) { Nip86Method.SUPPORTED_METHODS -> { - result(buildJsonArray { supportedMethods.forEach { add(JsonPrimitive(it)) } }) + ok(buildJsonArray { supportedMethods.forEach { add(JsonPrimitive(it)) } }) } Nip86Method.BAN_PUBKEY -> { - val (pk, reason) = req.params.stringPair() ?: return malformed("expected [pubkey, reason?]") - if (!isHex64(pk)) return malformed("pubkey must be 64-char hex") - banStore.banPubkey(pk, reason) - result(JsonPrimitive(true)) + withHexAndReason(req, "pubkey") { pk, reason -> banStore.banPubkey(pk, reason) } } Nip86Method.UNBAN_PUBKEY -> { - val (pk, _) = req.params.stringPair() ?: return malformed("expected [pubkey]") - if (!isHex64(pk)) return malformed("pubkey must be 64-char hex") - banStore.unbanPubkey(pk) - result(JsonPrimitive(true)) + withHex(req, "pubkey") { pk -> banStore.unbanPubkey(pk) } } Nip86Method.LIST_BANNED_PUBKEYS -> { - result( - banStore - .listBannedPubkeys() - .map { (pk, r) -> BannedPubkey(pk, r) } - .toJsonArray(BannedPubkey.serializer()), - ) + ok(banStore.listBannedPubkeys().map { (pk, r) -> BannedPubkey(pk, r) }.toJsonArray(BannedPubkey.serializer())) } Nip86Method.ALLOW_PUBKEY -> { - val (pk, reason) = req.params.stringPair() ?: return malformed("expected [pubkey, reason?]") - if (!isHex64(pk)) return malformed("pubkey must be 64-char hex") - banStore.allowPubkey(pk, reason) - result(JsonPrimitive(true)) + withHexAndReason(req, "pubkey") { pk, reason -> banStore.allowPubkey(pk, reason) } } Nip86Method.UNALLOW_PUBKEY -> { - val (pk, _) = req.params.stringPair() ?: return malformed("expected [pubkey]") - if (!isHex64(pk)) return malformed("pubkey must be 64-char hex") - banStore.unallowPubkey(pk) - result(JsonPrimitive(true)) + withHex(req, "pubkey") { pk -> banStore.unallowPubkey(pk) } } Nip86Method.LIST_ALLOWED_PUBKEYS -> { - result( - banStore - .listAllowedPubkeys() - .map { (pk, r) -> AllowedPubkey(pk, r) } - .toJsonArray(AllowedPubkey.serializer()), - ) + ok(banStore.listAllowedPubkeys().map { (pk, r) -> AllowedPubkey(pk, r) }.toJsonArray(AllowedPubkey.serializer())) } Nip86Method.BAN_EVENT -> { - val (id, reason) = req.params.stringPair() ?: return malformed("expected [event_id, reason?]") - if (!isHex64(id)) return malformed("event_id must be 64-char hex") - banStore.banEvent(id, reason) - // Also remove the event from the store if it's there. - store?.delete(Filter(ids = listOf(id))) - result(JsonPrimitive(true)) + withHexAndReason(req, "event_id") { id, reason -> + banStore.banEvent(id, reason) + // Also remove the event from the store if present. + store?.delete(Filter(ids = listOf(id))) + } } Nip86Method.ALLOW_EVENT -> { - val (id, _) = req.params.stringPair() ?: return malformed("expected [event_id]") - if (!isHex64(id)) return malformed("event_id must be 64-char hex") - banStore.allowEvent(id) - result(JsonPrimitive(true)) + withHex(req, "event_id") { id -> banStore.allowEvent(id) } } Nip86Method.LIST_BANNED_EVENTS -> { - result( - banStore - .listBannedEvents() - .map { (id, r) -> BannedEvent(id, r) } - .toJsonArray(BannedEvent.serializer()), - ) + ok(banStore.listBannedEvents().map { (id, r) -> BannedEvent(id, r) }.toJsonArray(BannedEvent.serializer())) } Nip86Method.ALLOW_KIND -> { - val k = req.params.firstInt() ?: return malformed("expected [kind]") - banStore.allowKind(k) - result(JsonPrimitive(true)) + withInt(req, "kind") { k -> banStore.allowKind(k) } } Nip86Method.DISALLOW_KIND -> { - val k = req.params.firstInt() ?: return malformed("expected [kind]") - banStore.disallowKind(k) - result(JsonPrimitive(true)) + withInt(req, "kind") { k -> banStore.disallowKind(k) } } Nip86Method.LIST_ALLOWED_KINDS -> { - result(buildJsonArray { banStore.listAllowedKinds().forEach { add(JsonPrimitive(it)) } }) + ok(buildJsonArray { banStore.listAllowedKinds().forEach { add(JsonPrimitive(it)) } }) } Nip86Method.CHANGE_RELAY_NAME -> { - val name = req.params.firstString() ?: return malformed("expected [name]") - rewriteInfo { it.copy(name = name) } - result(JsonPrimitive(true)) + withString(req, "name") { name -> rewriteInfo { it.copy(name = name) } } } Nip86Method.CHANGE_RELAY_DESCRIPTION -> { - val desc = req.params.firstString() ?: return malformed("expected [description]") - rewriteInfo { it.copy(description = desc) } - result(JsonPrimitive(true)) + withString(req, "description") { desc -> rewriteInfo { it.copy(description = desc) } } } Nip86Method.CHANGE_RELAY_ICON -> { - val icon = req.params.firstString() ?: return malformed("expected [icon_url]") - rewriteInfo { it.copy(icon = icon) } - result(JsonPrimitive(true)) + withString(req, "icon_url") { icon -> rewriteInfo { it.copy(icon = icon) } } } else -> { @@ -217,50 +178,63 @@ class Nip86Server( // CancellationException must propagate so structured // concurrency works — swallowing it would let a parent // cancellation be reported as a benign RPC error. - if (e is kotlinx.coroutines.CancellationException) throw e + if (e is CancellationException) throw e Nip86Response(error = "internal: ${e.message ?: e::class.simpleName}") } + private inline fun withHex( + req: Nip86Request, + label: String, + action: (String) -> Unit, + ): Nip86Response { + val (value, _) = req.params.stringPair() ?: return malformed("expected [$label]") + if (!Hex.isHex64(value)) return malformed("$label must be 64-char hex") + action(value) + return okTrue + } + + private suspend inline fun withHexAndReason( + req: Nip86Request, + label: String, + action: suspend (String, String?) -> Unit, + ): Nip86Response { + val (value, reason) = req.params.stringPair() ?: return malformed("expected [$label, reason?]") + if (!Hex.isHex64(value)) return malformed("$label must be 64-char hex") + action(value, reason) + return okTrue + } + + private inline fun withInt( + req: Nip86Request, + label: String, + action: (Int) -> Unit, + ): Nip86Response { + val v = req.params.firstInt() ?: return malformed("expected [$label]") + action(v) + return okTrue + } + + private inline fun withString( + req: Nip86Request, + label: String, + action: (String) -> Unit, + ): Nip86Response { + val v = req.params.firstString() ?: return malformed("expected [$label]") + action(v) + return okTrue + } + private fun rewriteInfo(transform: (Nip11RelayInformation) -> Nip11RelayInformation) { val current = infoHolder.get().document infoHolder.set(RelayInfo(transform(current))) } - - /** [Nip11RelayInformation] is not a `data class`; do a manual field-by-field copy. */ - private fun Nip11RelayInformation.copy( - name: String? = this.name, - description: String? = this.description, - icon: String? = this.icon, - ) = Nip11RelayInformation( - id = this.id, - name = name, - description = description, - icon = icon, - pubkey = this.pubkey, - self = this.self, - contact = this.contact, - supported_nips = this.supported_nips, - supported_nip_extensions = this.supported_nip_extensions, - software = this.software, - version = this.version, - limitation = this.limitation, - relay_countries = this.relay_countries, - language_tags = this.language_tags, - tags = this.tags, - posting_policy = this.posting_policy, - privacy_policy = this.privacy_policy, - terms_of_service = this.terms_of_service, - payments_url = this.payments_url, - retention = this.retention, - fees = this.fees, - nip50 = this.nip50, - supported_grasps = this.supported_grasps, - ) } private fun malformed(reason: String) = Nip86Response(error = "invalid params: $reason") -private fun result(j: JsonElement) = Nip86Response(result = j, error = null) +private fun ok(j: JsonElement) = Nip86Response(result = j, error = null) + +private val okTrue = ok(JsonPrimitive(true)) private val rpcJson = Json { encodeDefaults = false } @@ -280,7 +254,3 @@ private fun JsonArray.firstInt(): Int? = }.getOrNull() private fun JsonPrimitive.contentOrNull(): String? = if (this == JsonNull) null else content - -private val HEX64 = Regex("[0-9a-fA-F]{64}") - -private fun isHex64(s: String): Boolean = HEX64.matches(s) diff --git a/quartz-relay/src/main/kotlin/com/vitorpamplona/quartz/relay/config/RelayConfig.kt b/quartz-relay/src/main/kotlin/com/vitorpamplona/quartz/relay/config/RelayConfig.kt index 9762c667a..e1f6d83b3 100644 --- a/quartz-relay/src/main/kotlin/com/vitorpamplona/quartz/relay/config/RelayConfig.kt +++ b/quartz-relay/src/main/kotlin/com/vitorpamplona/quartz/relay/config/RelayConfig.kt @@ -51,21 +51,16 @@ data class RelayConfig( fun resolveInfo(advertisedUrl: NormalizedRelayUrl): RelayInfo = RelayInfo( Nip11RelayInformation( - name = info.name ?: "quartz-relay", - description = info.description ?: "Embedded Nostr relay from the Amethyst quartz library.", + name = info.name ?: RelayInfo.NAME, + description = info.description ?: RelayInfo.DESCRIPTION, pubkey = info.pubkey, contact = info.contact, icon = info.icon, - software = - info.software - ?: "https://github.com/vitorpamplona/amethyst/tree/main/quartz-relay", - version = info.version ?: "1.08.0", + software = info.software ?: RelayInfo.SOFTWARE, + version = info.version ?: RelayInfo.VERSION, supported_nips = info.supported_nips?.map(Int::toString) - // Keep in sync with `RelayInfo.default()` — - // both lists must reflect the NIPs actually - // wired into the relay. - ?: listOf("1", "9", "11", "40", "42", "45", "50", "62", "77", "86"), + ?: RelayInfo.SUPPORTED_NIPS, privacy_policy = info.privacy_policy, terms_of_service = info.terms_of_service, relay_countries = info.relay_countries, diff --git a/quartz-relay/src/main/kotlin/com/vitorpamplona/quartz/relay/server/Nip86HttpRoute.kt b/quartz-relay/src/main/kotlin/com/vitorpamplona/quartz/relay/server/Nip86HttpRoute.kt new file mode 100644 index 000000000..e73d1653e --- /dev/null +++ b/quartz-relay/src/main/kotlin/com/vitorpamplona/quartz/relay/server/Nip86HttpRoute.kt @@ -0,0 +1,181 @@ +/* + * 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.quartz.relay.server + +import com.vitorpamplona.quartz.nip01Core.core.HexKey +import com.vitorpamplona.quartz.nip01Core.core.JsonMapper +import com.vitorpamplona.quartz.nip86RelayManagement.rpc.Nip86Request +import com.vitorpamplona.quartz.nip86RelayManagement.rpc.Nip86Response +import com.vitorpamplona.quartz.relay.admin.Nip86Server +import com.vitorpamplona.quartz.relay.admin.Nip98AuthVerifier +import io.ktor.http.ContentType +import io.ktor.http.HttpHeaders +import io.ktor.http.HttpStatusCode +import io.ktor.server.application.ApplicationCall +import io.ktor.server.request.header +import io.ktor.server.request.receiveChannel +import io.ktor.server.response.respondText +import io.ktor.utils.io.readAvailable + +/** + * NIP-86 admin POST handler. Owns the gating order: + * 1. 403 if no admin pubkey list is configured (endpoint disabled). + * 2. 413 if body exceeds [maxBodyBytes] (declared or actual). + * 3. 401 if the NIP-98 Authorization header is missing/invalid. + * 4. 403 if the verified pubkey isn't in [allowList]. + * 5. 400 if the body isn't a valid Nip86Request. + * 6. 200 with a Nip86Response JSON body otherwise. + * + * The [signedUrlFor] callback resolves what URL the client must have + * signed in their NIP-98 token. Operators configure the canonical + * `publicUrl`; loopback tests fall back to the request's `Host` + * header. We pass it as a callback rather than a string so the route + * doesn't need to know about Ktor request internals. + */ +internal class Nip86HttpRoute( + private val server: Nip86Server, + private val verifier: Nip98AuthVerifier, + private val allowList: Set, + private val maxBodyBytes: Int, + private val signedUrlFor: (ApplicationCall) -> String, +) { + suspend fun handle(call: ApplicationCall) { + if (allowList.isEmpty()) { + call.respondText( + "NIP-86 management API is not enabled on this relay.", + ContentType.Text.Plain, + HttpStatusCode.Forbidden, + ) + return + } + + val body = readBoundedBody(call) ?: return + val pubkey = verifyAuth(call, body) ?: return + if (pubkey.lowercase() !in allowList) { + call.respondText( + "pubkey is not on the admin list", + ContentType.Text.Plain, + HttpStatusCode.Forbidden, + ) + return + } + + val req = + try { + JsonMapper.fromJson(body.decodeToString()) + } catch (e: Exception) { + call.respondText( + "invalid Nip86Request: ${e.message ?: e::class.simpleName}", + ContentType.Text.Plain, + HttpStatusCode.BadRequest, + ) + return + } + + val response: Nip86Response = server.dispatch(req) + audit(pubkey, req, response) + call.respondText( + JsonMapper.toJson(response), + ContentType.parse("application/nostr+json+rpc"), + HttpStatusCode.OK, + ) + } + + private suspend fun readBoundedBody(call: ApplicationCall): ByteArray? { + val declared = call.request.headers[HttpHeaders.ContentLength]?.toLongOrNull() + if (declared != null && declared > maxBodyBytes) { + call.respondText( + "request body exceeds $maxBodyBytes-byte cap", + ContentType.Text.Plain, + HttpStatusCode.PayloadTooLarge, + ) + return null + } + val ch = call.receiveChannel() + val buf = ByteArray(maxBodyBytes + 1) + var pos = 0 + while (pos <= maxBodyBytes) { + val read = ch.readAvailable(buf, pos, buf.size - pos) + if (read <= 0) break + pos += read + } + if (pos > maxBodyBytes) { + call.respondText( + "request body exceeds $maxBodyBytes-byte cap", + ContentType.Text.Plain, + HttpStatusCode.PayloadTooLarge, + ) + return null + } + return buf.copyOfRange(0, pos) + } + + private suspend fun verifyAuth( + call: ApplicationCall, + body: ByteArray, + ): HexKey? { + val header = call.request.header(HttpHeaders.Authorization) + val verification = verifier.verify(header, method = "POST", url = signedUrlFor(call), body = body) + return when (verification) { + is Nip98AuthVerifier.Result.Verified -> { + verification.pubkey + } + + Nip98AuthVerifier.Result.Missing -> { + call.response.headers.append(HttpHeaders.WWWAuthenticate, Nip98AuthVerifier.SCHEME.trim()) + call.respondText( + "missing Authorization header (NIP-98)", + ContentType.Text.Plain, + HttpStatusCode.Unauthorized, + ) + null + } + + is Nip98AuthVerifier.Result.Malformed -> { + call.respondText( + "invalid NIP-98 Authorization: ${verification.reason}", + ContentType.Text.Plain, + HttpStatusCode.Unauthorized, + ) + null + } + } + } + + /** + * Audit log: structured single line so an operator can grep + * "nip86" / pubkey / method without a logging framework + * dependency. Best-effort — a missing log line shouldn't fail + * the response. + */ + private fun audit( + pubkey: HexKey, + req: Nip86Request, + response: Nip86Response, + ) { + runCatching { + System.err.println( + "nip86 audit pubkey=$pubkey method=${req.method} ok=${response.error == null}" + + (response.error?.let { " error=$it" } ?: ""), + ) + } + } +} diff --git a/quartz-relay/src/main/kotlin/com/vitorpamplona/quartz/relay/server/WebSocketSessionPump.kt b/quartz-relay/src/main/kotlin/com/vitorpamplona/quartz/relay/server/WebSocketSessionPump.kt new file mode 100644 index 000000000..952e6131f --- /dev/null +++ b/quartz-relay/src/main/kotlin/com/vitorpamplona/quartz/relay/server/WebSocketSessionPump.kt @@ -0,0 +1,114 @@ +/* + * 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.quartz.relay.server + +import com.vitorpamplona.quartz.nip01Core.relay.server.NostrServer +import com.vitorpamplona.quartz.nip01Core.relay.server.RelaySession +import io.ktor.server.websocket.DefaultWebSocketServerSession +import io.ktor.websocket.Frame +import io.ktor.websocket.readText +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.channels.ClosedSendChannelException +import kotlinx.coroutines.channels.consumeEach +import kotlinx.coroutines.launch + +/** + * Per-WebSocket pump that owns the bounded outbound queue and the + * writer coroutine. Pulled out of `LocalRelayServer` so that file + * stays focused on Ktor wiring; the slow-client / backpressure + * policy now lives next to the data structures it manages. + * + * Lifecycle: + * 1. `connect(server, registerSession)` opens a [RelaySession], + * registers it with the supplied callback, and starts the + * writer coroutine that drains [outQueue] into [outgoing]. + * 2. `pump()` reads inbound frames until the socket closes. + * 3. `finally`-style teardown closes the queue, cancels the + * writer, unregisters the session, and closes it. + * + * Slow-client policy: when [outQueue] fills, [SESSION_OUTGOING_BUFFER] + * frames behind, the connection is dropped rather than silently + * losing EVENT/EOSE — silent drop would corrupt NIP-01. + */ +internal class WebSocketSessionPump( + private val ws: DefaultWebSocketServerSession, +) { + private val outQueue = Channel(capacity = SESSION_OUTGOING_BUFFER) + private var droppedForBackpressure = false + + suspend fun pump( + server: NostrServer, + registerSession: (RelaySession) -> Unit, + unregisterSession: (RelaySession) -> Unit, + ) { + val writerJob = + ws.launch { + try { + for (json in outQueue) { + ws.outgoing.send(Frame.Text(json)) + } + } catch (_: ClosedSendChannelException) { + // socket closed — outer handler runs normal teardown. + } + } + val session = + server.connect { json -> + val res = outQueue.trySend(json) + if (!res.isSuccess && !res.isClosed) { + // Buffer is full → slow client. Mark + close the + // queue; the writer drains, then the outer handler + // closes the WS session. + droppedForBackpressure = true + outQueue.close() + } + } + registerSession(session) + try { + ws.incoming.consumeEach { frame -> + if (droppedForBackpressure) return@consumeEach + if (frame is Frame.Text) { + session.receive(frame.readText()) + } + } + } finally { + outQueue.close() + writerJob.cancel() + unregisterSession(session) + session.close() + } + } + + companion object { + /** + * Per-session outbound buffer size. When a slow client falls + * this many frames behind, we close their connection rather + * than silently dropping further frames (which would corrupt + * NIP-01 by missing EVENT/EOSE messages). + * + * Sized to hold fan-out for a connection holding several + * thousand subscriptions when one event matches all of them + * — the realistic upper bound for a relay client. At ~250B + * per frame this caps per-session memory at ~2 MiB before + * we drop the connection. + */ + const val SESSION_OUTGOING_BUFFER: Int = 8192 + } +} diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/LiveEventStore.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/LiveEventStore.kt index 43543b1e5..00d2436f9 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/LiveEventStore.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/LiveEventStore.kt @@ -103,4 +103,22 @@ class LiveEventStore( * moment the NEG-OPEN arrives, not a streamed/live result. */ suspend fun snapshotQuery(filter: Filter): List = store.query(filter) + + /** + * Multi-filter snapshot. Unions the per-filter results and + * deduplicates by event id so an event matching N filters is + * yielded once. Used by NIP-77 NEG-OPEN when the policy stack + * rewrote the single incoming filter into several. + */ + suspend fun snapshotQuery(filters: List): List { + if (filters.size == 1) return snapshotQuery(filters[0]) + val seen = HashSet() + val merged = ArrayList() + for (f in filters) { + for (e in store.query(f)) { + if (seen.add(e.id)) merged += e + } + } + return merged + } } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NegSessionRegistry.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NegSessionRegistry.kt new file mode 100644 index 000000000..f85b723d0 --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NegSessionRegistry.kt @@ -0,0 +1,115 @@ +/* + * 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.quartz.nip01Core.relay.server + +import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message +import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.ReqCmd +import com.vitorpamplona.quartz.nip77Negentropy.NegCloseCmd +import com.vitorpamplona.quartz.nip77Negentropy.NegErrMessage +import com.vitorpamplona.quartz.nip77Negentropy.NegMsgCmd +import com.vitorpamplona.quartz.nip77Negentropy.NegOpenCmd +import com.vitorpamplona.quartz.nip77Negentropy.NegentropyServerSession + +/** + * Per-connection NIP-77 negentropy state and dispatch. + * + * Owns the map of active reconciliation sessions keyed by NEG-OPEN + * subId, and the open/msg/close handlers. Pulled out of [RelaySession] + * so the connection class only routes commands while this class owns + * the negentropy lifecycle and error mapping. + * + * Plain [HashMap] is sufficient because the registry is mutated only + * from [RelaySession.receive] — that path is single-threaded per the + * WebSocket handler contract. + */ +class NegSessionRegistry( + private val store: LiveEventStore, + private val send: (Message) -> Unit, +) { + private val sessions = HashMap() + + /** + * Open a reconciliation session. The relay snapshots its matching + * events at this instant — concurrent inserts during the sync are + * not surfaced; clients re-open if they want fresh state. + * + * Access control reuses the REQ policy hook: a relay that requires + * AUTH or has kind/pubkey allow-deny lists applies the same rules + * to NEG-OPEN as it does to subscription REQs. + */ + suspend fun open( + cmd: NegOpenCmd, + policy: IRelayPolicy, + ) { + val gate = policy.accept(ReqCmd(cmd.subId, listOf(cmd.filter))) + if (gate is PolicyResult.Rejected) { + send(NegErrMessage(cmd.subId, gate.reason)) + return + } + val filters = (gate as PolicyResult.Accepted).cmd.filters + + // NIP-77: same-subId OPEN replaces any prior session. + sessions.remove(cmd.subId) + + val events = store.snapshotQuery(filters) + val session = NegentropyServerSession(cmd.subId, events) + sessions[cmd.subId] = session + + runMessage(cmd.subId, session) { it.processMessage(cmd.initialMessage) } + } + + fun msg(cmd: NegMsgCmd) { + val session = sessions[cmd.subId] + if (session == null) { + send(NegErrMessage(cmd.subId, "error: no negentropy session for ${cmd.subId}")) + return + } + runMessage(cmd.subId, session) { it.processMessage(cmd.message) } + } + + /** + * Spec: clients send NEG-CLOSE to free server-side state. + * Silent no-op if the session is unknown — there's no authoritative + * error response in NIP-77 for an unknown close. + */ + fun close(cmd: NegCloseCmd) { + sessions.remove(cmd.subId) + } + + /** Dropped on `RelaySession.cancelAllSubscriptions`. */ + fun clear() { + sessions.clear() + } + + private inline fun runMessage( + subId: String, + session: NegentropyServerSession, + block: (NegentropyServerSession) -> Message?, + ) { + try { + val response = block(session) + if (response != null) send(response) + } catch (e: Exception) { + sessions.remove(subId) + send(NegErrMessage(subId, "error: ${e.message ?: e::class.simpleName}")) + } + } +} diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt index 9baf769a9..d72ff491c 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt @@ -35,12 +35,11 @@ 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.nip77Negentropy.NegCloseCmd -import com.vitorpamplona.quartz.nip77Negentropy.NegErrMessage import com.vitorpamplona.quartz.nip77Negentropy.NegMsgCmd import com.vitorpamplona.quartz.nip77Negentropy.NegOpenCmd -import com.vitorpamplona.quartz.nip77Negentropy.NegentropyServerSession import com.vitorpamplona.quartz.utils.Log import com.vitorpamplona.quartz.utils.cache.LargeCache +import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Job import kotlinx.coroutines.launch @@ -58,13 +57,8 @@ class RelaySession( ) : AutoCloseable { private val subscriptions = LargeCache() - /** - * NIP-77 negentropy reconciliation sessions, keyed by NEG-OPEN - * subId. Plain hash map here (not [LargeCache]) because it's - * mutated only from the single-threaded `receive()` path — - * RelaySession.receive is serialised by the WebSocket handler. - */ - private val negSessions = HashMap() + /** NIP-77 negentropy state for this connection. */ + private val negentropy = NegSessionRegistry(store, ::send) private fun addSubscription( subId: String, @@ -80,7 +74,7 @@ class RelaySession( fun cancelAllSubscriptions() { subscriptions.forEach { _, job -> job.cancel() } subscriptions.clear() - negSessions.clear() + negentropy.clear() } fun send(message: Message) { @@ -121,9 +115,9 @@ class RelaySession( is ReqCmd -> handleReq(cmd) is CloseCmd -> handleClose(cmd) is CountCmd -> handleCount(cmd) - is NegOpenCmd -> handleNegOpen(cmd) - is NegMsgCmd -> handleNegMsg(cmd) - is NegCloseCmd -> handleNegClose(cmd) + is NegOpenCmd -> negentropy.open(cmd, policy) + is NegMsgCmd -> negentropy.msg(cmd) + is NegCloseCmd -> negentropy.close(cmd) else -> send(NoticeMessage("error: unsupported command ${cmd.label()}")) } } @@ -195,7 +189,7 @@ class RelaySession( }, onEose = { send(EoseMessage(cmd.subId)) }, ) - } catch (_: kotlinx.coroutines.CancellationException) { + } catch (_: CancellationException) { // Subscription was closed – this is expected. } } @@ -211,82 +205,6 @@ class RelaySession( } } - // -- NIP-77: NEG-OPEN ----------------------------------------------------- - - /** - * Open a negentropy reconciliation session. The relay snapshots its - * matching events at this instant — concurrent inserts during the - * sync are not surfaced; clients re-open if they want fresh state. - * - * Access control reuses the REQ policy hook: a relay that requires - * AUTH or has kind/pubkey allow-deny lists applies the same rules - * to NEG-OPEN as it does to subscription REQs. - */ - private suspend fun handleNegOpen(cmd: NegOpenCmd) { - // Run the same access controls as REQ would. - val asReq = ReqCmd(cmd.subId, listOf(cmd.filter)) - val gate = policy.accept(asReq) - if (gate is PolicyResult.Rejected) { - send(NegErrMessage(cmd.subId, gate.reason)) - return - } - val filters = (gate as PolicyResult.Accepted).cmd.filters - - // Drop any prior session at this subId (NIP-77: same-subId - // OPEN replaces). - negSessions.remove(cmd.subId) - - val events = - if (filters.size == 1) { - store.snapshotQuery(filters[0]) - } else { - // Multiple filters: union the snapshots and dedupe by id. - val seen = HashSet() - val merged = mutableListOf() - for (f in filters) { - for (e in store.snapshotQuery(f)) { - if (seen.add(e.id)) merged += e - } - } - merged - } - - val neg = NegentropyServerSession(cmd.subId, events) - negSessions[cmd.subId] = neg - - try { - val response = neg.processMessage(cmd.initialMessage) - if (response != null) send(response) - } catch (e: Exception) { - negSessions.remove(cmd.subId) - send(NegErrMessage(cmd.subId, "error: ${e.message ?: e::class.simpleName}")) - } - } - - // -- NIP-77: NEG-MSG ------------------------------------------------------ - private fun handleNegMsg(cmd: NegMsgCmd) { - val neg = negSessions[cmd.subId] - if (neg == null) { - send(NegErrMessage(cmd.subId, "error: no negentropy session for ${cmd.subId}")) - return - } - try { - val response = neg.processMessage(cmd.message) - if (response != null) send(response) - } catch (e: Exception) { - negSessions.remove(cmd.subId) - send(NegErrMessage(cmd.subId, "error: ${e.message ?: e::class.simpleName}")) - } - } - - // -- NIP-77: NEG-CLOSE ---------------------------------------------------- - private fun handleNegClose(cmd: NegCloseCmd) { - // Spec: clients send NEG-CLOSE to free server-side state. - // Silent no-op if the session is unknown — there's no authoritative - // error response in NIP-77 for an unknown close. - negSessions.remove(cmd.subId) - } - init { policy.onConnect(::send) } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip11RelayInfo/Nip11RelayInformation.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip11RelayInfo/Nip11RelayInformation.kt index 2478fcaea..9558d8998 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip11RelayInfo/Nip11RelayInformation.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip11RelayInfo/Nip11RelayInformation.kt @@ -27,7 +27,7 @@ import kotlinx.serialization.Serializable @Stable @Serializable -class Nip11RelayInformation( +data class Nip11RelayInformation( val id: String? = null, val name: String? = null, val description: String? = null, @@ -59,7 +59,7 @@ class Nip11RelayInformation( @Stable @Serializable - class RelayInformationFee( + data class RelayInformationFee( val amount: Int? = null, val unit: String? = null, val period: Int? = null, @@ -68,7 +68,7 @@ class Nip11RelayInformation( @Stable @Serializable - class RelayInformationFees( + data class RelayInformationFees( val admission: List? = null, val subscription: List? = null, val publication: List? = null, @@ -76,7 +76,7 @@ class Nip11RelayInformation( @Stable @Serializable - class RelayInformationLimitation( + data class RelayInformationLimitation( val max_message_length: Int? = null, val max_subscriptions: Int? = null, val max_filters: Int? = null, @@ -96,7 +96,7 @@ class Nip11RelayInformation( @Stable @Serializable - class RelayInformationRetentionData( + data class RelayInformationRetentionData( val kinds: ArrayList? = null, val time: Int? = null, val count: Int? = null,