Adds a simple relay to quartz

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

@@ -0,0 +1,64 @@
+/*
+ * 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.core.Event
+import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
+import com.vitorpamplona.quartz.nip01Core.store.IEventStore
+import kotlinx.coroutines.channels.BufferOverflow
+import kotlinx.coroutines.flow.MutableSharedFlow
+
+class LiveEventStore(
+    private val store: IEventStore,
+) {
+    private val newEventStream =
+        MutableSharedFlow<Event>(
+            replay = 0,
+            extraBufferCapacity = 100, // Optional: adjust for backpressure
+            onBufferOverflow = BufferOverflow.DROP_LATEST, // Default behavior
+        )
+
+    fun insert(event: Event) {
+        store.insert(event)
+        newEventStream.tryEmit(event)
+    }
+
+    suspend fun query(
+        filters: List<Filter>,
+        onEach: (Event) -> Unit,
+        onEose: () -> Unit,
+    ) {
+        // 1. Replay stored events matching filters.
+        store.query(filters, onEach)
+
+        // 2. Signal end of stored events.
+        onEose()
+
+        // 3. Stream live events until cancelled.
+        newEventStream.collect { newEvent ->
+            if (filters.any { it.match(newEvent) }) {
+                onEach(newEvent)
+            }
+        }
+    }
+
+    fun count(filters: List<Filter>) = store.count(filters)
+}

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

@@ -0,0 +1,88 @@
+/*
+ * 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.core.Event
+import com.vitorpamplona.quartz.nip01Core.crypto.verify
+import com.vitorpamplona.quartz.nip01Core.store.IEventStore
+import com.vitorpamplona.quartz.utils.cache.LargeCache
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.cancel
+import kotlin.coroutines.CoroutineContext
+
+/**
+ * This class manages per-connection subscriptions as coroutines. Each
+ * subscription ([REQ]) launches a child coroutine that first replays stored
+ * events matching the filters, sends EOSE, and then streams live events.
+ * Closing a subscription ([CLOSE]) immediately cancels its coroutine.
+ *
+ * The server is transport-agnostic: callers feed incoming JSON via
+ * [processMessage] and receive outgoing JSON via the [send] callback
+ * provided to [connect]. This allows use with any WebSocket library.
+ *
+ * @param store The [EventStore] backing this relay.
+ * @param verify Validates incoming events. Defaults to cryptographic
+ *                      verification (id + signature). Override for testing.
+ */
+class NostrServer(
+    private val store: IEventStore,
+    private val parentContext: CoroutineContext = SupervisorJob(),
+    private val verify: (Event) -> Boolean = { it.verify() },
+) {
+    private val subStore = LiveEventStore(store)
+
+    private val scope = CoroutineScope(parentContext + SupervisorJob())
+
+    /** Active client sessions keyed by an opaque connection id. */
+    private val connections = LargeCache<Int, RelaySession>()
+
+    /**
+     * Registers a new client connection.
+     *
+     * @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(
+            store = subStore,
+            verify = verify,
+            scope = scope,
+            onSend = send,
+            onClose = ::disconnect,
+        ).also { session -> connections.put(session.hashCode(), session) }
+
+    /**
+     * Removes a client connection and cancels all its subscriptions.
+     */
+    private fun disconnect(session: RelaySession) {
+        connections.remove(session.hashCode())
+    }
+
+    /**
+     * Shuts down the server, cancelling all subscriptions and sessions.
+     */
+    fun shutdown() {
+        connections.forEach { id, session -> session.cancelAllSubscriptions() }
+        connections.clear()
+        scope.cancel()
+    }
+}

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

