feat(quartz): NostrServer.ingest — local write path with per-submission verify skip

Adds a public local-ingest entry to NostrServer for events that don't
arrive over a client connection (mirror/sync workers, import jobs). It
routes through the same group-commit IngestQueue and live fanout as a
client publish, and each Submission can opt out of the parallel
Schnorr-verify hook — the relay-to-relay trust model, for events
streamed from an upstream that already verified them (verify profiles
at ~8% of busy ingest CPU).

Library defaults are unchanged: skipVerify defaults to false everywhere
and nothing in the client-publish path can set it.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TtDNpayEYvJH7QuPswND3A

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

@@ -20,6 +20,7 @@
  */
 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.relay.server.backend.IngestQueue
 import com.vitorpamplona.quartz.nip01Core.relay.server.backend.LiveEventStore
@@ -95,7 +96,28 @@ class NostrServer(
                 },
         )
 
-    override val backend: SessionBackend = LiveEventStore(store, ingest)
+    private val liveStore = LiveEventStore(store, ingest)
+
+    override val backend: SessionBackend = liveStore
+
+    /**
+     * Local ingestion path for events that did not arrive over a client
+     * connection — e.g. a mirror worker streaming a trusted upstream
+     * relay, or an import job. Routes through the same group-commit
+     * [IngestQueue] and live fanout as a client EVENT publish, but skips
+     * the per-connection policy chain (there is no connection).
+     *
+     * [skipVerify] exempts the event from the parallel signature-verify
+     * hook — the relay-to-relay trust model: pass `true` only for events
+     * from an explicitly configured upstream that already verified them
+     * (Schnorr verify profiles at ~8% of busy ingest CPU). The default
+     * `false` keeps verify-everything semantics.
+     */
+    suspend fun ingest(
+        event: Event,
+        skipVerify: Boolean = false,
+        onComplete: (IEventStore.InsertOutcome) -> Unit,
+    ) = liveStore.submit(event, skipVerify, onComplete)
 
     init {
         // Deferred-FTS catch-up worker: tokenizes in the gaps between

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

@@ -27,7 +27,6 @@ import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.async
-import kotlinx.coroutines.awaitAll
 import kotlinx.coroutines.cancel
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.ClosedReceiveChannelException
@@ -115,9 +114,15 @@ class IngestQueue(
     /**
      * One outstanding ingest request: the event to insert plus the
      * callback the writer fires once the row's outcome is known.
+     * [skipVerify] exempts this row from the [verify] hook — the
+     * relay-to-relay trust model: set by local ingestion paths for
+     * events streamed from an explicitly configured upstream relay
+     * that already verified them (see
+     * [com.vitorpamplona.quartz.nip01Core.relay.server.NostrServer.ingest]).
      */
     class Submission(
         val event: Event,
+        val skipVerify: Boolean,
         val onComplete: (IEventStore.InsertOutcome) -> Unit,
     )
 
@@ -159,11 +164,12 @@ class IngestQueue(
      */
     suspend fun submit(
         event: Event,
+        skipVerify: Boolean = false,
         onComplete: (IEventStore.InsertOutcome) -> Unit,
     ) {
         ensureWriterStarted()
         pending.addAndFetch(1)
-        incoming.send(Submission(event, onComplete))
+        incoming.send(Submission(event, skipVerify, onComplete))
     }
 
     private fun ensureWriterStarted() {
@@ -234,15 +240,26 @@ class IngestQueue(
      * configured (skip the stage entirely). For multi-event batches
      * each verify runs as its own `async(Default)` so they spread
      * across CPU cores; single-event batches short-circuit to a
-     * direct call to avoid coroutine-scope overhead.
+     * direct call to avoid coroutine-scope overhead. Rows flagged
+     * [Submission.skipVerify] (trusted publishers) pass without
+     * invoking the hook.
      */
     private suspend fun verifyBatch(batch: List<Submission>): BooleanArray? {
         val hook = verify ?: return null
-        if (batch.size == 1) return BooleanArray(1) { hook(batch[0].event) }
+        if (batch.size == 1) {
+            val sub = batch[0]
+            return BooleanArray(1) { sub.skipVerify || hook(sub.event) }
+        }
+        if (batch.all { it.skipVerify }) return BooleanArray(batch.size) { true }
         return coroutineScope {
             batch
-                .map { sub -> async(Dispatchers.Default) { hook(sub.event) } }
-                .awaitAll()
+                .map { sub ->
+                    if (sub.skipVerify) {
+                        null
+                    } else {
+                        async(Dispatchers.Default) { hook(sub.event) }
+                    }
+                }.map { it?.await() ?: true }
                 .toBooleanArray()
         }
     }

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

@@ -91,8 +91,23 @@ class LiveEventStore(
     override suspend fun submit(
         event: Event,
         onComplete: (IEventStore.InsertOutcome) -> Unit,
+    ) = submit(event, skipVerify = false, onComplete = onComplete)
+
+    /**
+     * [submit] variant for locally-originated traffic (mirror/sync
+     * workers rather than client connections). [skipVerify] exempts
+     * this event from the [IngestQueue]'s signature-verification hook —
+     * the relay-to-relay trust model: set it only for events streamed
+     * from a configured upstream relay that already verified them.
+     * Accepted events fan out to live subscribers exactly like a
+     * client publish.
+     */
+    suspend fun submit(
+        event: Event,
+        skipVerify: Boolean,
+        onComplete: (IEventStore.InsertOutcome) -> Unit,
     ) {
-        ingest.submit(event) { outcome ->
+        ingest.submit(event, skipVerify) { outcome ->
             if (outcome is IEventStore.InsertOutcome.Accepted) {
                 writeGeneration.addAndFetch(1L)
                 fanout(event)

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

@@ -0,0 +1,162 @@
+/*
+ * 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.KeyPair
+import com.vitorpamplona.quartz.nip01Core.relay.server.policies.EmptyPolicy
+import com.vitorpamplona.quartz.nip01Core.store.IEventStore
+import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore
+import com.vitorpamplona.quartz.utils.DeterministicSigner
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.CoroutineDispatcher
+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
+
+/**
+ * The local-ingest path ([NostrServer.ingest]) and its relay-to-relay
+ * trust switch: `skipVerify = true` (a mirror streaming from a trusted
+ * upstream) must land forged-signature events, while the default keeps
+ * verify-everything semantics through the IngestQueue's parallel hook.
+ */
+@OptIn(ExperimentalCoroutinesApi::class)
+class NostrServerIngestTest {
+    private val signer = DeterministicSigner(KeyPair())
+
+    private fun signedEvent(content: String = "hello"): Event = signer.sign(createdAt = 1000L, kind = 1, tags = emptyArray(), content = content)
+
+    /** A structurally valid event whose signature verifies against nothing. */
+    private fun forgedEvent(id: Int = 1): Event =
+        Event(
+            id = id.toString().padStart(64, '0'),
+            pubKey = signer.pubKey,
+            createdAt = 1000L + id,
+            kind = 1,
+            tags = emptyArray(),
+            content = "forged",
+            sig = "f".repeat(128),
+        )
+
+    private fun createServer(dispatcher: CoroutineDispatcher): NostrServer =
+        NostrServer(
+            store = EventStore(null),
+            policyBuilder = { EmptyPolicy },
+            parentContext = dispatcher,
+            parallelVerify = true,
+        )
+
+    private suspend fun NostrServer.ingestOutcome(
+        event: Event,
+        skipVerify: Boolean,
+    ): IEventStore.InsertOutcome {
+        val outcome = CompletableDeferred<IEventStore.InsertOutcome>()
+        ingest(event, skipVerify) { outcome.complete(it) }
+        return outcome.await()
+    }
+
+    @Test
+    fun forgedEventIsRejectedByDefault() =
+        runTest {
+            val server = createServer(UnconfinedTestDispatcher(testScheduler))
+
+            val outcome = server.ingestOutcome(forgedEvent(), skipVerify = false)
+
+            assertTrue(outcome is IEventStore.InsertOutcome.Rejected)
+            assertTrue(outcome.reason.contains("signature"))
+
+            server.close()
+        }
+
+    @Test
+    fun forgedEventLandsWhenTrusted() =
+        runTest {
+            val server = createServer(UnconfinedTestDispatcher(testScheduler))
+
+            val outcome = server.ingestOutcome(forgedEvent(), skipVerify = true)
+
+            assertEquals(IEventStore.InsertOutcome.Accepted, outcome)
+
+            server.close()
+        }
+
+    @Test
+    fun validEventLandsEitherWay() =
+        runTest {
+            val server = createServer(UnconfinedTestDispatcher(testScheduler))
+
+            assertEquals(
+                IEventStore.InsertOutcome.Accepted,
+                server.ingestOutcome(signedEvent("via verify"), skipVerify = false),
+            )
+            assertEquals(
+                IEventStore.InsertOutcome.Accepted,
+                server.ingestOutcome(signedEvent("via trust"), skipVerify = true),
+            )
+
+            server.close()
+        }
+
+    @Test
+    fun trustIsPerSubmissionNotPerQueue() =
+        runTest {
+            // A trusted mirror and an untrusted publisher share the same
+            // IngestQueue; the skip must apply row-by-row, never leak from
+            // one submission to the next.
+            val server = createServer(UnconfinedTestDispatcher(testScheduler))
+
+            assertEquals(
+                IEventStore.InsertOutcome.Accepted,
+                server.ingestOutcome(forgedEvent(1), skipVerify = true),
+            )
+            val untrusted = server.ingestOutcome(forgedEvent(2), skipVerify = false)
+            assertTrue(untrusted is IEventStore.InsertOutcome.Rejected)
+
+            server.close()
+        }
+
+    @Test
+    fun trustedIngestFansOutToLiveSubscribers() =
+        runTest {
+            // Mirror traffic must feed live REQs exactly like a client
+            // publish — the skip changes verification, not delivery.
+            val dispatcher = UnconfinedTestDispatcher(testScheduler)
+            val server = createServer(dispatcher)
+
+            val received = mutableListOf<String>()
+            val session = server.connect { received.add(it) }
+            session.receive("""["REQ","live",{"kinds":[1]}]""")
+            assertTrue(received.any { it.startsWith("[\"EOSE\"") })
+
+            val forged = forgedEvent()
+            assertEquals(
+                IEventStore.InsertOutcome.Accepted,
+                server.ingestOutcome(forged, skipVerify = true),
+            )
+
+            assertTrue(received.any { it.startsWith("[\"EVENT\",\"live\"") && it.contains(forged.id) })
+
+            server.close()
+        }
+}