refactor(quartz): clarity pass on relay server internals

Behaviour-preserving readability improvements (full suite green):

- LimitsPolicy: drop the <T : Command> generic reject helpers (which forced
  rejectSubId<ReqCmd>(...) call-site type args). Each check is now a
  "rejection reason or null" function (eventRejection / subscriptionRejection)
  and the accept overloads just wrap a non-null reason — the three overloads
  read almost identically.
- Extract ConnectionRegistry: NostrServer and ReqResponderServer duplicated
  the connection bookkeeping (stable-id keying, active gauge, once-only
  teardown accounting). That subtle logic now lives in one named class both
  servers delegate to; their connect()/close() shrink to the parts that
  actually differ (the backend and what else teardown closes).
- HyperLogLog.addPubKey: rename `ri` -> `registerIndex`.

https://claude.ai/code/session_016YBS2pWCBSDgAMthHzfCTr

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/ConnectionRegistry.kt

@@ -0,0 +1,78 @@
+/*
+ * 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.utils.cache.LargeCache
+import kotlin.concurrent.atomics.AtomicLong
+import kotlin.concurrent.atomics.ExperimentalAtomicApi
+
+/**
+ * Tracks the live [RelaySession]s of one relay server and fires the
+ * [RelayServerListener] as they come and go. Shared by [NostrServer] and
+ * [ReqResponderServer] so the connection bookkeeping — the stable-id keying,
+ * the [active] gauge, and the once-only teardown accounting — lives in exactly
+ * one place.
+ */
+@OptIn(ExperimentalAtomicApi::class)
+class ConnectionRegistry(
+    private val listener: RelayServerListener,
+) {
+    private val connections = LargeCache<Long, RelaySession>()
+    private val activeCount = AtomicLong(0L)
+
+    /** Number of connections currently registered. */
+    val active: Long get() = activeCount.load()
+
+    /** Records [session] and fires [RelayServerListener.onConnect]. Returns it for chaining. */
+    fun register(session: RelaySession): RelaySession {
+        connections.put(session.id, session)
+        activeCount.addAndFetch(1L)
+        listener.onConnect(session.id)
+        return session
+    }
+
+    /**
+     * Drops the connection with [id], decrementing [active] and firing
+     * [RelayServerListener.onDisconnect] — but only on the first call for a
+     * given connection, so a double `close()` can't underflow the gauge or
+     * double-fire the listener.
+     */
+    fun unregister(id: Long) {
+        if (connections.remove(id) != null) {
+            activeCount.addAndFetch(-1L)
+            listener.onDisconnect(id)
+        }
+    }
+
+    /**
+     * Cancels every still-open connection's subscriptions and fires
+     * [RelayServerListener.onDisconnect] for each, then resets the registry.
+     * Used by the server's `close()`.
+     */
+    fun closeAll() {
+        connections.forEach { _, session ->
+            session.cancelAllSubscriptions()
+            listener.onDisconnect(session.id)
+        }
+        connections.clear()
+        activeCount.store(0L)
+    }
+}

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServer.kt

@@ -25,12 +25,9 @@ import com.vitorpamplona.quartz.nip01Core.relay.server.policies.LimitsPolicy
 import com.vitorpamplona.quartz.nip01Core.relay.server.policies.VerifyPolicy
 import com.vitorpamplona.quartz.nip01Core.store.IEventStore
 import com.vitorpamplona.quartz.nip77Negentropy.NegentropySettings
-import com.vitorpamplona.quartz.utils.cache.LargeCache
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.cancel
-import kotlin.concurrent.atomics.AtomicLong
-import kotlin.concurrent.atomics.ExperimentalAtomicApi
 import kotlin.coroutines.CoroutineContext
 
 /**
@@ -57,22 +54,18 @@ import kotlin.coroutines.CoroutineContext
  *   subscription caps) and advertised via [RelayLimits.toNip11Limitation].
  *   Null disables limit enforcement.
  */