@@ -0,0 +1,163 @@
+/*
+ * 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.core.Event
+import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper
+import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.ClosedMessage
+import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.CountMessage
+import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.CountResult
+import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.EoseMessage
+import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.EventMessage
+import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message
+import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.NoticeMessage
+import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.OkMessage
+import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.CloseCmd
+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.utils.Log
+import com.vitorpamplona.quartz.utils.cache.LargeCache
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.launch
+
+/**
+ * Represents a single connected client with its active subscriptions.
+ */
+class RelaySession(
+    private val store: LiveEventStore,
+    private val verify: (Event) -> Boolean,
+    private val scope: CoroutineScope,
+    private val onSend: (String) -> Unit,
+    private val onClose: (RelaySession) -> Unit,
+) : AutoCloseable {
+    private val subscriptions = LargeCache<String, Job>()
+
+    fun send(message: Message) {
+        try {
+            onSend(OptimizedJsonMapper.toJson(message))
+        } catch (e: Exception) {
+            Log.w("ClientSession", "Failed to send to ${e.message}")
+        }
+    }
+
+    fun addSubscription(
+        subId: String,
+        job: Job,
+    ) = subscriptions.put(subId, job)
+
+    fun cancelSubscription(subId: String): Boolean =
+        subscriptions.remove(subId)?.let {
+            it.cancel()
+            true
+        } ?: false
+
+    fun cancelAllSubscriptions() {
+        subscriptions.forEach { _, job -> job.cancel() }
+        subscriptions.clear()
+    }
+
+    override fun close() {
+        cancelAllSubscriptions()
+        onClose(this)
+    }
+
+    /**
+     * Processes a raw JSON message from a client.
+     *
+     * Parses the message as a NIP-01 command and dispatches it.
+     */
+    suspend fun processMessage(message: String) {
+        val cmd =
+            try {
+                OptimizedJsonMapper.fromJsonToCommand(message)
+            } catch (_: Exception) {
+                send(NoticeMessage("error: could not parse message"))
+                return
+            }
+
+        if (!cmd.isValid()) {
+            send(NoticeMessage("error: invalid command"))
+            return
+        }
+
+        when (cmd) {
+            is EventCmd -> handleEvent(cmd)
+            is ReqCmd -> handleReq(cmd)
+            is CloseCmd -> handleClose(cmd)
+            is CountCmd -> handleCount(cmd)
+            else -> send(NoticeMessage("error: unsupported command ${cmd.label()}"))
+        }
+    }
+
+    // -- NIP-01: EVENT --------------------------------------------------------
+    private fun handleEvent(cmd: EventCmd) {
+        val event = cmd.event
+
+        if (!verify(event)) {
+            send(OkMessage(event.id, false, "invalid: bad signature or id"))
+            return
+        }
+
+        try {
+            store.insert(event)
+            send(OkMessage(event.id, true, ""))
+        } catch (e: Exception) {
+            send(OkMessage(event.id, false, e.message ?: e::class.simpleName ?: "unkown error"))
+        }
+    }
+
+    // -- NIP-01: REQ ----------------------------------------------------------
+    private fun handleReq(cmd: ReqCmd) {
+        // Cancel any existing subscription with the same id (NIP-01 spec).
+        cancelSubscription(cmd.subId)
+
+        val job =
+            scope.launch {
+                try {
+                    store.query(
+                        filters = cmd.filters,
+                        onEach = { send(EventMessage(cmd.subId, it)) },
+                        onEose = { send(EoseMessage(cmd.subId)) },
+                    )
+                } catch (_: kotlinx.coroutines.CancellationException) {
+                    // Subscription was closed – this is expected.
+                }
+            }
+
+        addSubscription(cmd.subId, job)
+    }
+
+    // -- NIP-01: CLOSE --------------------------------------------------------
+    private fun handleClose(cmd: CloseCmd) {
+        val cancelled = cancelSubscription(cmd.subId)
+        if (!cancelled) {
+            send(ClosedMessage(cmd.subId, "error: no such subscription"))
+        }
+    }
+
+    // -- NIP-45: COUNT --------------------------------------------------------
+    private fun handleCount(cmd: CountCmd) {
+        val total = store.count(cmd.filters)
+        send(CountMessage(cmd.queryId, CountResult(total)))
+    }
+}

quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServerTest.kt

