Merge pull request #3883 from vitorpamplona/fix/quartz-nip66-prober-merge

fix(quartz): merge probe verdicts into the record they replace

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProber.kt

@@ -39,6 +39,8 @@ import com.vitorpamplona.quartz.nip66RelayMonitor.discovery.RelayDiscoveryEvent
 import com.vitorpamplona.quartz.nip66RelayMonitor.discovery.networkType
 import com.vitorpamplona.quartz.nip66RelayMonitor.discovery.requirement
 import com.vitorpamplona.quartz.nip66RelayMonitor.discovery.rtt
+import com.vitorpamplona.quartz.nip66RelayMonitor.discovery.tags.NetworkTypeTag
+import com.vitorpamplona.quartz.nip66RelayMonitor.discovery.tags.RequirementTag
 import com.vitorpamplona.quartz.nip66RelayMonitor.discovery.tags.RttType
 import com.vitorpamplona.quartz.utils.TimeUtils
 import com.vitorpamplona.quartz.utils.concurrent.ConcurrentMap
@@ -430,8 +432,18 @@ class RelayProber(
 fun RelayProber.Verdict.toDiscoveryEventTemplate(
     createdAt: Long = TimeUtils.now(),
     readWrite: RelayProber.ReadWriteVerdict? = null,
+    current: RelayDiscoveryEvent? = null,
 ): EventTemplate<RelayDiscoveryEvent> =
-    RelayDiscoveryEvent.build(relay, createdAt = createdAt) {
+    RelayDiscoveryEvent.build(
+        relay,
+        current?.content ?: "",
+        // Strictly newer than what it replaces, or a store enforcing
+        // replaceable semantics rejects it and the probe is lost silently.
+        createdAt = maxOf(createdAt, (current?.createdAt ?: 0L) + 1),
+    ) {
+        current?.tags?.forEach { tag ->
+            if (tag.firstOrNull() != "d" && !probeOwns(tag, readWrite != null)) add(tag)
+        }
         networkType(RelayReachabilityStore.networkTypeOf(relay))
         if (reachable) rtt(RttType.OPEN, rttOpenMs.coerceAtLeast(0))
         if (readWrite != null) {
@@ -441,6 +453,37 @@ fun RelayProber.Verdict.toDiscoveryEventTemplate(
         val authWalled =
             error?.startsWith("closed:auth-required") == true ||
                 readWrite?.writeMessage?.startsWith("auth-required") == true
-        if (authWalled) requirement("auth")
-        if (readWrite?.writeMessage?.startsWith("pow:") == true) requirement("pow")
+        if (authWalled) requirement(RelayReachabilityStore.AUTH_REQUIREMENT)
+        if (readWrite?.writeMessage?.startsWith("pow:") == true) requirement(POW_REQUIREMENT)
+    }
+
+/** The NIP-66 requirement a write probe can prove, alongside `auth`. */
+private const val POW_REQUIREMENT = "pow"
+
+/**
+ * What a probe verdict measured, and may therefore replace in [current].
+ *
+ * A 30166 carries ONE `created_at`, so any tag carried across is re-dated as a
+ * current measurement — this verdict's own facts must be rewritten wholesale or
+ * a stale latency is republished as today's number.
+ *
+ * [hasReadWrite] narrows it: without a [RelayProber.ReadWriteVerdict] this probe
+ * never exercised the write path, so `R pow` is somebody else's finding and is
+ * carried across rather than deleted. Both polarities of each requirement are
+ * owned, so an update cannot leave the record asserting `pow` and `!pow` at once.
+ */
+private fun probeOwns(
+    tag: Array<String>,
+    hasReadWrite: Boolean,
+): Boolean =
+    when (tag.firstOrNull()) {
+        NetworkTypeTag.TAG_NAME -> true
+        RttType.OPEN.tagName, RttType.READ.tagName, RttType.WRITE.tagName -> true
+        RequirementTag.TAG_NAME ->
+            when (tag.getOrNull(1)) {
+                RelayReachabilityStore.AUTH_REQUIREMENT, "!" + RelayReachabilityStore.AUTH_REQUIREMENT -> true
+                POW_REQUIREMENT, "!" + POW_REQUIREMENT -> hasReadWrite
+                else -> false
+            }
+        else -> false
     }

quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/RelayProberFlowTest.kt

@@ -34,6 +34,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
 import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer
 import com.vitorpamplona.quartz.nip01Core.signers.EventTemplate
 import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerInternal
+import com.vitorpamplona.quartz.nip66RelayMonitor.discovery.RelayDiscoveryEvent
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.launch
@@ -472,4 +473,58 @@ class RelayProberFlowTest {
 
             assertTrue(listOf("n", "tor") in tagsOf(template))
         }
+    // ---- merging into an existing record ----------------------------------
+
+    /**
+     * A 30166 is addressable, so a consumer that follows this function's KDoc —
+     * sign with the monitor key, insert — replaces whatever else is on that
+     * address. Built from the verdict alone it deletes it.
+     */
+    @Test
+    fun probeTemplateKeepsTagsItDidNotMeasure() {
+        val verdict = RelayProber.Verdict(fast, reachable = true, rttOpenMs = 120, rttEoseMs = 200, error = null)
+        val existing =
+            RelayDiscoveryEvent(
+                "id",
+                "pubkey",
+                1_000,
+                arrayOf(
+                    arrayOf("d", fast.url),
+                    arrayOf("R", "pow"),
+                    arrayOf("redirect", "wss://canonical.example.com/"),
+                ),
+                "",
+                "sig",
+            )
+
+        val tags = tagsOf(verdict.toDiscoveryEventTemplate(createdAt = 2_000, current = existing))
+
+        assertTrue(listOf("redirect", "wss://canonical.example.com/") in tags, "a foreign tag was deleted: $tags")
+        // No write probe ran, so `R pow` is somebody else's finding.
+        assertTrue(listOf("R", "pow") in tags, "an unmeasured requirement was deleted: $tags")
+    }
+
+    /** With a write verdict the probe DOES measure pow, so a stale one must not be re-dated. */
+    @Test
+    fun probeTemplateReplacesRequirementsItDidMeasure() {
+        val verdict = RelayProber.Verdict(fast, reachable = true, rttOpenMs = 120, rttEoseMs = 200, error = null)
+        val existing =
+            RelayDiscoveryEvent("id", "pubkey", 1_000, arrayOf(arrayOf("d", fast.url), arrayOf("R", "pow")), "", "sig")
+        val clean = RelayProber.ReadWriteVerdict(fast, rttReadMs = 10, rttWriteMs = 20, writeAccepted = true, writeMessage = null)
+
+        val tags = tagsOf(verdict.toDiscoveryEventTemplate(createdAt = 2_000, readWrite = clean, current = existing))
+
+        assertTrue(listOf("R", "pow") !in tags, "a stale requirement was carried onto a fresh measurement: $tags")
+    }
+
+    /** Replaceable ordering: an update not strictly newer is rejected and lost. */
+    @Test
+    fun probeTemplateStampsPastTheRecordItReplaces() {
+        val verdict = RelayProber.Verdict(fast, reachable = true, rttOpenMs = 120, rttEoseMs = 200, error = null)
+        val existing = RelayDiscoveryEvent("id", "pubkey", 9_000, arrayOf(arrayOf("d", fast.url)), "", "sig")
+
+        val template = verdict.toDiscoveryEventTemplate(createdAt = 2_000, current = existing)
+
+        assertTrue(template.createdAt > 9_000, "stamped ${template.createdAt}, which cannot replace 9000")
+    }
 }