feat(geode): per-upstream [[mirror]] filters, strfry-router parity

Each [[mirror]] entry takes an optional NIP-01 filter as a JSON object
string, like strfry-router's per-stream filter. It is applied twice,
matching strfry's design (cmd_router.cpp):

 - it shapes the REQ sent upstream (the mirror still owns since via
   backfill_seconds and strips limit — the subscription is unbounded);
 - every delivered event is re-checked against it before ingest, so an
   upstream answering outside its REQ — including a trusted one whose
   events skip signature verification — can only inject events inside
   the operator-declared scope. Out-of-scope deliveries surface on a
   'filtered' counter.

Malformed filter JSON fails the boot, not the first delivery. Several
disjoint scopes for one upstream = repeat [[mirror]] with the same url.

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

geode/config.example.toml

@@ -99,10 +99,20 @@ require_auth = false
 # mirror-but-verify. Only trust relays you operate or whose ingest
 # discipline you'd stake your own db on.
 #
+# `filter` (optional) scopes an upstream, as a NIP-01 filter JSON
+# object — same idea as strfry-router's per-stream filter. It shapes
+# the REQ sent upstream AND every delivered event is re-checked
+# against it before ingest, so even a trusted upstream can only
+# inject events inside the declared scope. `since`/`limit` inside it
+# are ignored (backfill_seconds owns the time window). Omit to mirror
+# everything; for several disjoint scopes, repeat [[mirror]] with the
+# same url.
+#
 # [[mirror]]
 # url = "wss://upstream.example.com/"
 # trusted = true
 # backfill_seconds = 3600
+# filter = '{"kinds":[0,1,3,7],"#t":["nostr"]}'
 
 [admin]
 # NIP-86 relay management API. When `pubkeys` is non-empty, the relay

geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt

@@ -26,6 +26,8 @@ import com.vitorpamplona.geode.config.RuntimeConfigData
 import com.vitorpamplona.geode.config.StaticConfig
 import com.vitorpamplona.geode.mirror.MirrorUpstream
 import com.vitorpamplona.geode.mirror.MirrorWorker
+import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper
+import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
 import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
 import com.vitorpamplona.quartz.nip01Core.relay.normalizer.normalizeRelayUrl
 import com.vitorpamplona.quartz.nip01Core.relay.server.policies.EmptyPolicy