@@ -0,0 +1,395 @@
+/*
+ * 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.core.Event
+import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper
+import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.EoseMessage
+import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.EventMessage
+import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.EventCmd
+import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
+import com.vitorpamplona.quartz.nip01Core.store.IEventStore
+import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.test.UnconfinedTestDispatcher
+import kotlinx.coroutines.test.runTest
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertTrue
+
+@OptIn(ExperimentalCoroutinesApi::class)
+class NostrServerTest {
+    private val pubkey = "46fcbe3065eaf1ae7811465924e48923363ff3f526bd6f73d7c184b16bd8ce4d"
+    private val sig = "4aa5264965018fa12a326686ad3d3bd8beae3218dcc83689b19ca1e6baeb791531943c15363aa6707c7c0c8b2d601deca1f20c32078b2872d356cdca03b04cce"
+
+    private fun hexId(n: Int): String = n.toString().padStart(64, '0')
+
+    private fun testEvent(
+        id: String = hexId(1),
+        kind: Int = 1,
+        createdAt: Long = 1000L,
+        content: String = "hello",
+        tags: Array<Array<String>> = emptyArray(),
+    ) = Event(id, pubkey, createdAt, kind, tags, content, sig)
+
+    /**
+     * Creates a server using the given dispatcher so coroutines run eagerly
+     * in tests (UnconfinedTestDispatcher).
+     */
+    private fun createServer(
+        store: IEventStore = EventStore(null),
+        dispatcher: kotlinx.coroutines.CoroutineDispatcher,
+    ): NostrServer =
+        NostrServer(
+            store = store,
+            parentContext = dispatcher,
+            verify = { true },
+        )
+
+    private suspend fun RelaySession.insert(event: Event) {
+        val cmd = EventCmd(event)
+        this.processMessage(OptimizedJsonMapper.toJson(cmd))
+    }
+
+    /** Collects sent JSON messages for a connection. */
+    private class MessageCollector {
+        val messages = mutableListOf<String>()
+
+        val sendCallback: (String) -> Unit = { messages.add(it) }
+
+        /**
+         * Parses messages that can be round-tripped (EVENT, EOSE, NOTICE,
+         * CLOSED). OkMessage and CountMessage serialization uses formats
+         * incompatible with the client-side deserializer, so check those
+         * via [rawMessagesContaining].
+         */
+        fun parsedEventMessages() =
+            messages
+                .filter { it.startsWith("[\"EVENT\"") || it.startsWith("[\"EOSE\"") }
+                .map { OptimizedJsonMapper.fromJsonToMessage(it) }
+
+        fun rawMessagesContaining(label: String) = messages.filter { it.contains("\"$label\"") }
+    }
+
+    // -- EVENT command ---------------------------------------------------------
+
+    @Test
+    fun eventCommandStoresAndRespondsOk() =
+        runTest {
+            val dispatcher = UnconfinedTestDispatcher(testScheduler)
+            val store = EventStore(null)
+            val server = createServer(store, dispatcher)
+            val collector = MessageCollector()
+
+            val c1 = server.connect(collector.sendCallback)
+
+            val event = testEvent()
+            val eventJson = """["EVENT",${event.toJson()}]"""
+            c1.processMessage(eventJson)
+
+            val okMessages = collector.rawMessagesContaining("OK")
+            assertEquals(1, okMessages.size)
+            assertTrue(okMessages[0].contains("\"true\""))
+
+            // Event should be in store
+            val stored = store.query<Event>(Filter(ids = listOf(event.id)))
+            assertEquals(1, stored.size)
+
+            server.shutdown()
+        }
+
+    @Test
+    fun duplicateEventReturnsOkFalse() =
+        runTest {
+            val dispatcher = UnconfinedTestDispatcher(testScheduler)
+            val store = EventStore(null)
+            val server = createServer(store, dispatcher)
+            val collector = MessageCollector()
+
+            val c1 = server.connect(collector.sendCallback)
+
+            val event = testEvent()
+            val eventJson = """["EVENT",${event.toJson()}]"""
+            c1.processMessage(eventJson)
+            c1.processMessage(eventJson)
+
+            val okMessages = collector.rawMessagesContaining("OK")
+            assertEquals(2, okMessages.size)
+            assertTrue(okMessages[0].contains("\"true\""))
+            assertTrue(okMessages[1].contains("\"false\""))
+
+            server.shutdown()
+        }
+
+    // -- REQ command -----------------------------------------------------------
+
+    @Test
+    fun reqReturnsStoredEventsAndEose() =
+        runTest {
+            val dispatcher = UnconfinedTestDispatcher(testScheduler)
+            val store = EventStore(null)
+            val server = createServer(store, dispatcher)
+
+            // Pre-populate store
+            store.insert(testEvent(hexId(1), kind = 1, createdAt = 100L))
+            store.insert(testEvent(hexId(2), kind = 1, createdAt = 200L))
+            store.insert(testEvent(hexId(3), kind = 4, createdAt = 300L))
+
+            val collector = MessageCollector()
+            val c1 = server.connect(collector.sendCallback)
+
+            val reqJson = """["REQ","sub1",{"kinds":[1]}]"""
+            c1.processMessage(reqJson)
+
+            val parsed = collector.parsedEventMessages()
+            val events = parsed.filterIsInstance<EventMessage>()
+            val eose = parsed.filterIsInstance<EoseMessage>()
+
+            assertEquals(2, events.size)
+            assertEquals(1, eose.size)
+            assertEquals("sub1", eose[0].subId)
+
+            // Events should be newest first
+            assertTrue(events[0].event.createdAt >= events[1].event.createdAt)
+
+            server.shutdown()
+        }
+
+    @Test
+    fun reqWithLimitRespectsLimit() =
+        runTest {
+            val dispatcher = UnconfinedTestDispatcher(testScheduler)
+            val store = EventStore(null)
+            val server = createServer(store, dispatcher)
+
+            for (i in 1..10) {
+                store.insert(testEvent(hexId(i), createdAt = i.toLong()))
+            }
+
+            val collector = MessageCollector()
+            val c1 = server.connect(collector.sendCallback)
+
+            val reqJson = """["REQ","sub1",{"limit":3}]"""
+            c1.processMessage(reqJson)
+
+            val events = collector.parsedEventMessages().filterIsInstance<EventMessage>()
+            assertEquals(3, events.size)
+
+            server.shutdown()
+        }
+
+    // -- Live subscription -----------------------------------------------------
+
+    @Test
+    fun liveSubscriptionReceivesNewEvents() =
+        runTest {
+            val dispatcher = UnconfinedTestDispatcher(testScheduler)
+            val store = EventStore(null)
+            val server = createServer(store, dispatcher)
+            val collector1 = MessageCollector()
+            val collector2 = MessageCollector()
+
+            val c1 = server.connect(collector1.sendCallback)
+            val c2 = server.connect(collector2.sendCallback)
+
+            // Subscribe to kind 1
+            val reqJson = """["REQ","sub1",{"kinds":[1]}]"""
+            c1.processMessage(reqJson)
+
+            // After REQ, we should have EOSE
+            val countAfterEose = collector1.messages.size
+
+            // Now store a new event — should be pushed to subscription
+            c2.insert(testEvent(hexId(1), kind = 1))
+
+            val newMessages = collector1.messages.drop(countAfterEose)
+            assertTrue(newMessages.isNotEmpty())
+            assertTrue(newMessages[0].contains("\"EVENT\""))
+
+            server.shutdown()
+        }
+
+    @Test
+    fun liveSubscriptionFiltersNonMatchingEvents() =
+        runTest {
+            val dispatcher = UnconfinedTestDispatcher(testScheduler)
+            val store = EventStore(null)
+            val server = createServer(store, dispatcher)
+            val collector1 = MessageCollector()
+            val collector2 = MessageCollector()
+
+            val c1 = server.connect(collector1.sendCallback)
+            val c2 = server.connect(collector2.sendCallback)
+
+            val reqJson = """["REQ","sub1",{"kinds":[1]}]"""
+            c1.processMessage(reqJson)
+
+            val countAfterEose = collector1.messages.size
+
+            // Store a kind 4 event — should NOT match kind 1 subscription
+            c2.insert(testEvent(hexId(1), kind = 4))
+
+            assertEquals(countAfterEose, collector1.messages.size)
+
+            server.shutdown()
+        }
+
+    // -- CLOSE command ---------------------------------------------------------
+
+    @Test
+    fun closeStopsSubscription() =
+        runTest {
+            val dispatcher = UnconfinedTestDispatcher(testScheduler)
+            val store = EventStore(null)
+            val server = createServer(store, dispatcher)
+            val collector1 = MessageCollector()
+            val collector2 = MessageCollector()
+
+            val c1 = server.connect(collector1.sendCallback)
+            val c2 = server.connect(collector2.sendCallback)
+
+            val reqJson = """["REQ","sub1",{"kinds":[1]}]"""
+            c1.processMessage(reqJson)
+
+            // Close the subscription
+            val closeJson = """["CLOSE","sub1"]"""
+            c1.processMessage(closeJson)
+
+            val countAfterClose = collector1.messages.size
+
+            // New events should NOT reach this subscription
+            c2.insert(testEvent(hexId(1), kind = 1))
+
+            assertEquals(countAfterClose, collector1.messages.size)
+
+            server.shutdown()
+        }
+
+    @Test
+    fun replacingSubscriptionCancelsOld() =
+        runTest {
+            val dispatcher = UnconfinedTestDispatcher(testScheduler)
+            val server = createServer(EventStore(null), dispatcher)
+            val collector1 = MessageCollector()
+            val collector2 = MessageCollector()
+
+            val c1 = server.connect(collector1.sendCallback)
+            val c2 = server.connect(collector2.sendCallback)
+
+            // First subscription for kind 1
+            c1.processMessage("""["REQ","sub1",{"kinds":[1]}]""")
+
+            // Replace with kind 4
+            c1.processMessage("""["REQ","sub1",{"kinds":[4]}]""")
+
+            val countAfterReplace = collector1.messages.size
+
+            // Kind 1 events should not match anymore
+            c2.insert(testEvent(hexId(1), kind = 1))
+
+            assertEquals(countAfterReplace, collector1.messages.size)
+
+            // Kind 4 events should match
+            c2.insert(testEvent(hexId(2), kind = 4))
+
+            val newMessages = collector1.messages.drop(countAfterReplace)
+
+            assertTrue(newMessages.isNotEmpty())
+            assertTrue(newMessages[0].contains("\"EVENT\""))
+            assertTrue(newMessages[0].contains(hexId(2)))
+
+            server.shutdown()
+        }
+
+    // -- COUNT command (NIP-45) ------------------------------------------------
+
+    @Test
+    fun countReturnsMatchingEventCount() =
+        runTest {
+            val dispatcher = UnconfinedTestDispatcher(testScheduler)
+            val store = EventStore(null)
+            val server = createServer(store, dispatcher)
+
+            store.insert(testEvent(hexId(1), kind = 1))
+            store.insert(testEvent(hexId(2), kind = 1))
+            store.insert(testEvent(hexId(3), kind = 4))
+
+            val collector = MessageCollector()
+            val c1 = server.connect(collector.sendCallback)
+
+            val countJson = """["COUNT","q1",{"kinds":[1]}]"""
+            c1.processMessage(countJson)
+
+            val countMessages = collector.rawMessagesContaining("COUNT")
+            assertEquals(1, countMessages.size)
+            assertTrue(countMessages[0].contains("\"count\":2"))
+
+            server.shutdown()
+        }
+
+    // -- Disconnect ------------------------------------------------------------
+
+    @Test
+    fun disconnectCancelsAllSubscriptions() =
+        runTest {
+            val dispatcher = UnconfinedTestDispatcher(testScheduler)
+            val store = EventStore(null)
+            val server = createServer(store, dispatcher)
+            val collector1 = MessageCollector()
+            val collector2 = MessageCollector()
+
+            val c1 = server.connect(collector1.sendCallback)
+            val c2 = server.connect(collector2.sendCallback)
+
+            c1.processMessage("""["REQ","sub1",{"kinds":[1]}]""")
+            c1.processMessage("""["REQ","sub2",{"kinds":[4]}]""")
+
+            c1.close()
+
+            val countAfterDisconnect = collector1.messages.size
+
+            c2.insert(testEvent(hexId(1), kind = 1))
+            c2.insert(testEvent(hexId(2), kind = 4))
+
+            assertEquals(countAfterDisconnect, collector1.messages.size)
+            assertEquals(2, collector2.messages.size)
+
+            server.shutdown()
+        }
+
+    // -- Invalid messages ------------------------------------------------------
+
+    @Test
+    fun invalidJsonReturnsNotice() =
+        runTest {
+            val dispatcher = UnconfinedTestDispatcher(testScheduler)
+            val server = createServer(dispatcher = dispatcher)
+            val collector = MessageCollector()
+
+            val c1 = server.connect(collector.sendCallback)
+            c1.processMessage("not valid json")
+
+            assertEquals(1, collector.messages.size)
+            assertTrue(collector.messages[0].contains("NOTICE"))
+
+            server.shutdown()
+        }
+}