feat: add NostrRelayServer implementing NIP-01 and NIP-45

Transport-agnostic relay server class in quartz that:
- Handles EVENT, REQ, CLOSE (NIP-01) and COUNT (NIP-45) commands
- Manages subscriptions as coroutines with immediate cancel on CLOSE
- Replays stored events then streams live events via SharedFlow
- Includes EventStore interface and InMemoryEventStore with support
  for regular, replaceable, addressable, and ephemeral event semantics
- Comprehensive test coverage for all protocol commands and store logic

https://claude.ai/code/session_013A1bn4DZUKYtEzZ5BWUXVh

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<Event>
+
+    /**
+     * 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<Event>
+
+    /**
+     * Returns the number of stored events matching [filter] (NIP-45).
+     */
+    suspend fun count(filter: Filter): Int
+}

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<HexKey, Event>()
+
+    /** Newest replaceable event per "pubKey:kind". */
+    private val replaceableIndex = HashMap<String, HexKey>()
+
+    /** Newest addressable event per "pubKey:kind:dTag". */
+    private val addressableIndex = HashMap<String, HexKey>()
+
+    private val _newEvents = MutableSharedFlow<Event>(extraBufferCapacity = 256)
+    override val newEvents: SharedFlow<Event> = _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<Event> =
+        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) }
+        }
+}

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<String, ClientSession>()
+    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<String, Job>()
+    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()
+        }
+    }
+}

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<Array<String>> = 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)
+        }
+}

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<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: 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<String>()
+
+        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<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 = 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<EventMessage>()
+            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()
+        }
+}