@@ -181,11 +183,29 @@ fun main(args: Array<String>) {
     // anyway. Never mirror ourselves: a self-URL would echo every local
     // publish back forever.
     val upstreams =
-        config.mirror.map {
+        config.mirror.map { m ->
+            // The optional scope filter is parsed eagerly so a malformed
+            // JSON object fails the boot, not the first delivery. Its
+            // since/limit are stripped: the mirror owns the time window
+            // (backfill_seconds) and never bounds the subscription.
+            val scope =
+                m.filter?.let { json ->
+                    val parsed =
+                        try {
+                            OptimizedJsonMapper.fromJsonTo<Filter>(json)
+                        } catch (e: Exception) {
+                            throw IllegalArgumentException(
+                                "[[mirror]] filter for ${m.url} is not a valid NIP-01 filter object: $json",
+                                e,
+                            )
+                        }
+                    parsed.copy(since = null, limit = null)
+                }
             MirrorUpstream(
-                url = it.url.normalizeRelayUrl(),
-                trusted = it.trusted,
-                backfillSeconds = it.backfill_seconds,
+                url = m.url.normalizeRelayUrl(),
+                trusted = m.trusted,
+                backfillSeconds = m.backfill_seconds,
+                filter = scope,
             )
         }
     require(upstreams.none { it.url == advertisedUrl }) {

geode/src/main/kotlin/com/vitorpamplona/geode/config/StaticConfig.kt

@@ -181,6 +181,18 @@ data class StaticConfig(
         val trusted: Boolean = false,
         /** How far back the initial subscription reaches. 0 = live-only. */
         val backfill_seconds: Long = 0L,
+        /**
+         * Optional NIP-01 filter as a JSON object string (strfry-router
+         * parity), e.g. `'{"kinds":[0,1,3],"#t":["nostr"]}'`. Scopes what
+         * this upstream is asked for AND what it is allowed to deliver —
+         * every received event is re-checked against it before ingest, so
+         * even a trusted upstream can't push events outside the declared
+         * scope. `since` is managed by the mirror (see [backfill_seconds])
+         * and `limit` is transport-level, so both are ignored if present.
+         * Omitted = mirror everything. For several disjoint filters, add
+         * several `[[mirror]]` entries with the same url.
+         */
+        val filter: String? = null,
     )
 
     /**

geode/src/main/kotlin/com/vitorpamplona/geode/mirror/MirrorWorker.kt

@@ -55,6 +55,16 @@ class MirrorUpstream(
     val trusted: Boolean,
     /** How far back the initial REQ reaches. 0 = live-only from connect. */
     val backfillSeconds: Long = 0L,
+    /**
+     * Optional scope for this upstream (strfry-router's per-stream
+     * `filter`). Used twice: it shapes the REQ sent upstream, and every
+     * delivered event is re-checked against it before ingest — so even a
+     * [trusted] upstream can only inject events inside the declared
+     * scope. Its `since`/`limit` are ignored ([backfillSeconds] owns the
+     * time window; the subscription is unbounded). `null` mirrors
+     * everything.
+     */
+    val filter: Filter? = null,
 )
 
 /**
@@ -113,6 +123,13 @@ class MirrorWorker(
     /** Events the store rejected — mostly duplicate replays after a reconnect. */
     val rejected = AtomicLong(0)
 
+    /**
+     * Deliveries dropped by the [MirrorUpstream.filter] re-check before
+     * ever reaching the store — an upstream sending these is answering
+     * outside the REQ it was given.
+     */
+    val filtered = AtomicLong(0)
+
     /** Dials every upstream and starts streaming. Call once. */
     fun start() {
         scope.launch {
@@ -151,6 +168,17 @@ class MirrorWorker(
                         relay: NormalizedRelayUrl,
                         forFilters: List<Filter>?,
                     ) {
+                        // strfry-router parity: never take the upstream's
+                        // word for what matched. Re-checking the configured
+                        // scope here means even a trusted (skip-verify)
+                        // upstream can only inject events the operator
+                        // declared — the REQ shapes what we ask for, this
+                        // shapes what we accept.
+                        if (up.filter != null && !up.filter.match(event)) {
+                            filtered.incrementAndGet()
+                            Log.d("MirrorWorker") { "out-of-scope from ${relay.url}: ${event.id}" }
+                            return
+                        }
                         inbound.trySend(Inbound(event, up.trusted))
                     }
 
@@ -162,9 +190,16 @@ class MirrorWorker(
                         Log.w("MirrorWorker") { "cannot reach upstream ${relay.url}: $message" }
                     }
                 }
+            // The operator's filter scopes the REQ; the mirror owns the
+            // time window (since) and never bounds the result (limit).
+            val reqFilter =
+                (up.filter ?: Filter()).copy(
+                    since = since - up.backfillSeconds,
+                    limit = null,
+                )
             client.subscribe(
                 subId = "geode-mirror-$i",
-                filters = mapOf(up.url to listOf(Filter(since = since - up.backfillSeconds))),
+                filters = mapOf(up.url to listOf(reqFilter)),
                 listener = listener,
             )
         }

geode/src/test/kotlin/com/vitorpamplona/geode/config/StaticConfigTest.kt

@@ -20,6 +20,8 @@
  */
 package com.vitorpamplona.geode.config
 
+import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper
+import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
 import java.io.File
 import kotlin.test.Test
 import kotlin.test.assertEquals
@@ -51,6 +53,16 @@ class StaticConfigTest {
         assertTrue(StaticConfig.fromToml("").mirror.isEmpty())
     }
 
+    @Test
+    fun mirrorFilterJsonParsesToANip01Filter() {
+        // The exact parse Main.kt runs on [[mirror]].filter at boot.
+        val f = OptimizedJsonMapper.fromJsonTo<Filter>("""{"kinds":[0,1],"#t":["nostr"],"since":123,"limit":5}""")
+        assertEquals(listOf(0, 1), f.kinds)
+        assertEquals(listOf("nostr"), f.tags?.get("t"))
+        assertEquals(123L, f.since)
+        assertEquals(5, f.limit)
+    }
+
     @Test
     fun parsesMirrorUpstreams() {
         val toml =
@@ -59,6 +71,7 @@ class StaticConfigTest {
             url = "wss://trusted.upstream.example/"
             trusted = true
             backfill_seconds = 3600
+            filter = '{"kinds":[0,1,3],"#t":["nostr"]}'
 
             [[mirror]]
             url = "wss://public.upstream.example/"
@@ -70,10 +83,13 @@ class StaticConfigTest {
         assertEquals("wss://trusted.upstream.example/", c.mirror[0].url)
         assertEquals(true, c.mirror[0].trusted)
         assertEquals(3600L, c.mirror[0].backfill_seconds)
-        // Trust is opt-in per upstream: the default is mirror-but-verify.
+        assertEquals("""{"kinds":[0,1,3],"#t":["nostr"]}""", c.mirror[0].filter)
+        // Trust and scoping are opt-in per upstream: the default is
+        // mirror-everything-but-verify.
         assertEquals("wss://public.upstream.example/", c.mirror[1].url)
         assertEquals(false, c.mirror[1].trusted)
         assertEquals(0L, c.mirror[1].backfill_seconds)
+        assertEquals(null, c.mirror[1].filter)
     }
 
     @Test

geode/src/test/kotlin/com/vitorpamplona/geode/mirror/MirrorWorkerTest.kt

@@ -77,9 +77,12 @@ class MirrorWorkerTest {
         hub.close()
     }
 
-    private fun startMirror(trusted: Boolean): MirrorWorker =
+    private fun startMirror(
+        trusted: Boolean,
+        filter: Filter? = null,
+    ): MirrorWorker =
         MirrorWorker(
-            upstreams = listOf(MirrorUpstream(upstreamUrl, trusted = trusted, backfillSeconds = 3600)),
+            upstreams = listOf(MirrorUpstream(upstreamUrl, trusted = trusted, backfillSeconds = 3600, filter = filter)),
             server = downstream.server,
             websocketBuilder = hub,
         ).also {
@@ -87,12 +90,15 @@ class MirrorWorkerTest {
             it.start()
         }
 
-    private fun forgedEvent(idSeed: Int): Event =
+    private fun forgedEvent(
+        idSeed: Int,
+        kind: Int = 1,
+    ): Event =
         Event(
             id = idSeed.toString().padStart(64, '0'),
             pubKey = "1".repeat(64),
             createdAt = TimeUtils.now() - idSeed,
-            kind = 1,
+            kind = kind,
             tags = emptyArray(),
             content = "forged $idSeed",
             sig = "f".repeat(128),
@@ -144,6 +150,29 @@ class MirrorWorkerTest {
             assertEquals(setOf(stored.id, live.id), ids)
         }
 
+    @Test
+    fun filterScopesTheMirrorToDeclaredKinds() =
+        runBlocking {
+            val wantedStored = forgedEvent(1, kind = 1)
+            val unwantedStored = forgedEvent(2, kind = 7)
+            hub.getOrCreate(upstreamUrl).preload(wantedStored, unwantedStored)
+
+            startMirror(trusted = true, filter = Filter(kinds = listOf(1)))
+            awaitDownstreamCount(1)
+
+            // Live tail: the out-of-scope kind is published FIRST, so by the
+            // time the in-scope one lands downstream (same connection, same
+            // ordered pipeline), the kind-7 has already had its chance.
+            hub.getOrCreate(upstreamUrl).publish(forgedEvent(3, kind = 7))
+            val wantedLive = forgedEvent(4, kind = 1)
+            hub.getOrCreate(upstreamUrl).publish(wantedLive)
+            awaitDownstreamCount(2)
+
+            val stored = downstreamStore.query<Event>(Filter())
+            assertEquals(setOf(wantedStored.id, wantedLive.id), stored.map { it.id }.toSet())
+            assertTrue(stored.all { it.kind == 1 })
+        }
+
     @Test
     fun untrustedUpstreamStillVerifiesEverything() =
         runBlocking {