diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/EventStore.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/EventStore.kt new file mode 100644 index 0000000000..eab898c757 --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/EventStore.kt @@ -0,0 +1,60 @@ +/* + * 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 kotlinx.coroutines.flow.SharedFlow + +/** + * Storage backend for the relay server. + * + * Implementations must be thread-safe. The [newEvents] flow is used by the + * server to push live events to active subscriptions. + */ +interface EventStore { + /** Emits every event that is successfully stored. */ + val newEvents: SharedFlow + + /** + * Persists [event] and returns `true` if it was accepted. + * + * Implementations should handle deduplication (by event id), + * replaceable-event semantics (NIP-01 kinds 0/3/10000-19999), + * and addressable-event semantics (kinds 30000-39999). + * Ephemeral events (kinds 20000-29999) should be emitted on + * [newEvents] but may be discarded without persistence. + */ + suspend fun store(event: Event): Boolean + + /** + * Queries stored events matching the given [filter]. + * + * The returned list respects the filter's `limit` field when present, + * ordered newest-first. + */ + suspend fun query(filter: Filter): List + + /** + * Returns the number of stored events matching [filter] (NIP-45). + */ + suspend fun count(filter: Filter): Int +} diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/InMemoryEventStore.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/InMemoryEventStore.kt new file mode 100644 index 0000000000..85176cc26b --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/InMemoryEventStore.kt @@ -0,0 +1,135 @@ +/* + * 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.HexKey +import com.vitorpamplona.quartz.nip01Core.core.isAddressable +import com.vitorpamplona.quartz.nip01Core.core.isEphemeral +import com.vitorpamplona.quartz.nip01Core.core.isReplaceable +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import kotlinx.coroutines.flow.MutableSharedFlow +import kotlinx.coroutines.flow.SharedFlow +import kotlinx.coroutines.flow.asSharedFlow +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.withLock + +/** + * Thread-safe in-memory [EventStore] implementation. + * + * - Regular events are stored by id. + * - Replaceable events (kinds 0, 3, 10000-19999) keep only the newest per + * (pubKey, kind) pair. + * - Addressable events (kinds 30000-39999) keep only the newest per + * (pubKey, kind, d-tag) triple. + * - Ephemeral events (kinds 20000-29999) are broadcast but not persisted. + */ +class InMemoryEventStore : EventStore { + private val mutex = Mutex() + + /** All non-ephemeral events indexed by id. */ + private val eventsById = LinkedHashMap() + + /** Newest replaceable event per "pubKey:kind". */ + private val replaceableIndex = HashMap() + + /** Newest addressable event per "pubKey:kind:dTag". */ + private val addressableIndex = HashMap() + + private val _newEvents = MutableSharedFlow(extraBufferCapacity = 256) + override val newEvents: SharedFlow = _newEvents.asSharedFlow() + + override suspend fun store(event: Event): Boolean { + if (event.kind.isEphemeral()) { + _newEvents.emit(event) + return true + } + + val accepted = + mutex.withLock { + if (eventsById.containsKey(event.id)) return@withLock false + + when { + event.kind.isReplaceable() -> { + storeReplaceable(event) + } + + event.kind.isAddressable() -> { + storeAddressable(event) + } + + else -> { + eventsById[event.id] = event + true + } + } + } + + if (accepted) { + _newEvents.emit(event) + } + return accepted + } + + /** Must be called while holding [mutex]. */ + private fun storeReplaceable(event: Event): Boolean { + val key = "${event.pubKey}:${event.kind}" + val existingId = replaceableIndex[key] + if (existingId != null) { + val existing = eventsById[existingId] + if (existing != null && existing.createdAt >= event.createdAt) return false + eventsById.remove(existingId) + } + eventsById[event.id] = event + replaceableIndex[key] = event.id + return true + } + + /** Must be called while holding [mutex]. */ + private fun storeAddressable(event: Event): Boolean { + val dTag = event.tags.firstOrNull { it.size > 1 && it[0] == "d" }?.get(1) ?: "" + val key = "${event.pubKey}:${event.kind}:$dTag" + val existingId = addressableIndex[key] + if (existingId != null) { + val existing = eventsById[existingId] + if (existing != null && existing.createdAt >= event.createdAt) return false + eventsById.remove(existingId) + } + eventsById[event.id] = event + addressableIndex[key] = event.id + return true + } + + override suspend fun query(filter: Filter): List = + mutex.withLock { + val matched = + eventsById.values + .filter { filter.match(it) } + .sortedByDescending { it.createdAt } + + if (filter.limit != null) matched.take(filter.limit) else matched + } + + override suspend fun count(filter: Filter): Int = + mutex.withLock { + eventsById.values.count { filter.match(it) } + } +} diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrRelayServer.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrRelayServer.kt new file mode 100644 index 0000000000..eeb5124583 --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrRelayServer.kt @@ -0,0 +1,276 @@ +/* + * 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.crypto.verify +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.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 kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Job +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.cancel +import kotlinx.coroutines.launch +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.withLock +import kotlin.coroutines.CoroutineContext + +/** + * Nostr relay server implementing NIP-01 and NIP-45. + * + * 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 eventVerifier Validates incoming events. Defaults to cryptographic + * verification (id + signature). Override for testing. + */ +class NostrRelayServer( + val store: EventStore, + parentContext: CoroutineContext = SupervisorJob(), + private val eventVerifier: (Event) -> Boolean = { it.verify() }, +) { + private val scope = CoroutineScope(parentContext + SupervisorJob()) + + /** Active client sessions keyed by an opaque connection id. */ + private val sessions = HashMap() + private val sessionsMutex = Mutex() + + /** + * Registers a new client connection. + * + * @param connectionId Unique id for this connection (e.g. WebSocket session id). + * @param send Callback the server uses to send JSON messages to this client. + * Implementations must be safe to call from any coroutine. + */ + suspend fun connect( + connectionId: String, + send: suspend (String) -> Unit, + ) { + sessionsMutex.withLock { + sessions[connectionId] = ClientSession(connectionId, send) + } + } + + /** + * Removes a client connection and cancels all its subscriptions. + */ + suspend fun disconnect(connectionId: String) { + sessionsMutex + .withLock { + sessions.remove(connectionId) + }?.cancelAll() + } + + /** + * Processes a raw JSON message from a client. + * + * Parses the message as a NIP-01 command and dispatches it. + */ + suspend fun processMessage( + connectionId: String, + message: String, + ) { + val session = + sessionsMutex.withLock { sessions[connectionId] } + ?: return + + val command = + try { + OptimizedJsonMapper.fromJsonToCommand(message) + } catch (e: Exception) { + Log.w(TAG, "Invalid message from $connectionId: ${e.message}") + session.send(NoticeMessage("error: could not parse message")) + return + } + + if (!command.isValid()) { + session.send(NoticeMessage("error: invalid command")) + return + } + + when (command) { + is EventCmd -> handleEvent(session, command) + is ReqCmd -> handleReq(session, command) + is CloseCmd -> handleClose(session, command) + is CountCmd -> handleCount(session, command) + else -> session.send(NoticeMessage("error: unsupported command ${command.label()}")) + } + } + + // -- NIP-01: EVENT -------------------------------------------------------- + + private suspend fun handleEvent( + session: ClientSession, + cmd: EventCmd, + ) { + val event = cmd.event + + if (!eventVerifier(event)) { + session.send(OkMessage(event.id, false, "invalid: bad signature or id")) + return + } + + val accepted = store.store(event) + if (accepted) { + session.send(OkMessage(event.id, true, "")) + } else { + session.send(OkMessage(event.id, false, "duplicate: already have this event")) + } + } + + // -- NIP-01: REQ ---------------------------------------------------------- + + private suspend fun handleReq( + session: ClientSession, + cmd: ReqCmd, + ) { + // Cancel any existing subscription with the same id (NIP-01 spec). + session.cancelSubscription(cmd.subId) + + val job = + scope.launch { + try { + // 1. Replay stored events matching filters. + for (filter in cmd.filters) { + val events = store.query(filter) + for (event in events) { + session.send(EventMessage(cmd.subId, event)) + } + } + + // 2. Signal end of stored events. + session.send(EoseMessage(cmd.subId)) + + // 3. Stream live events until cancelled. + store.newEvents.collect { event -> + if (cmd.filters.any { it.match(event) }) { + session.send(EventMessage(cmd.subId, event)) + } + } + } catch (_: kotlinx.coroutines.CancellationException) { + // Subscription was closed – this is expected. + } + } + + session.addSubscription(cmd.subId, job) + } + + // -- NIP-01: CLOSE -------------------------------------------------------- + + private suspend fun handleClose( + session: ClientSession, + cmd: CloseCmd, + ) { + val cancelled = session.cancelSubscription(cmd.subId) + if (!cancelled) { + session.send(ClosedMessage(cmd.subId, "error: no such subscription")) + } + } + + // -- NIP-45: COUNT -------------------------------------------------------- + + private suspend fun handleCount( + session: ClientSession, + cmd: CountCmd, + ) { + var total = 0 + for (filter in cmd.filters) { + total += store.count(filter) + } + session.send(CountMessage(cmd.queryId, CountResult(total))) + } + + /** + * Shuts down the server, cancelling all subscriptions and sessions. + */ + suspend fun shutdown() { + sessionsMutex.withLock { + sessions.values.forEach { it.cancelAll() } + sessions.clear() + } + scope.cancel() + } + + companion object { + private const val TAG = "NostrRelayServer" + } +} + +/** + * Represents a single connected client with its active subscriptions. + */ +internal class ClientSession( + val connectionId: String, + private val sendCallback: suspend (String) -> Unit, +) { + private val subscriptions = HashMap() + private val mutex = Mutex() + + suspend fun send(message: com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message) { + try { + sendCallback(OptimizedJsonMapper.toJson(message)) + } catch (e: Exception) { + Log.w("ClientSession", "Failed to send to $connectionId: ${e.message}") + } + } + + suspend fun addSubscription( + subId: String, + job: Job, + ) { + mutex.withLock { + subscriptions[subId] = job + } + } + + suspend fun cancelSubscription(subId: String): Boolean = + mutex.withLock { + subscriptions.remove(subId)?.let { + it.cancel() + true + } ?: false + } + + suspend fun cancelAll() { + mutex.withLock { + subscriptions.values.forEach { it.cancel() } + subscriptions.clear() + } + } +} diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/InMemoryEventStoreTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/InMemoryEventStoreTest.kt new file mode 100644 index 0000000000..2a7cd5f654 --- /dev/null +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/InMemoryEventStoreTest.kt @@ -0,0 +1,272 @@ +/* + * 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 kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.async +import kotlinx.coroutines.flow.first +import kotlinx.coroutines.test.UnconfinedTestDispatcher +import kotlinx.coroutines.test.runTest +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFalse +import kotlin.test.assertTrue + +@OptIn(ExperimentalCoroutinesApi::class) +class InMemoryEventStoreTest { + private val pubkey1 = "46fcbe3065eaf1ae7811465924e48923363ff3f526bd6f73d7c184b16bd8ce4d" + private val pubkey2 = "22aa81510ee63fe2b16cae16e0921f78e9ba9882e2868e7e63ad6d08ae9b5954" + private val sig = "4aa5264965018fa12a326686ad3d3bd8beae3218dcc83689b19ca1e6baeb791531943c15363aa6707c7c0c8b2d601deca1f20c32078b2872d356cdca03b04cce" + + private fun event( + id: String, + pubKey: String = pubkey1, + createdAt: Long = 1000L, + kind: Int = 1, + tags: Array> = emptyArray(), + content: String = "hello", + ) = Event(id, pubKey, createdAt, kind, tags, content, sig) + + private fun hexId(n: Int): String = n.toString().padStart(64, '0') + + @Test + fun storeAndQueryRegularEvent() = + runTest { + val store = InMemoryEventStore() + val ev = event(hexId(1)) + + assertTrue(store.store(ev)) + + val results = store.query(Filter(ids = listOf(hexId(1)))) + assertEquals(1, results.size) + assertEquals(ev.id, results[0].id) + } + + @Test + fun rejectsDuplicateEvent() = + runTest { + val store = InMemoryEventStore() + val ev = event(hexId(1)) + + assertTrue(store.store(ev)) + assertFalse(store.store(ev)) + } + + @Test + fun replaceableEventKeepsNewest() = + runTest { + val store = InMemoryEventStore() + // Kind 0 (metadata) is replaceable + val older = event(hexId(1), kind = 0, createdAt = 100L) + val newer = event(hexId(2), kind = 0, createdAt = 200L) + + assertTrue(store.store(older)) + assertTrue(store.store(newer)) + + val results = store.query(Filter(kinds = listOf(0), authors = listOf(pubkey1))) + assertEquals(1, results.size) + assertEquals(newer.id, results[0].id) + } + + @Test + fun replaceableEventRejectsOlderVersion() = + runTest { + val store = InMemoryEventStore() + val newer = event(hexId(1), kind = 0, createdAt = 200L) + val older = event(hexId(2), kind = 0, createdAt = 100L) + + assertTrue(store.store(newer)) + assertFalse(store.store(older)) + + val results = store.query(Filter(kinds = listOf(0))) + assertEquals(1, results.size) + assertEquals(newer.id, results[0].id) + } + + @Test + fun addressableEventKeepsNewestPerDTag() = + runTest { + val store = InMemoryEventStore() + // Kind 30000 is addressable + val older = + event( + hexId(1), + kind = 30000, + createdAt = 100L, + tags = arrayOf(arrayOf("d", "mylist")), + ) + val newer = + event( + hexId(2), + kind = 30000, + createdAt = 200L, + tags = arrayOf(arrayOf("d", "mylist")), + ) + val different = + event( + hexId(3), + kind = 30000, + createdAt = 150L, + tags = arrayOf(arrayOf("d", "otherlist")), + ) + + assertTrue(store.store(older)) + assertTrue(store.store(newer)) + assertTrue(store.store(different)) + + val results = store.query(Filter(kinds = listOf(30000))) + assertEquals(2, results.size) + } + + @Test + fun ephemeralEventNotStored() = + runTest { + val store = InMemoryEventStore() + // Kind 20000 is ephemeral + val ev = event(hexId(1), kind = 20000) + + assertTrue(store.store(ev)) + + val results = store.query(Filter(kinds = listOf(20000))) + assertEquals(0, results.size) + } + + @Test + fun ephemeralEventEmittedOnNewEvents() = + runTest(UnconfinedTestDispatcher()) { + val store = InMemoryEventStore() + val ev = event(hexId(1), kind = 20000) + + val received = + async { + store.newEvents.first() + } + + store.store(ev) + + assertEquals(ev.id, received.await().id) + } + + @Test + fun queryWithKindFilter() = + runTest { + val store = InMemoryEventStore() + store.store(event(hexId(1), kind = 1)) + store.store(event(hexId(2), kind = 4)) + store.store(event(hexId(3), kind = 1)) + + val results = store.query(Filter(kinds = listOf(1))) + assertEquals(2, results.size) + } + + @Test + fun queryWithAuthorFilter() = + runTest { + val store = InMemoryEventStore() + store.store(event(hexId(1), pubKey = pubkey1)) + store.store(event(hexId(2), pubKey = pubkey2)) + + val results = store.query(Filter(authors = listOf(pubkey1))) + assertEquals(1, results.size) + assertEquals(pubkey1, results[0].pubKey) + } + + @Test + fun queryWithLimit() = + runTest { + val store = InMemoryEventStore() + for (i in 1..10) { + store.store(event(hexId(i), createdAt = i.toLong())) + } + + val results = store.query(Filter(limit = 3)) + assertEquals(3, results.size) + // Newest first + assertTrue(results[0].createdAt >= results[1].createdAt) + assertTrue(results[1].createdAt >= results[2].createdAt) + } + + @Test + fun queryWithSinceAndUntil() = + runTest { + val store = InMemoryEventStore() + store.store(event(hexId(1), createdAt = 100L)) + store.store(event(hexId(2), createdAt = 200L)) + store.store(event(hexId(3), createdAt = 300L)) + + val results = store.query(Filter(since = 150L, until = 250L)) + assertEquals(1, results.size) + assertEquals(hexId(2), results[0].id) + } + + @Test + fun countMatchingEvents() = + runTest { + val store = InMemoryEventStore() + store.store(event(hexId(1), kind = 1)) + store.store(event(hexId(2), kind = 1)) + store.store(event(hexId(3), kind = 4)) + + assertEquals(2, store.count(Filter(kinds = listOf(1)))) + assertEquals(1, store.count(Filter(kinds = listOf(4)))) + assertEquals(0, store.count(Filter(kinds = listOf(99)))) + } + + @Test + fun queryWithTagFilter() = + runTest { + val store = InMemoryEventStore() + store.store( + event( + hexId(1), + tags = arrayOf(arrayOf("t", "nostr")), + ), + ) + store.store( + event( + hexId(2), + tags = arrayOf(arrayOf("t", "bitcoin")), + ), + ) + + val results = store.query(Filter(tags = mapOf("t" to listOf("nostr")))) + assertEquals(1, results.size) + assertEquals(hexId(1), results[0].id) + } + + @Test + fun newEventsFlowEmitsOnStore() = + runTest(UnconfinedTestDispatcher()) { + val store = InMemoryEventStore() + val ev = event(hexId(1)) + + val received = + async { + store.newEvents.first() + } + + store.store(ev) + + assertEquals(ev.id, received.await().id) + } +} diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrRelayServerTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrRelayServerTest.kt new file mode 100644 index 0000000000..9103d31810 --- /dev/null +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrRelayServerTest.kt @@ -0,0 +1,375 @@ +/* + * 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.filters.Filter +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 NostrRelayServerTest { + 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: EventStore = InMemoryEventStore(), + dispatcher: kotlinx.coroutines.CoroutineDispatcher, + ): NostrRelayServer = + NostrRelayServer( + store = store, + parentContext = dispatcher, + eventVerifier = { true }, + ) + + /** Collects sent JSON messages for a connection. */ + private class MessageCollector { + val messages = mutableListOf() + + val sendCallback: suspend (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 = InMemoryEventStore() + val server = createServer(store, dispatcher) + val collector = MessageCollector() + + server.connect("c1", collector.sendCallback) + + val event = testEvent() + val eventJson = """["EVENT",${event.toJson()}]""" + server.processMessage("c1", 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 = InMemoryEventStore() + val server = createServer(store, dispatcher) + val collector = MessageCollector() + + server.connect("c1", collector.sendCallback) + + val event = testEvent() + val eventJson = """["EVENT",${event.toJson()}]""" + server.processMessage("c1", eventJson) + server.processMessage("c1", 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 = InMemoryEventStore() + val server = createServer(store, dispatcher) + + // Pre-populate store + store.store(testEvent(hexId(1), kind = 1, createdAt = 100L)) + store.store(testEvent(hexId(2), kind = 1, createdAt = 200L)) + store.store(testEvent(hexId(3), kind = 4, createdAt = 300L)) + + val collector = MessageCollector() + server.connect("c1", collector.sendCallback) + + val reqJson = """["REQ","sub1",{"kinds":[1]}]""" + server.processMessage("c1", 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 = InMemoryEventStore() + val server = createServer(store, dispatcher) + + for (i in 1..10) { + store.store(testEvent(hexId(i), createdAt = i.toLong())) + } + + val collector = MessageCollector() + server.connect("c1", collector.sendCallback) + + val reqJson = """["REQ","sub1",{"limit":3}]""" + server.processMessage("c1", reqJson) + + val events = collector.parsedEventMessages().filterIsInstance() + assertEquals(3, events.size) + + server.shutdown() + } + + // -- Live subscription ----------------------------------------------------- + + @Test + fun liveSubscriptionReceivesNewEvents() = + runTest { + val dispatcher = UnconfinedTestDispatcher(testScheduler) + val store = InMemoryEventStore() + val server = createServer(store, dispatcher) + val collector = MessageCollector() + + server.connect("c1", collector.sendCallback) + + // Subscribe to kind 1 + val reqJson = """["REQ","sub1",{"kinds":[1]}]""" + server.processMessage("c1", reqJson) + + // After REQ, we should have EOSE + val countAfterEose = collector.messages.size + + // Now store a new event — should be pushed to subscription + store.store(testEvent(hexId(1), kind = 1)) + + val newMessages = collector.messages.drop(countAfterEose) + assertTrue(newMessages.isNotEmpty()) + assertTrue(newMessages[0].contains("\"EVENT\"")) + + server.shutdown() + } + + @Test + fun liveSubscriptionFiltersNonMatchingEvents() = + runTest { + val dispatcher = UnconfinedTestDispatcher(testScheduler) + val store = InMemoryEventStore() + val server = createServer(store, dispatcher) + val collector = MessageCollector() + + server.connect("c1", collector.sendCallback) + + val reqJson = """["REQ","sub1",{"kinds":[1]}]""" + server.processMessage("c1", reqJson) + + val countAfterEose = collector.messages.size + + // Store a kind 4 event — should NOT match kind 1 subscription + store.store(testEvent(hexId(1), kind = 4)) + + assertEquals(countAfterEose, collector.messages.size) + + server.shutdown() + } + + // -- CLOSE command --------------------------------------------------------- + + @Test + fun closeStopsSubscription() = + runTest { + val dispatcher = UnconfinedTestDispatcher(testScheduler) + val store = InMemoryEventStore() + val server = createServer(store, dispatcher) + val collector = MessageCollector() + + server.connect("c1", collector.sendCallback) + + val reqJson = """["REQ","sub1",{"kinds":[1]}]""" + server.processMessage("c1", reqJson) + + // Close the subscription + val closeJson = """["CLOSE","sub1"]""" + server.processMessage("c1", closeJson) + + val countAfterClose = collector.messages.size + + // New events should NOT reach this subscription + store.store(testEvent(hexId(1), kind = 1)) + + assertEquals(countAfterClose, collector.messages.size) + + server.shutdown() + } + + @Test + fun replacingSubscriptionCancelsOld() = + runTest { + val dispatcher = UnconfinedTestDispatcher(testScheduler) + val store = InMemoryEventStore() + val server = createServer(store, dispatcher) + val collector = MessageCollector() + + server.connect("c1", collector.sendCallback) + + // First subscription for kind 1 + server.processMessage("c1", """["REQ","sub1",{"kinds":[1]}]""") + + // Replace with kind 4 + server.processMessage("c1", """["REQ","sub1",{"kinds":[4]}]""") + + val countAfterReplace = collector.messages.size + + // Kind 1 events should not match anymore + store.store(testEvent(hexId(1), kind = 1)) + assertEquals(countAfterReplace, collector.messages.size) + + // Kind 4 events should match + store.store(testEvent(hexId(2), kind = 4)) + + val newMessages = collector.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 = InMemoryEventStore() + val server = createServer(store, dispatcher) + + store.store(testEvent(hexId(1), kind = 1)) + store.store(testEvent(hexId(2), kind = 1)) + store.store(testEvent(hexId(3), kind = 4)) + + val collector = MessageCollector() + server.connect("c1", collector.sendCallback) + + val countJson = """["COUNT","q1",{"kinds":[1]}]""" + server.processMessage("c1", 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 = InMemoryEventStore() + val server = createServer(store, dispatcher) + val collector = MessageCollector() + + server.connect("c1", collector.sendCallback) + + server.processMessage("c1", """["REQ","sub1",{"kinds":[1]}]""") + server.processMessage("c1", """["REQ","sub2",{"kinds":[4]}]""") + + server.disconnect("c1") + + val countAfterDisconnect = collector.messages.size + + store.store(testEvent(hexId(1), kind = 1)) + store.store(testEvent(hexId(2), kind = 4)) + + assertEquals(countAfterDisconnect, collector.messages.size) + + server.shutdown() + } + + // -- Invalid messages ------------------------------------------------------ + + @Test + fun invalidJsonReturnsNotice() = + runTest { + val dispatcher = UnconfinedTestDispatcher(testScheduler) + val server = createServer(dispatcher = dispatcher) + val collector = MessageCollector() + + server.connect("c1", collector.sendCallback) + server.processMessage("c1", "not valid json") + + assertEquals(1, collector.messages.size) + assertTrue(collector.messages[0].contains("NOTICE")) + + server.shutdown() + } +}