-@OptIn(ExperimentalAtomicApi::class)
 class NostrServer(
     private val store: IEventStore,
     private val policyBuilder: () -> IRelayPolicy = { VerifyPolicy },
     private val parentContext: CoroutineContext = SupervisorJob(),
     parallelVerify: Boolean = false,
     private val negentropySettings: NegentropySettings = NegentropySettings.Default,
-    private val listener: RelayServerListener = RelayServerListener.None,
+    listener: RelayServerListener = RelayServerListener.None,
     val limits: RelayLimits? = null,
 ) : AutoCloseable {
     /** Scope for all subscriptions. */
     private val scope = CoroutineScope(parentContext + SupervisorJob())
 
-    /** Live count of registered connections; backs [activeConnections]. */
-    private val activeCount = AtomicLong(0L)
-
     /**
      * Group-commit writer shared across every connected session.
      * Sessions hand off EVENT publishes here instead of awaiting
@@ -89,11 +82,10 @@ class NostrServer(
 
     private val subStore = LiveEventStore(store, ingest)
 
-    /** Active client sessions keyed by [RelaySession.id]. */
-    private val connections = LargeCache<Long, RelaySession>()
+    private val connections = ConnectionRegistry(listener)
 
     /** Number of connections currently registered with this server. */
-    val activeConnections: Long get() = activeCount.load()
+    val activeConnections: Long get() = connections.active
 
     /**
      * Builds the per-connection policy, prepending a [LimitsPolicy] when
@@ -111,29 +103,17 @@ class NostrServer(
      * @param send Callback the server uses to send JSON messages to this client.
      *             Implementations must be safe to call from any coroutine.
      */
-    fun connect(send: (String) -> Unit): RelaySession {
-        val session =
+    fun connect(send: (String) -> Unit): RelaySession =
+        connections.register(
             RelaySession(
                 policy = buildPolicy(),
                 store = subStore,
                 scope = scope,
                 onSend = send,
-                onClose = { closed ->
-                    // Idempotent: only account for the first teardown of a
-                    // given connection so a double close() can't underflow
-                    // the gauge or double-fire the listener.
-                    if (connections.remove(closed.id) != null) {
-                        activeCount.addAndFetch(-1L)
-                        listener.onDisconnect(closed.id)
-                    }
-                },
+                onClose = { connections.unregister(it.id) },
                 negentropySettings = negentropySettings,
-            )
-        connections.put(session.id, session)
-        activeCount.addAndFetch(1L)
-        listener.onConnect(session.id)
-        return session
-    }
+            ),
+        )
 
     /**
      * Registers a new client connection and serves it for the duration of
@@ -159,12 +139,7 @@ class NostrServer(
      * Shuts down the server, cancelling all subscriptions and closing the store.
      */
     override fun close() {
-        connections.forEach { _, session ->
-            session.cancelAllSubscriptions()
-            listener.onDisconnect(session.id)
-        }
-        connections.clear()
-        activeCount.store(0L)
+        connections.closeAll()
         ingest.close()
         scope.cancel()
         store.close()

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/ReqResponderServer.kt

@@ -23,12 +23,9 @@ package com.vitorpamplona.quartz.nip01Core.relay.server
 import com.vitorpamplona.quartz.nip01Core.relay.server.policies.EmptyPolicy
 import com.vitorpamplona.quartz.nip01Core.relay.server.policies.LimitsPolicy
 import com.vitorpamplona.quartz.nip77Negentropy.NegentropySettings
-import com.vitorpamplona.quartz.utils.cache.LargeCache
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.cancel
-import kotlin.concurrent.atomics.AtomicLong
-import kotlin.concurrent.atomics.ExperimentalAtomicApi
 import kotlin.coroutines.CoroutineContext
 
 /**
@@ -72,13 +69,12 @@ import kotlin.coroutines.CoroutineContext
  *   subscription caps) and advertised via [RelayLimits.toNip11Limitation].
  *   Null disables limit enforcement.
  */
-@OptIn(ExperimentalAtomicApi::class)
 class ReqResponderServer(
     responder: ReqResponder,
     private val policyBuilder: () -> IRelayPolicy = { EmptyPolicy },
     parentContext: CoroutineContext = SupervisorJob(),
     private val negentropySettings: NegentropySettings = NegentropySettings.Default,
-    private val listener: RelayServerListener = RelayServerListener.None,
+    listener: RelayServerListener = RelayServerListener.None,
     val limits: RelayLimits? = null,
 ) : AutoCloseable {
     /** Scope for all subscriptions. */
@@ -86,14 +82,10 @@ class ReqResponderServer(
 
     private val backend = ReqResponderBackend(responder)
 
-    /** Live count of registered connections; backs [activeConnections]. */
-    private val activeCount = AtomicLong(0L)
-
-    /** Active client sessions keyed by [RelaySession.id]. */
-    private val connections = LargeCache<Long, RelaySession>()
+    private val connections = ConnectionRegistry(listener)
 
     /** Number of connections currently registered with this server. */
-    val activeConnections: Long get() = activeCount.load()
+    val activeConnections: Long get() = connections.active
 
     private fun buildPolicy(): IRelayPolicy {
         val base = policyBuilder()
@@ -106,27 +98,17 @@ class ReqResponderServer(
      * @param send Callback the server uses to send JSON messages to this client.
      *             Implementations must be safe to call from any coroutine.
      */
-    fun connect(send: (String) -> Unit): RelaySession {
-        val session =
+    fun connect(send: (String) -> Unit): RelaySession =
+        connections.register(
             RelaySession(
                 policy = buildPolicy(),
                 store = backend,
                 scope = scope,
                 onSend = send,
-                onClose = { closed ->
-                    // Idempotent teardown accounting (see NostrServer.connect).
-                    if (connections.remove(closed.id) != null) {
-                        activeCount.addAndFetch(-1L)
-                        listener.onDisconnect(closed.id)
-                    }
-                },
+                onClose = { connections.unregister(it.id) },
                 negentropySettings = negentropySettings,
-            )
-        connections.put(session.id, session)
-        activeCount.addAndFetch(1L)
-        listener.onConnect(session.id)
-        return session
-    }
+            ),
+        )
 
     /**
      * Registers a new client connection and serves it for the duration of
@@ -150,12 +132,7 @@ class ReqResponderServer(
 
     /** Shuts down the server, cancelling all subscriptions. */
     override fun close() {
-        connections.forEach { _, session ->
-            session.cancelAllSubscriptions()
-            listener.onDisconnect(session.id)
-        }
-        connections.clear()
-        activeCount.store(0L)
+        connections.closeAll()
         scope.cancel()
     }
 }

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/policies/LimitsPolicy.kt

@@ -20,8 +20,8 @@
  */
 package com.vitorpamplona.quartz.nip01Core.relay.server.policies
 
+import com.vitorpamplona.quartz.nip01Core.core.Event
 import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.MachineReadablePrefix
-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
@@ -42,13 +42,16 @@ import com.vitorpamplona.quartz.nip01Core.relay.server.RelayLimits
  * caps ([RelayLimits.maxMessageLength], [RelayLimits.maxSubscriptions]) are
  * enforced through the [acceptMessage] / [acceptSubscription] policy hooks, so
  * everything limit-related composes uniformly across a [PolicyStack].
+ *
+ * Each check is expressed as a "rejection reason or null" helper; the command
+ * `accept` overloads just wrap a non-null reason in [PolicyResult.Rejected].
  */
 class LimitsPolicy(
     private val limits: RelayLimits,
 ) : PassThroughPolicy() {
     override fun acceptMessage(message: String): String? {
         val max = limits.maxMessageLength ?: return null
-        return if (message.length > max) MachineReadablePrefix.INVALID.format("message too large (max $max)") else null
+        return if (message.length > max) invalid("message too large (max $max)") else null
     }
 
     override fun acceptSubscription(
@@ -63,49 +66,41 @@ class LimitsPolicy(
         }
     }
 
-    override fun accept(cmd: EventCmd): PolicyResult<EventCmd> {
-        val event = cmd.event
-        limits.maxContentLength?.let {
-            if (event.content.length > it) return reject("content too large (max $it)")
-        }
-        limits.maxEventTags?.let {
-            if (event.tags.size > it) return reject("too many tags (max $it)")
-        }
-        limits.createdAtLowerLimit?.let {
-            if (event.createdAt < it) return reject("created_at is before the relay's lower limit")
-        }
-        limits.createdAtUpperLimit?.let {
-            if (event.createdAt > it) return reject("created_at is after the relay's upper limit")
-        }
-        return PolicyResult.Accepted(cmd)
-    }
+    override fun accept(cmd: EventCmd): PolicyResult<EventCmd> = eventRejection(cmd.event)?.let { PolicyResult.Rejected(it) } ?: PolicyResult.Accepted(cmd)
 
     override fun accept(cmd: ReqCmd): PolicyResult<ReqCmd> {
-        rejectSubId<ReqCmd>(cmd.subId)?.let { return it }
-        rejectFilterCount<ReqCmd>(cmd.filters)?.let { return it }
+        subscriptionRejection(cmd.subId, cmd.filters)?.let { return PolicyResult.Rejected(it) }
         val clamped = clampLimits(cmd.filters)
         return PolicyResult.Accepted(if (clamped === cmd.filters) cmd else ReqCmd(cmd.subId, clamped))
     }
 
     override fun accept(cmd: CountCmd): PolicyResult<CountCmd> {
-        rejectSubId<CountCmd>(cmd.queryId)?.let { return it }
-        rejectFilterCount<CountCmd>(cmd.filters)?.let { return it }
+        subscriptionRejection(cmd.queryId, cmd.filters)?.let { return PolicyResult.Rejected(it) }
         val clamped = clampLimits(cmd.filters)
         return PolicyResult.Accepted(if (clamped === cmd.filters) cmd else CountCmd(cmd.queryId, clamped))
     }
 
-    private fun <T : Command> reject(message: String): PolicyResult<T> = PolicyResult.Rejected(MachineReadablePrefix.INVALID.format(message))
-
-    private fun <T : Command> rejectSubId(subId: String): PolicyResult<T>? {
-        val max = limits.maxSubidLength ?: return null
-        return if (subId.length > max) reject("subscription id too long (max $max)") else null
+    /** The reason an EVENT violates a limit, or null when it's within bounds. */
+    private fun eventRejection(event: Event): String? {
+        limits.maxContentLength?.let { if (event.content.length > it) return invalid("content too large (max $it)") }
+        limits.maxEventTags?.let { if (event.tags.size > it) return invalid("too many tags (max $it)") }
+        limits.createdAtLowerLimit?.let { if (event.createdAt < it) return invalid("created_at is before the relay's lower limit") }
+        limits.createdAtUpperLimit?.let { if (event.createdAt > it) return invalid("created_at is after the relay's upper limit") }
+        return null
     }
 
-    private fun <T : Command> rejectFilterCount(filters: List<Filter>): PolicyResult<T>? {
-        val max = limits.maxFilters ?: return null
-        return if (filters.size > max) reject("too many filters (max $max)") else null
+    /** The reason a REQ/COUNT violates a limit, or null when it's within bounds. */
+    private fun subscriptionRejection(
+        subId: String,
+        filters: List<Filter>,
+    ): String? {
+        limits.maxSubidLength?.let { if (subId.length > it) return invalid("subscription id too long (max $it)") }
+        limits.maxFilters?.let { if (filters.size > it) return invalid("too many filters (max $it)") }
+        return null
     }
 
+    private fun invalid(message: String): String = MachineReadablePrefix.INVALID.format(message)
+
     /** Returns the same list reference when nothing changes, so callers can skip rebuilding the command. */
     private fun clampLimits(filters: List<Filter>): List<Filter> {
         if (limits.maxLimit == null && limits.defaultLimit == null) return filters

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip45Count/HyperLogLog.kt

@@ -131,9 +131,9 @@ object HyperLogLog {
         offset: Int,
     ) {
         if (offset < 0 || offset >= pubKey.size) return
-        val ri = pubKey[offset].toInt() and 0xFF
+        val registerIndex = pubKey[offset].toInt() and 0xFF
         val value = (leadingZeroBits(pubKey, offset + 1) + 1).coerceAtMost(0xFF)
-        if (value > (registers[ri].toInt() and 0xFF)) registers[ri] = value.toByte()
+        if (value > (registers[registerIndex].toInt() and 0xFF)) registers[registerIndex] = value.toByte()
     }
 
     /**