refactor(relay): split overgrown files + reduce duplication

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.
2026-05-07 12:31:27 +00:00
parent c440186575
commit e78561af67
10 changed files with 565 additions and 446 deletions
@@ -103,4 +103,22 @@ class LiveEventStore(
     * moment the NEG-OPEN arrives, not a streamed/live result.
     */
    suspend fun snapshotQuery(filter: Filter): List<Event> = 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<Filter>): List<Event> {
+        if (filters.size == 1) return snapshotQuery(filters[0])
+        val seen = HashSet<String>()
+        val merged = ArrayList<Event>()
+        for (f in filters) {
+            for (e in store.query<Event>(f)) {
+                if (seen.add(e.id)) merged += e
+            }
+        }
+        return merged
+    }
 }
@@ -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<String, NegentropyServerSession>()
+
+    /**
+     * 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}"))
+        }
+    }
+}
@@ -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<String, Job>()

-    /**
-     * 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<String, NegentropyServerSession>()
+    /** 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<String>()
-                val merged = mutableListOf<com.vitorpamplona.quartz.nip01Core.core.Event>()
-                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)
    }
@@ -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<RelayInformationFee>? = null,
        val subscription: List<RelayInformationFee>? = null,
        val publication: List<RelayInformationFee>? = 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<Int>? = null,
        val time: Int? = null,
        val count: Int? = null,