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 new file mode 100644 index 000000000..77f1df4fb --- /dev/null +++ b/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( + 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, + 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) = store.count(filters) +} diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServer.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServer.kt new file mode 100644 index 000000000..96476ab24 --- /dev/null +++ b/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() + + /** + * 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() + } +} 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 new file mode 100644 index 000000000..da27c4a45 --- /dev/null +++ b/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() + + 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))) + } +} diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServerTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServerTest.kt new file mode 100644 index 000000000..d9eb0e09c --- /dev/null +++ b/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> = 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() + + 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(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() + val eose = parsed.filterIsInstance() + + 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() + 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() + } +}