Merge pull request #3857 from vitorpamplona/claude/relay-url-normalizer-mggnkh

NIP-66 relay monitoring: streaming probes, read/write checks, URL fixes

cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/RelayCommands.kt

@@ -28,6 +28,7 @@ import com.vitorpamplona.quartz.marmot.mip00KeyPackages.KeyPackageRelayListEvent
 import com.vitorpamplona.quartz.nip01Core.core.Event
 import com.vitorpamplona.quartz.nip01Core.core.HexKey
 import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
+import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer
 import com.vitorpamplona.quartz.nip01Core.relay.normalizer.normalizeRelayUrlOrNull
 import com.vitorpamplona.quartz.nip01Core.relay.normalizer.toHttp
 import com.vitorpamplona.quartz.nip11RelayInfo.Nip11RelayInformation
@@ -50,6 +51,7 @@ import com.vitorpamplona.quartz.nip65RelayList.tags.AdvertisedRelayType
 import com.vitorpamplona.quartz.nip66RelayMonitor.reachability.RelayProber
 import okhttp3.OkHttpClient
 import okhttp3.Request
+import java.io.File
 
 /**
  * `amy relay …` — manage every relay list this account maintains, mirroring
@@ -112,10 +114,12 @@ object RelayCommands {
         |  relay info URL                fetch + print a relay's NIP-11 info document (stateless)
         |  relay probe [--timeout SECS]  relay census: mass-connect every relay the store
         |    [--concurrency N]            knows and record live/dead + measured rtt-open
-        |                                 into the reachability cache (NIP-66 kind:30166),
+        |    [--file PATH]                into the reachability cache (NIP-66 kind:30166),
         |                                 so reachability-aware commands (graperank crawl/
         |                                 refresh) skip dead relays and wait once
-        |                                 (--timeout: per wave, default 15s)
+        |                                 (--timeout: per wave, default 15s; --file: also
+        |                                 probe candidate urls, one per line, each run
+        |                                 through the relay url normalizer first)
         """.trimMargin()
 
     // ------------------------------------------------------------------
@@ -337,12 +341,41 @@ object RelayCommands {
         // Relays dialed at once; --relay-concurrency accepted as the alias the
         // graperank verbs spell it with.
         val waveSize = args.intFlag("concurrency", args.intFlag("relay-concurrency", Context.defaultPreconnectCap))
+        // Optional external candidate list: one raw url per line, run through the
+        // same RelayUrlNormalizer the app uses, so a probe doubles as a census of
+        // how a corpus of relay hints normalizes (rejects are counted, not dialed).
+        val fromFile = args.flag("file")
         args.rejectUnknown()
 
+        var fileRaw = 0
+        var fileRejected = 0
+        var fileOnion = 0
+        val fileRelays = HashSet<NormalizedRelayUrl>()
+        if (fromFile != null) {
+            val candidates = File(fromFile)
+            if (!candidates.canRead()) return Output.error("bad_args", "cannot read --file $fromFile")
+            candidates.forEachLine { line ->
+                if (line.isBlank()) return@forEachLine
+                fileRaw++
+                val normalized = line.normalizeRelayUrlOrNull()
+                if (normalized == null) {
+                    fileRejected++
+                } else if (RelayUrlNormalizer.isOnion(normalized.url)) {
+                    fileOnion++
+                } else {
+                    fileRelays.add(normalized)
+                }
+            }
+            System.err.println(
+                "[relay-probe] $fromFile: $fileRaw urls → ${fileRelays.size} unique clearnet relays " +
+                    "($fileRejected rejected by the normalizer, $fileOnion onion skipped)",
+            )
+        }
+
         Context.openOrAnonymous(dataDir).use { ctx ->
             ctx.prepare()
             val cached = ctx.reachability.snapshot()
-            val universe = RelayProber.knownRelayUniverse(ctx.store) + cached.live + cached.dead
+            val universe = RelayProber.knownRelayUniverse(ctx.store) + cached.live + cached.dead + fileRelays
             if (universe.isEmpty()) {
                 Output.emit(
                     linkedMapOf<String, Any?>(
@@ -381,6 +414,10 @@ object RelayCommands {
             Output.emit(
                 linkedMapOf<String, Any?>(
                     "probed" to result.verdicts.size,
+                    "file_urls" to (if (fromFile != null) fileRaw else null),
+                    "file_normalized" to (if (fromFile != null) fileRelays.size else null),
+                    "file_rejected" to (if (fromFile != null) fileRejected else null),
+                    "file_onion_skipped" to (if (fromFile != null) fileOnion else null),
                     "reachable" to result.reachable.size,
                     "dead" to result.dead.size,
                     "closed_by_policy" to authWalled,

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientPublishExt.kt

@@ -34,6 +34,7 @@ import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED
 import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.withTimeoutOrNull
+import kotlin.time.TimeSource
 
 /**
  * One relay's verdict on a published event: [accepted] plus the reason the
@@ -45,6 +46,12 @@ import kotlinx.coroutines.withTimeoutOrNull
 class PublishResult(
     val accepted: Boolean,
     val message: String,
+    /**
+     * Milliseconds from the publish to this relay's OK (true or false — a rejection
+     * is still a measured round trip), or -1 when the relay never answered with an
+     * OK. On an already-open socket this is an honest NIP-66 `rtt-write`.
+     */
+    val elapsedMs: Long = -1,
 ) {
     /**
      * True when this failure came from the transport (never connected,
@@ -102,6 +109,7 @@ suspend fun INostrClient.publishAndCollectResults(
     timeoutInSeconds: Long = 15,
 ): Map<NormalizedRelayUrl, PublishResult> {
     val resultChannel = Channel<DetailedResult>(UNLIMITED)
+    val mark = TimeSource.Monotonic.markNow()
 
     Log.d("publishAndConfirm") { "Waiting for ${relayList.size} responses" }
 
@@ -133,8 +141,13 @@ suspend fun INostrClient.publishAndCollectResults(
 
                 when (msg) {
                     is OkMessage -> {
-                        if (msg.eventId == event.id) {
-                            resultChannel.trySend(DetailedResult(relay.url, msg.success, msg.message))
+                        // The relayList guard matters, not just the id: the same event may
+                        // have been published to OTHER relays by an earlier call (probe
+                        // waves, republish), and counting their late OKs here would inflate
+                        // receivedResults and end the wait loop before every listed relay
+                        // answered — misreporting the missing ones as NO_RESPONSE.
+                        if (msg.eventId == event.id && relay.url in relayList) {
+                            resultChannel.trySend(DetailedResult(relay.url, msg.success, msg.message, mark.elapsedNow().inWholeMilliseconds))
                             Log.d("publishAndConfirm") { "onSendResponse Received response for ${msg.eventId} from relay ${relay.url} message ${msg.message} success ${msg.success}" }
                         }
                     }
@@ -160,7 +173,7 @@ suspend fun INostrClient.publishAndCollectResults(
                                     val currentResult = receivedResults[result.relay]
                                     // do not override a successful result.
                                     if (currentResult == null || !currentResult.accepted) {
-                                        receivedResults[result.relay] = PublishResult(result.success, result.message)
+                                        receivedResults[result.relay] = PublishResult(result.success, result.message, result.elapsedMs)
                                     }
                                 }
                             }
@@ -191,4 +204,5 @@ private class DetailedResult(
     val relay: NormalizedRelayUrl,
     val success: Boolean,
     val message: String,
+    val elapsedMs: Long = -1,
 )

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/normalizer/RelayUrlNormalizer.kt

@@ -84,17 +84,107 @@ class RelayUrlNormalizer {
 
         private fun norm(url: String) = NormalizedRelayUrl(Rfc3986.normalize(url))
 
+        private fun isInvisible(c: Char) = c == '\u200B' || c == '\u200C' || c == '\u200D' || c == '\u2060' || c == '\uFEFF'
+
+        /**
+         * Scans the authority (host[:port]) that starts at [start] and ends at the first
+         * `/`, `?` or `#`. Returns the end index, or -1 when the authority is empty or
+         * contains characters that never appear in a real relay host (`@` userinfo,
+         * percent-encoding, commas).
+         */
+        private fun authorityEnd(
+            url: String,
+            start: Int,
+        ): Int {
+            if (start >= url.length) return -1
+            var i = start
+            if (url[i] == '[') {
+                // IPv6 literal: defer validation to the RFC 3986 parser
+                while (i < url.length && url[i] != '/' && url[i] != '?' && url[i] != '#') i++
+                return i
+            }
+            while (i < url.length) {
+                val c = url[i]
+                if (c == '/' || c == '?' || c == '#') break
+                if (c == '@' || c == '%' || c == ',') return -1
+                i++
+            }
+            return if (i == start) -1 else i
+        }
+
+        /**
+         * Accepts a ws/wss url whose host starts at [hostStart] if the authority is sane
+         * and the path does not start with `//` (the signature of a second URL or a broken
+         * `https//` pasted after the scheme, e.g. `wss://https//nostr.watch/relay/x`).
+         */
+        private fun fixWs(
+            url: String,
+            hostStart: Int,
+        ): String? {
+            val end = authorityEnd(url, hostStart)
+            if (end < 0) return null
+            if (end + 1 < url.length && url[end] == '/' && url[end + 1] == '/') return null
+            return url
+        }
+
+        /**
+         * Converts an http(s) url to ws(s) only when it is a bare host — nothing after
+         * `host[:port]` but an optional trailing `/`. An http url with a path, query or
+         * fragment (Mastodon actor urls from bridge `proxy` tags, web pages, images) is
+         * a web resource, not a relay: converting it creates a wss:// url that can never
+         * answer and only wastes connection attempts.
+         */
+        private fun fixHttp(
+            url: String,
+            hostStart: Int,
+            newScheme: String,
+        ): String? {
+            val end = authorityEnd(url, hostStart)
+            if (end < 0) return null
+            val bareHost = end == url.length || (end == url.length - 1 && url[end] == '/')
+            if (!bareHost) return null
+            return "$newScheme${url.substring(hostStart)}"
+        }
+
+        /**
+         * Validates a schemeless candidate: the part before the first `/` must look like
+         * `host` or `host:port` — letters, digits, `.`, `-`, `_`, plus at most one `:`
+         * followed by digits only. Rejects addressable-event pointers (`31990:hex:dtag`),
+         * bare scheme leftovers (`wss:`) and anything else that would otherwise be blindly
+         * prefixed with `wss://`.
+         */
+        private fun isBareHostAndPath(url: String): Boolean {
+            if (url[0] == '[') return true // IPv6 literal: defer to the RFC 3986 parser
+            var i = 0
+            var portStart = -1
+            while (i < url.length) {
+                val c = url[i]
+                if (c == '/') break
+                if (c == ':') {
+                    if (portStart >= 0) return false
+                    portStart = i + 1
+                } else if (portStart >= 0) {
+                    if (c < '0' || c > '9') return false
+                } else if (!c.isLetterOrDigit() && c != '.' && c != '-' && c != '_') {
+                    return false
+                }
+                i++
+            }
+            if (i == 0) return false
+            if (portStart >= 0 && portStart == i) return false
+            return true
+        }
+
         @OptIn(ExperimentalContracts::class)
         fun fix(rawUrl: String): String? {
             if (rawUrl.length < 4) return null
             if (rawUrl.contains("%00")) return null
 
-            // Trim trailing %20 (percent-encoded spaces from malformed event data)
-            val url =
-                rawUrl.trimEnd('%', '2', '0').let { trimmed ->
-                    // Only accept if we actually removed a trailing %20 pattern
-                    if (trimmed.length < rawUrl.length && rawUrl.endsWith("%20")) trimmed else rawUrl
-                }
+            // Trim trailing %20 (percent-encoded spaces from malformed event data).
+            // The endsWith gate keeps the hot path allocation-free: trimEnd would
+            // copy the string for ANY url merely ending in '%', '2' or '0' — which
+            // includes every port ending in zero ("wss://host:3030").
+            val url = if (rawUrl.endsWith("%20")) rawUrl.trimEnd('%', '2', '0') else rawUrl
             if (url.length < 4) return null
 
             // Reject URLs with %20 in the middle — these are garbage
@@ -109,17 +199,33 @@ class RelayUrlNormalizer {
                 }
             }
 
-            val trimmed =
+            var trimmed =
                 if (url[0].isWhitespace() || url[url.length - 1].isWhitespace()) {
                     url.trim()
                 } else {
                     url
                 }
 
+            // Single pass: interior whitespace means multiple urls or prose in one field,
+            // backslashes never appear in a real relay url; both are garbage. Invisible
+            // characters (zero-width spaces, BOM) are copy-paste artifacts — strip them.
+            var hasInvisible = false
+            for (c in trimmed) {
+                if (c == '\\') return null
+                if (c.isWhitespace()) return null
+                if (isInvisible(c)) hasInvisible = true
+            }
+            if (hasInvisible) {
+                trimmed = buildString(trimmed.length) { for (c in trimmed) if (!isInvisible(c)) append(c) }
+                if (trimmed.length < 4) return null
+            }
+
             // fast for good wss:// urls
             if (isRelaySchemePrefix(trimmed)) {
-                if (isRelaySchemePrefixSecure(trimmed) || isRelaySchemePrefixInsecure(trimmed)) {
-                    return trimmed
+                if (isRelaySchemePrefixSecure(trimmed)) {
+                    return fixWs(trimmed, 6)
+                } else if (isRelaySchemePrefixInsecure(trimmed)) {
+                    return fixWs(trimmed, 5)
                 }
             }
 
@@ -127,31 +233,31 @@ class RelayUrlNormalizer {
             if (isHttpPrefix(trimmed)) {
                 if (isHttpSSuffix(trimmed)) {
                     // https://
-                    return "wss://${trimmed.drop(8)}"
+                    return fixHttp(trimmed, 8, "wss://")
                 } else if (isHttpSuffix(trimmed)) {
                     // http://
-                    return "ws://${trimmed.drop(7)}"
+                    return fixHttp(trimmed, 7, "ws://")
                 }
             }
 
             // fast for good ww:// urls
             if (trimmed.startsWith("ww://")) {
-                return "wss://${trimmed.drop(5)}"
+                return fixWs("wss://${trimmed.drop(5)}", 6)
             }
 
             // fast for good ww:// urls
             if (trimmed.startsWith("was://")) {
-                return "wss://${trimmed.drop(6)}"
+                return fixWs("wss://${trimmed.drop(6)}", 6)
             }
 
             // fast for good ww:// urls
             if (trimmed.startsWith("Wws://")) {
-                return "wss://${trimmed.drop(6)}"
+                return fixWs("wss://${trimmed.drop(6)}", 6)
             }
 
             // fast for good ww:// urls
             if (trimmed.startsWith("Wss://")) {
-                return "wss://${trimmed.drop(6)}"
+                return fixWs("wss://${trimmed.drop(6)}", 6)
             }
 
             if (trimmed.contains("://")) {
@@ -160,10 +266,19 @@ class RelayUrlNormalizer {
                 return null
             }
 
-            return if (isOnion(trimmed) || isLocalHost(trimmed)) {
-                "ws://$trimmed"
+            // protocol-relative urls (`//host/`) are just missing the scheme
+            val bare = if (trimmed.startsWith("//")) trimmed.drop(2) else trimmed
+            if (bare.length < 4) return null
+
+            if (!isBareHostAndPath(bare)) {
+                Log.d("RelayUrlNormalizer") { "Rejected $url" }
+                return null
+            }
+
+            return if (isOnion(bare) || isLocalHost(bare)) {
+                "ws://$bare"
             } else {
-                "wss://$trimmed"
+                "wss://$bare"
             }
         }
 
@@ -186,6 +301,13 @@ class RelayUrlNormalizer {
                 val fixed = fix(url)
                 if (fixed != null) {
                     val normalized = norm(fixed)
+                    // the RFC 3986 parser can drop or replace the scheme on odd inputs;
+                    // anything that is not ws(s):// at this point cannot be connected to.
+                    if (!isRelayUrl(normalized.url)) {
+                        Log.d("NormalizedRelayUrl") { "Rejected $url" }
+                        normalizedUrls.put(url, NormalizationResult.Error)
+                        return null
+                    }
                     normalizedUrls.put(url, NormalizationResult.Success(normalized))
                     normalized
                 } else {

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip11RelayInfo/CachedNip11Fetcher.kt

@@ -0,0 +1,97 @@
+/*
+ * 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.nip11RelayInfo
+
+import androidx.collection.LruCache
+import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
+import com.vitorpamplona.quartz.utils.TimeUtils
+import kotlinx.coroutines.CancellationException
+
+/**
+ * TTL cache around any [Nip11Fetcher]. NIP-11 documents change rarely, so a
+ * successful fetch is served from memory for [ttlSeconds]; a FAILED fetch is
+ * also remembered — for the shorter [errorTtlSeconds] — so a mass census does
+ * not hammer a host that just refused, while still retrying it soon.
+ *
+ * Concurrent first fetches of the same relay are not deduplicated: both hit the
+ * network and the second result wins the cache slot. That is harmless (the
+ * document is idempotent) and keeps this class lock-free.
+ */
+class CachedNip11Fetcher(
+    private val delegate: Nip11Fetcher,
+    private val ttlSeconds: Long = DEFAULT_TTL_SECONDS,
+    private val errorTtlSeconds: Long = DEFAULT_ERROR_TTL_SECONDS,
+    maxEntries: Int = 1000,
+    private val now: () -> Long = { TimeUtils.now() },
+) : Nip11Fetcher {
+    private sealed interface Cached {
+        val at: Long
+    }
+
+    private class Hit(
+        val info: Nip11RelayInformation,
+        override val at: Long,
+    ) : Cached
+
+    private class Miss(
+        val message: String?,
+        override val at: Long,
+    ) : Cached
+
+    private val cache = LruCache<NormalizedRelayUrl, Cached>(maxEntries)
+
+    /** The cached document if present and fresh; null otherwise. Never touches the network. */
+    fun cachedOrNull(relay: NormalizedRelayUrl): Nip11RelayInformation? {
+        val hit = cache[relay] as? Hit ?: return null
+        return if (now() - hit.at < ttlSeconds) hit.info else null
+    }
+
+    /** Drops the cache entry (success or failure) so the next [fetch] is fresh. */
+    fun invalidate(relay: NormalizedRelayUrl) {
+        cache.remove(relay)
+    }
+
+    override suspend fun fetch(relay: NormalizedRelayUrl): Nip11RelayInformation {
+        when (val cached = cache[relay]) {
+            is Hit -> if (now() - cached.at < ttlSeconds) return cached.info
+            is Miss ->
+                if (now() - cached.at < errorTtlSeconds) {
+                    throw Nip11FetchException(cached.message ?: "cached NIP-11 failure for ${relay.url}")
+                }
+            null -> {}
+        }
+        return try {
+            delegate.fetch(relay).also { cache.put(relay, Hit(it, now())) }
+        } catch (e: Exception) {
+            if (e is CancellationException) throw e
+            cache.put(relay, Miss(e.message, now()))
+            throw e
+        }
+    }
+
+    companion object {
+        /** Documents are near-static: trust a success for a day. */
+        const val DEFAULT_TTL_SECONDS = 24L * 60 * 60
+
+        /** Failures are often transient: retry after five minutes. */
+        const val DEFAULT_ERROR_TTL_SECONDS = 5L * 60
+    }
+}

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip11RelayInfo/Nip11Fetcher.kt

@@ -0,0 +1,44 @@
+/*
+ * 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.nip11RelayInfo
+
+import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
+
+/**
+ * Fetches and parses a relay's NIP-11 information document (the
+ * `application/nostr+json` answer on the relay's https url). Mirrors the
+ * [com.vitorpamplona.quartz.nip05DnsIdentifiers.Nip05Fetcher] seam: the HTTP
+ * transport lives in a platform implementation (OkHttpNip11Fetcher on
+ * JVM/Android), so common code — probes, monitors, the CLI — depends only on
+ * this interface. Wrap any implementation in [CachedNip11Fetcher] to add a TTL
+ * cache.
+ *
+ * Throws [Nip11FetchException] (or a transport exception) when the document is
+ * unavailable or unparseable.
+ */
+interface Nip11Fetcher {
+    suspend fun fetch(relay: NormalizedRelayUrl): Nip11RelayInformation
+}
+
+/** The relay answered, but not with a usable NIP-11 document (bad status, not JSON). */
+class Nip11FetchException(
+    message: String,
+) : Exception(message)

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

@@ -0,0 +1,59 @@
+/*
+ * 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.nip66RelayMonitor.reachability
+
+import com.vitorpamplona.quartz.nip01Core.core.Event
+import com.vitorpamplona.quartz.nip01Core.signers.EventTemplate
+import com.vitorpamplona.quartz.nip01Core.signers.eventTemplate
+import com.vitorpamplona.quartz.nip40Expiration.ExpirationTag
+import com.vitorpamplona.quartz.utils.TimeUtils
+
+/**
+ * The event a NIP-66 monitor publishes to measure a relay's WRITE path: publish
+ * one of these (signed with the monitor key), time the `OK`, and read the
+ * rejection prefix when refused (`auth-required:`/`restricted:`/`pow:` map to
+ * the discovery record's `R` requirement tags; a timed acceptance is `rtt-write`).
+ *
+ * The kind is EPHEMERAL (20000–29999 per NIP-01), so a compliant relay serves it
+ * to current subscribers and never stores it — the probe leaves nothing behind.
+ * [KIND] 20166 is this library's convention (30166 discovery minus the
+ * addressable range), not something NIP-66 standardizes; any ephemeral kind
+ * works. Belt-and-braces, the template also carries a NIP-40 `expiration` tag
+ * [EXPIRATION_SECONDS] out, so a relay that stores unknown ephemeral kinds
+ * anyway purges it promptly.
+ *
+ * A rejection is still a MEASUREMENT: an `OK false` proves the write path works
+ * and documents the relay's policy. Only silence is a failed write test.
+ */
+object RelayProbeWriteTest {
+    const val KIND = 20166
+
+    /** Storage-window ceiling for non-compliant relays that store ephemeral events. */
+    const val EXPIRATION_SECONDS = 60L
+
+    fun build(
+        content: String = "NIP-66 write probe",
+        createdAt: Long = TimeUtils.now(),
+    ): EventTemplate<Event> =
+        eventTemplate(KIND, content, createdAt) {
+            add(ExpirationTag.assemble(createdAt + EXPIRATION_SECONDS))
+        }
+}

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

@@ -22,6 +22,8 @@ package com.vitorpamplona.quartz.nip66RelayMonitor.reachability
 
 import com.vitorpamplona.quartz.nip01Core.core.Event
 import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
+import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.PublishResult
+import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.publishAndCollectResults
 import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener
 import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener
 import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
@@ -29,10 +31,20 @@ import com.vitorpamplona.quartz.nip01Core.relay.client.single.newSubId
 import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
 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.NostrSigner
 import com.vitorpamplona.quartz.nip01Core.store.IEventStore
 import com.vitorpamplona.quartz.nip65RelayList.AdvertisedRelayListEvent
+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.RttType
+import com.vitorpamplona.quartz.utils.TimeUtils
 import com.vitorpamplona.quartz.utils.concurrent.ConcurrentMap
 import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.withTimeoutOrNull
 import kotlin.time.TimeSource
 
@@ -74,6 +86,19 @@ class RelayProber(
         val error: String?,
     )
 
+    /**
+     * One relay's read+write check outcome (see [readWriteCheck]). Latencies are
+     * -1 when unobserved; [writeAccepted] is null when the relay never answered
+     * the write with an OK (transport failure or silence).
+     */
+    class ReadWriteVerdict(
+        val relay: NormalizedRelayUrl,
+        val rttReadMs: Long,
+        val rttWriteMs: Long,
+        val writeAccepted: Boolean?,
+        val writeMessage: String?,
+    )
+
     class Result(
         val verdicts: List<Verdict>,
         val elapsedMs: Long,
@@ -91,18 +116,24 @@ class RelayProber(
     /**
      * Probe every relay in [relays], [waveSize] at a time, giving each wave up to
      * [timeoutMs] to reach terminals. Returns one [Verdict] per input relay.
+     *
+     * [filters] is the REQ each relay is asked to answer. The default
+     * [LIVENESS_FILTERS] matches nothing, so an EOSE proves liveness without
+     * streaming a payload; pass [readTestFilter] to make [Verdict.rttEoseMs] a
+     * real read test instead (the relay must query and stream an actual event).
      */
     suspend fun probe(
         relays: Collection<NormalizedRelayUrl>,
         timeoutMs: Long = 15_000,
         waveSize: Int = 1000,
+        filters: List<Filter> = LIVENESS_FILTERS,
     ): Result {
         val mark = TimeSource.Monotonic.markNow()
         val all = ArrayList<Verdict>(relays.size)
         val distinct = relays.toSet()
         var done = 0
         for (wave in distinct.chunked(waveSize.coerceAtLeast(1))) {
-            all += probeWave(wave, timeoutMs)
+            probeWave(wave, timeoutMs, filters) { all += it }
             done += wave.size
             if (distinct.size > wave.size) {
                 val liveSoFar = all.count { it.reachable }
@@ -112,10 +143,94 @@ class RelayProber(
         return Result(all, mark.elapsedNow().inWholeMilliseconds)
     }
 
+    /**
+     * Streaming variant of [probe]: a cold [Flow] that emits each relay's [Verdict]
+     * the moment that relay resolves — an answering relay's verdict arrives as soon
+     * as its EOSE/CLOSED/connect-failure lands, not when the whole census ends. Only
+     * relays that stay silent wait for their wave's [timeoutMs] deadline.
+     *
+     * [filters] picks the check, as in [probe]: [LIVENESS_FILTERS] (default) or
+     * [readTestFilter].
+     *
+     * Probing starts when the flow is collected, and emission is sequential — a slow
+     * collector delays the next wave AND eats into the current wave's [timeoutMs]
+     * window (the deadline is absolute; answers keep being recorded while the
+     * collector runs, but silent relays get less listening time). Keep per-verdict
+     * work light, or buffer, when precise deadlines matter. Pair each verdict with
+     * [toDiscoveryEventTemplate] to turn the stream into signable NIP-66 kind:30166
+     * records for another process to sign and publish.
+     */
+    fun probeFlow(
+        relays: Collection<NormalizedRelayUrl>,
+        timeoutMs: Long = 15_000,
+        waveSize: Int = 1000,
+        filters: List<Filter> = LIVENESS_FILTERS,
+    ): Flow<Verdict> =
+        flow {
+            for (wave in relays.toSet().chunked(waveSize.coerceAtLeast(1))) {
+                probeWave(wave, timeoutMs, filters) { emit(it) }
+            }
+        }
+
+    /**
+     * The deeper, still-honest check pair: READ (a real limit-[readLimit] REQ the
+     * relay must query its store for) and WRITE (one ephemeral [RelayProbeWriteTest]
+     * event signed by [signer], the monitor key, timed to its OK). Everything is a
+     * direct observation — nothing is copied from the relay's NIP-11 self-claims.
+     *
+     * Run it against relays ALREADY PROVEN LIVE — typically [Result.reachable] of a
+     * [probe] that just ran, while the pool's sockets are still open. On a warm
+     * socket both numbers are honest NIP-66 rtts (`rtt-read`, `rtt-write`); against
+     * a cold relay they silently include the dial, so don't.
+     *
+     * A write REJECTION is still a measurement: `OK false` proves the write path
+     * works and documents policy ([ReadWriteVerdict.writeMessage] keeps the NIP-01
+     * machine-readable reason; [toDiscoveryEventTemplate] maps `auth-required:` and
+     * `pow:` to `R` tags). Only silence leaves [ReadWriteVerdict.writeAccepted] null.
+     */
+    suspend fun readWriteCheck(
+        relays: Collection<NormalizedRelayUrl>,
+        signer: NostrSigner,
+        timeoutMs: Long = 15_000,
+        waveSize: Int = 1000,
+        readLimit: Int = 1,
+        readKinds: List<Int>? = listOf(0),
+    ): Map<NormalizedRelayUrl, ReadWriteVerdict> {
+        val out = HashMap<NormalizedRelayUrl, ReadWriteVerdict>()
+        val distinct = relays.toSet()
+        for ((waveIndex, wave) in distinct.chunked(waveSize.coerceAtLeast(1)).withIndex()) {
+            val reads = HashMap<NormalizedRelayUrl, Long>()
+            probeWave(wave, timeoutMs, readTestFilter(readLimit, readKinds)) { reads[it.relay] = it.rttEoseMs }
+
+            // A distinct event id per wave (createdAt has second granularity, so the
+            // content must vary) keeps a straggler OK from an earlier wave's relays
+            // from ever matching this wave's confirmation window.
+            val event = signer.sign(RelayProbeWriteTest.build(content = "NIP-66 write probe $waveIndex"))
+            val writes = client.publishAndCollectResults(event, wave.toSet(), (timeoutMs / 1000).coerceAtLeast(1))
+
+            for (relay in wave) {
+                // Only a real OK (true or false) counts as an answer; transport
+                // failures and silence leave the write side unobserved.
+                val answered = writes[relay]?.takeUnless { it.isTransportFailure || it.message == PublishResult.NO_RESPONSE }
+                out[relay] =
+                    ReadWriteVerdict(
+                        relay = relay,
+                        rttReadMs = reads[relay] ?: -1,
+                        rttWriteMs = answered?.elapsedMs ?: -1,
+                        writeAccepted = answered?.accepted,
+                        writeMessage = answered?.message,
+                    )
+            }
+        }
+        return out
+    }
+
     private suspend fun probeWave(
         wave: List<NormalizedRelayUrl>,
         timeoutMs: Long,
-    ): List<Verdict> {
+        filters: List<Filter>,
+        onVerdict: suspend (Verdict) -> Unit,
+    ) {
         val mark = TimeSource.Monotonic.markNow()
         val waveSet = wave.toHashSet()
         val openRtt = ConcurrentMap<NormalizedRelayUrl, Long>()
@@ -178,22 +293,7 @@ class RelayProber(
                 }
             }
 
-        client.addConnectionListener(connListener)
-        try {
-            client.subscribe(subId, wave.associateWith { PROBE_FILTERS }, subListener)
-            val remaining = wave.toMutableSet()
-            withTimeoutOrNull(timeoutMs) {
-                while (remaining.isNotEmpty()) {
-                    remaining.remove(terminals.receive())
-                }
-            }
-        } finally {
-            client.unsubscribe(subId)
-            client.removeConnectionListener(connListener)
-            terminals.close()
-        }
-
-        return wave.map { relay ->
+        fun verdictOf(relay: NormalizedRelayUrl): Verdict {
             val opened = openRtt[relay]
             val answered = eoseMs[relay]
             val error = errors[relay]
@@ -202,7 +302,7 @@ class RelayProber(
             // Only a connect failure, or silence with no socket, is dead.
             val cannot = error?.startsWith("cannot:") == true
             val reachable = !cannot && (opened != null || answered != null || error != null)
-            Verdict(
+            return Verdict(
                 relay = relay,
                 reachable = reachable,
                 rttOpenMs = opened ?: -1,
@@ -210,13 +310,54 @@ class RelayProber(
                 error = error,
             )
         }
+
+        client.addConnectionListener(connListener)
+        try {
+            client.subscribe(subId, wave.associateWith { filters }, subListener)
+            val remaining = wave.toMutableSet()
+            while (remaining.isNotEmpty()) {
+                val left = timeoutMs - mark.elapsedNow().inWholeMilliseconds
+                if (left <= 0) break
+                val relay = withTimeoutOrNull(left) { terminals.receive() } ?: break
+                // The emission happens OUTSIDE the timeout window so a collector that
+                // suspends on a verdict can never be cancelled mid-emission and lose it.
+                if (remaining.remove(relay)) onVerdict(verdictOf(relay))
+            }
+            // Whatever is left resolved nothing by the deadline: dead if the socket
+            // never opened, reachable-but-slow if it did.
+            for (relay in remaining) onVerdict(verdictOf(relay))
+        } finally {
+            client.unsubscribe(subId)
+            client.removeConnectionListener(connListener)
+            terminals.close()
+        }
     }
 
     companion object {
-        // A filter no event can match (ids are 64-hex of a hash): the relay answers
-        // with an immediate EOSE and never streams a payload. Same trick as the
-        // crawler's warm pool.
-        private val PROBE_FILTERS = listOf(Filter(ids = listOf("0".repeat(64))))
+        /**
+         * A filter no event can match (ids are 64-hex of a hash): the relay answers
+         * with an immediate EOSE and never streams a payload. Same trick as the
+         * crawler's warm pool. This is the default check — pure liveness.
+         */
+        val LIVENESS_FILTERS = listOf(Filter(ids = listOf("0".repeat(64))))
+
+        /**
+         * A REQ the relay must actually WORK for: query its store and stream up to
+         * [limit] real events before the EOSE. Pass to [probe]/[probeFlow] as
+         * [filters] to turn [Verdict.rttEoseMs] into a genuine read test rather
+         * than a liveness ping — the time still counts from the wave start (dial
+         * included), so compare it against [Verdict.rttOpenMs], not across waves.
+         *
+         * [kinds] defaults to kind 0: purpose relays (purplepag.es) reject any REQ
+         * that names no kind with `blocked: filters must specify at least one kind`,
+         * and practically every relay stores SOME profile — so a kind-0, limit-1
+         * query works everywhere. Pass a different list to probe a specific shelf,
+         * or null for a kind-less query on relays known to allow one.
+         */
+        fun readTestFilter(
+            limit: Int = 1,
+            kinds: List<Int>? = listOf(0),
+        ) = listOf(Filter(kinds = kinds, limit = limit))
 
         /**
          * The relay universe the local store knows: every read/write relay advertised
@@ -263,3 +404,43 @@ class RelayProber(
         }
     }
 }
+
+/**
+ * This verdict as an UNSIGNED NIP-66 kind:30166 Relay Discovery template — the d-tag
+ * is the normalized relay url; sign it with the consumer's own monitor key (per
+ * NIP-66 a monitor is its own identity, so the prober never signs on its own).
+ *
+ * Only facts a probe actually observed are tagged:
+ *  - `n` network type inferred from the url (clearnet/tor/i2p);
+ *  - `rtt-open` when the relay was reachable — the measured WS-upgrade round trip,
+ *    or 0 for "reachable, latency not observed" (liveness is the tag's PRESENCE);
+ *  - `rtt-read`/`rtt-write` when a [RelayProber.readWriteCheck] result is passed
+ *    as [readWrite] and actually measured that side;
+ *  - `R auth` when the relay answered a probe with a NIP-42 `auth-required`
+ *    (CLOSED on the REQ, or OK-false on the write) and `R pow` when the write
+ *    was refused with a `pow:` reason — observed walls, not NIP-11 claims.
+ *
+ * NIP-11-derived tags (`N` supported NIPs, `k` kinds, `T` type) are deliberately
+ * absent: those are the relay's self-claims, and asserting them under a monitor
+ * signature without a per-NIP compliance test would launder claims into
+ * measurements. [Verdict.rttEoseMs] is likewise never written as `rtt-read` — it
+ * is wave-relative (dial + TLS + queueing), so the honest read number only comes
+ * from [readWrite] (or the [RelayObserver]/[RelayMonitor] real-traffic path).
+ */
+fun RelayProber.Verdict.toDiscoveryEventTemplate(
+    createdAt: Long = TimeUtils.now(),
+    readWrite: RelayProber.ReadWriteVerdict? = null,
+): EventTemplate<RelayDiscoveryEvent> =
+    RelayDiscoveryEvent.build(relay, createdAt = createdAt) {
+        networkType(RelayReachabilityStore.networkTypeOf(relay))
+        if (reachable) rtt(RttType.OPEN, rttOpenMs.coerceAtLeast(0))
+        if (readWrite != null) {
+            if (readWrite.rttReadMs >= 0) rtt(RttType.READ, readWrite.rttReadMs)
+            if (readWrite.rttWriteMs >= 0) rtt(RttType.WRITE, readWrite.rttWriteMs)
+        }
+        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")
+    }

quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/RelayUrlFormatterTest.kt

@@ -52,4 +52,93 @@ class RelayUrlFormatterTest {
     fun weirdRelay() {
         assertNull(RelayUrlNormalizer.normalizeOrNull("wss://relay%20list%20to%20discover%20the%20user's%20content"))
     }
+
+    @Test
+    fun trailingPercentTwentyIsTrimmedButBareTrailingZeroIsNot() {
+        assertEquals("wss://nostr.mom/", RelayUrlNormalizer.normalizeOrNull("wss://nostr.mom%20")?.url)
+        // urls merely ending in '%', '2' or '0' (every port ending in zero) must pass untouched
+        assertEquals("wss://nostr.mom:3030/", RelayUrlNormalizer.normalizeOrNull("wss://nostr.mom:3030")?.url)
+        assertEquals("wss://nostr.mom:8020/", RelayUrlNormalizer.normalizeOrNull("wss://nostr.mom:8020")?.url)
+    }
+
+    @Test
+    fun httpWithPathIsNotARelay() {
+        // Mastodon/bridge actor urls from `proxy` tags: web resources, not relays
+        assertNull(RelayUrlNormalizer.normalizeOrNull("https://mastodon.social/users/amanita_muscaria"))
+        assertNull(RelayUrlNormalizer.normalizeOrNull("https://fosstodon.org/ap/users/115532410310000993"))
+        assertNull(RelayUrlNormalizer.normalizeOrNull("http://example.com/relay"))
+        assertNull(RelayUrlNormalizer.normalizeOrNull("https://nostr.mom/?author=0"))
+        assertNull(RelayUrlNormalizer.normalizeOrNull("https://nostr.mom/#section"))
+
+        // but bare hosts still convert, with or without port and trailing slash
+        assertEquals("wss://nostr.mom/", RelayUrlNormalizer.normalizeOrNull("https://nostr.mom")?.url)
+        assertEquals("wss://nostr.mom/", RelayUrlNormalizer.normalizeOrNull("https://nostr.mom/")?.url)
+        assertEquals("wss://nostr.mom:4443/", RelayUrlNormalizer.normalizeOrNull("https://nostr.mom:4443/")?.url)
+        assertEquals("ws://nostr.mom/", RelayUrlNormalizer.normalizeOrNull("http://nostr.mom")?.url)
+    }
+
+    @Test
+    fun wsWithPathIsStillARelay() {
+        assertEquals("wss://relay.nostr.band/all", RelayUrlNormalizer.normalizeOrNull("wss://relay.nostr.band/all")?.url)
+        assertEquals(
+            "wss://bostr.lecturify.net/?accept=0,1",
+            RelayUrlNormalizer.normalizeOrNull("wss://bostr.lecturify.net/?accept=0,1")?.url,
+        )
+    }
+
+    @Test
+    fun brokenSchemeGarbage() {
+        assertNull(RelayUrlNormalizer.normalizeOrNull("wss:"))
+        assertNull(RelayUrlNormalizer.normalizeOrNull("wss://https//nostr.watch/relay/nostr.21crypto.ch"))
+        assertNull(RelayUrlNormalizer.normalizeOrNull("wss://https://lockbox.fiatjaf.com"))
+        assertNull(RelayUrlNormalizer.normalizeOrNull("ws://http//nos.lol"))
+        assertNull(RelayUrlNormalizer.normalizeOrNull("wss://://plebstr.com"))
+    }
+
+    @Test
+    fun nostrUriIsNotARelay() {
+        assertNull(RelayUrlNormalizer.normalizeOrNull("nostr://nrelay1qqxhwumn8ghj77tpvf6jumt9e2ckgn/"))
+        assertNull(RelayUrlNormalizer.normalizeOrNull("nostr://npub1dwy079xmpz7mk02kvz6wan49h02635umk32aa4ufek8t8mjxv58qy2nr22/"))
+        assertNull(RelayUrlNormalizer.normalizeOrNull("nostr:nrelay1qq8k2cnfwejhyum99eek7cmfv9kqsm7sdm"))
+    }
+
+    @Test
+    fun addressablePointerIsNotARelay() {
+        assertNull(RelayUrlNormalizer.normalizeOrNull("31990:6be38f8c63df7dbf84db7ec4a6e6fbbd8d19dca3b980efad18585c46f04b26f9:mostr"))
+    }
+
+    @Test
+    fun authorityGarbage() {
+        // userinfo, percent-encoding and commas never appear in a real relay host
+        assertNull(RelayUrlNormalizer.normalizeOrNull("wss://catuaba@plebs.place/"))
+        assertNull(RelayUrlNormalizer.normalizeOrNull("wss://africa.nostr.joburg%0A/"))
+        assertNull(RelayUrlNormalizer.normalizeOrNull("wss://bitcoiner,social/"))
+        assertNull(RelayUrlNormalizer.normalizeOrNull("wss://#web3/"))
+        assertNull(RelayUrlNormalizer.normalizeOrNull("name@domain.com"))
+    }
+
+    @Test
+    fun interiorWhitespaceAndBackslashes() {
+        assertNull(RelayUrlNormalizer.normalizeOrNull("wss://nos lol"))
+        assertNull(RelayUrlNormalizer.normalizeOrNull("wss://nos.lol/ wss:/nostr.land/  avatar wss:/nostr.wine/"))
+        assertNull(RelayUrlNormalizer.normalizeOrNull("wss://\\\\relay.damus.io/"))
+    }
+
+    @Test
+    fun invisibleCharactersAreStripped() {
+        assertEquals("wss://nos.lol/", RelayUrlNormalizer.normalizeOrNull("wss://\u200Bnos.lol")?.url)
+        assertEquals("wss://nos.lol/", RelayUrlNormalizer.normalizeOrNull("\uFEFFwss://nos.lol")?.url)
+    }
+
+    @Test
+    fun protocolRelativeUrls() {
+        assertEquals("wss://relay.most.pub/", RelayUrlNormalizer.normalizeOrNull("//relay.most.pub/")?.url)
+        assertEquals("wss://nos.lol/", RelayUrlNormalizer.normalizeOrNull("//nos.lol/")?.url)
+    }
+
+    @Test
+    fun ipv6AndLanHostsStillWork() {
+        assertEquals("ws://[31b:6f20:c7f2:3ddf::3221]/", RelayUrlNormalizer.normalizeOrNull("ws://[31b:6f20:c7f2:3ddf::3221]/")?.url)
+        assertEquals("ws://geyser-relay:7777/", RelayUrlNormalizer.normalizeOrNull("ws://geyser-relay:7777/")?.url)
+    }
 }

quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip11RelayInfo/CachedNip11FetcherTest.kt

@@ -0,0 +1,125 @@
+/*
+ * 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.nip11RelayInfo
+
+import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
+import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer
+import kotlinx.coroutines.runBlocking
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertFailsWith
+import kotlin.test.assertNull
+
+class CachedNip11FetcherTest {
+    private val relay = RelayUrlNormalizer.normalize("wss://nostr.example.com")
+
+    /** Counts network hits; serves [doc] or throws when [failing]. */
+    private class FakeFetcher : Nip11Fetcher {
+        var calls = 0
+        var failing = false
+        var doc = Nip11RelayInformation(name = "v1")
+
+        override suspend fun fetch(relay: NormalizedRelayUrl): Nip11RelayInformation {
+            calls++
+            if (failing) throw Nip11FetchException("boom")
+            return doc
+        }
+    }
+
+    private fun cached(
+        delegate: FakeFetcher,
+        clock: () -> Long,
+    ) = CachedNip11Fetcher(delegate, ttlSeconds = 100, errorTtlSeconds = 10, now = clock)
+
+    @Test
+    fun freshSuccessIsServedFromCache() =
+        runBlocking {
+            val net = FakeFetcher()
+            var now = 0L
+            val fetcher = cached(net) { now }
+
+            assertEquals("v1", fetcher.fetch(relay).name)
+            now = 99
+            assertEquals("v1", fetcher.fetch(relay).name)
+            assertEquals(1, net.calls, "second fetch inside the TTL must not touch the network")
+        }
+
+    @Test
+    fun successExpiresAfterTtl() =
+        runBlocking {
+            val net = FakeFetcher()
+            var now = 0L
+            val fetcher = cached(net) { now }
+
+            fetcher.fetch(relay)
+            net.doc = Nip11RelayInformation(name = "v2")
+            now = 100
+            assertEquals("v2", fetcher.fetch(relay).name)
+            assertEquals(2, net.calls)
+        }
+
+    @Test
+    fun failureIsCachedForItsOwnShorterTtl() =
+        runBlocking {
+            val net = FakeFetcher().apply { failing = true }
+            var now = 0L
+            val fetcher = cached(net) { now }
+
+            assertFailsWith<Nip11FetchException> { fetcher.fetch(relay) }
+            now = 9
+            assertFailsWith<Nip11FetchException> { fetcher.fetch(relay) }
+            assertEquals(1, net.calls, "a fresh failure must be served from cache, not re-fetched")
+
+            now = 10
+            net.failing = false
+            assertEquals("v1", fetcher.fetch(relay).name)
+            assertEquals(2, net.calls, "an expired failure must be retried")
+        }
+
+    @Test
+    fun invalidateForcesAFreshFetch() =
+        runBlocking {
+            val net = FakeFetcher()
+            val fetcher = cached(net) { 0 }
+
+            fetcher.fetch(relay)
+            fetcher.invalidate(relay)
+            fetcher.fetch(relay)
+            assertEquals(2, net.calls)
+        }
+
+    @Test
+    fun cachedOrNullNeverTouchesTheNetwork() =
+        runBlocking {
+            val net = FakeFetcher()
+            var now = 0L
+            val fetcher = cached(net) { now }
+
+            assertNull(fetcher.cachedOrNull(relay))
+            assertEquals(0, net.calls)
+
+            fetcher.fetch(relay)
+            assertEquals("v1", fetcher.cachedOrNull(relay)?.name)
+            now = 100
+            assertNull(fetcher.cachedOrNull(relay), "a stale hit must not be served")
+            assertEquals(1, net.calls)
+        }
+}

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

@@ -0,0 +1,465 @@
+/*
+ * 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.nip66RelayMonitor.reachability
+
+import com.vitorpamplona.quartz.nip01Core.core.Event
+import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
+import com.vitorpamplona.quartz.nip01Core.relay.client.EmptyNostrClient
+import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
+import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener
+import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener
+import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
+import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.OkMessage
+import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.Command
+import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
+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 kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.currentTime
+import kotlinx.coroutines.test.runTest
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertFalse
+import kotlin.test.assertNull
+import kotlin.test.assertTrue
+
+/**
+ * Pins [RelayProber.probeFlow]'s streaming contract: an answering relay's verdict
+ * is emitted the moment its terminal arrives, while silent relays only resolve at
+ * the wave deadline — and pins the observed-facts-only tag set of
+ * [toDiscoveryEventTemplate].
+ */
+@OptIn(ExperimentalCoroutinesApi::class)
+class RelayProberFlowTest {
+    /** Captures the probe subscription and publish so the test can play the relays. */
+    private class ScriptedClient : INostrClient by EmptyNostrClient() {
+        var listener: SubscriptionListener? = null
+        var sentFilters: Map<NormalizedRelayUrl, List<Filter>>? = null
+        var published: Event? = null
+        val connListeners = mutableListOf<RelayConnectionListener>()
+
+        override fun subscribe(
+            subId: String,
+            filters: Map<NormalizedRelayUrl, List<Filter>>,
+            listener: SubscriptionListener?,
+        ) {
+            this.listener = listener
+            this.sentFilters = filters
+        }
+
+        override fun publish(
+            event: Event,
+            relayList: Set<NormalizedRelayUrl>,
+        ) {
+            published = event
+        }
+
+        override fun addConnectionListener(listener: RelayConnectionListener) {
+            connListeners += listener
+        }
+
+        override fun removeConnectionListener(listener: RelayConnectionListener) {
+            connListeners -= listener
+        }
+
+        /** Plays a relay's OK answer for the published event to every armed listener. */
+        fun answerOk(
+            relay: NormalizedRelayUrl,
+            success: Boolean,
+            message: String,
+        ) {
+            val ok = OkMessage(published!!.id, success, message)
+            connListeners.toList().forEach { it.onIncomingMessage(FakeRelayClient(relay), "", ok) }
+        }
+    }
+
+    private class FakeRelayClient(
+        override val url: NormalizedRelayUrl,
+    ) : IRelayClient {
+        override fun connect() = Unit
+
+        override fun needsToReconnect() = false
+
+        override fun connectAndSyncFiltersIfDisconnected(ignoreRetryDelays: Boolean) = Unit
+
+        override fun isConnected() = true
+
+        override fun sendOrConnectAndSync(cmd: Command) = Unit
+
+        override fun sendIfConnected(cmd: Command) = Unit
+
+        override fun disconnect() = Unit
+    }
+
+    private val fast = RelayUrlNormalizer.normalize("wss://fast.example.com")
+    private val silent = RelayUrlNormalizer.normalize("wss://silent.example.com")
+    private val walled = RelayUrlNormalizer.normalize("wss://walled.example.com")
+
+    @Test
+    fun answeringRelayStreamsBeforeTheWaveDeadline() =
+        runTest {
+            val client = ScriptedClient()
+            val arrivals = mutableListOf<Pair<RelayProber.Verdict, Long>>()
+
+            val collector =
+                launch {
+                    RelayProber(client)
+                        .probeFlow(listOf(fast, silent), timeoutMs = 10_000)
+                        .collect { arrivals += it to currentTime }
+                }
+            launch {
+                delay(200)
+                client.listener!!.onEose(fast, null)
+            }
+            collector.join()
+
+            assertEquals(listOf(fast, silent), arrivals.map { it.first.relay })
+            // The EOSE'd relay resolved when it answered, not at the deadline …
+            val (fastVerdict, fastAt) = arrivals[0]
+            assertTrue(fastVerdict.reachable)
+            assertTrue(fastAt < 1_000, "verdict should stream at answer time, arrived at ${fastAt}ms")
+            // … while the silent one waited out the wave and is dead (socket never opened).
+            val (silentVerdict, silentAt) = arrivals[1]
+            assertFalse(silentVerdict.reachable)
+            assertTrue(silentAt >= 10_000, "silent relay must wait for the deadline, arrived at ${silentAt}ms")
+        }
+
+    @Test
+    fun authWalledRelayIsReachableWithTheWallRecorded() =
+        runTest {
+            val client = ScriptedClient()
+            val arrivals = mutableListOf<RelayProber.Verdict>()
+
+            val collector =
+                launch {
+                    RelayProber(client)
+                        .probeFlow(listOf(walled), timeoutMs = 10_000)
+                        .collect { arrivals += it }
+                }
+            launch {
+                delay(100)
+                client.listener!!.onClosed("auth-required: sign in first", walled, null)
+            }
+            collector.join()
+
+            assertEquals(1, arrivals.size)
+            assertTrue(arrivals[0].reachable, "an auth wall is an app-level answer: the relay works")
+            assertEquals("closed:auth-required: sign in first", arrivals[0].error)
+        }
+
+    @Test
+    fun connectFailureStreamsImmediatelyAsDead() =
+        runTest {
+            val client = ScriptedClient()
+            val arrivals = mutableListOf<Pair<RelayProber.Verdict, Long>>()
+
+            val collector =
+                launch {
+                    RelayProber(client)
+                        .probeFlow(listOf(fast), timeoutMs = 10_000)
+                        .collect { arrivals += it to currentTime }
+                }
+            launch {
+                delay(50)
+                client.listener!!.onCannotConnect(fast, "dns failure", null)
+            }
+            collector.join()
+
+            val (verdict, at) = arrivals.single()
+            assertFalse(verdict.reachable)
+            assertEquals("cannot:dns failure", verdict.error)
+            assertTrue(at < 1_000, "a failed dial must not wait for the deadline, arrived at ${at}ms")
+        }
+
+    // ------------------------------------------------------------------
+    // Check options — liveness default, read-test override, write-test event
+    // ------------------------------------------------------------------
+
+    @Test
+    fun livenessFilterIsTheDefaultCheck() =
+        runTest {
+            val client = ScriptedClient()
+            val collector =
+                launch {
+                    RelayProber(client).probeFlow(listOf(fast), timeoutMs = 1_000).collect {}
+                }
+            launch {
+                delay(10)
+                assertEquals(RelayProber.LIVENESS_FILTERS, client.sentFilters!![fast])
+                client.listener!!.onEose(fast, null)
+            }
+            collector.join()
+        }
+
+    @Test
+    fun readTestFilterIsSentWhenChosen() =
+        runTest {
+            val client = ScriptedClient()
+            val collector =
+                launch {
+                    RelayProber(client)
+                        .probeFlow(listOf(fast), timeoutMs = 1_000, filters = RelayProber.readTestFilter())
+                        .collect {}
+                }
+            launch {
+                delay(10)
+                val sent = client.sentFilters!![fast]!!.single()
+                assertEquals(1, sent.limit, "read test defaults to limit 1")
+                assertEquals(listOf(0), sent.kinds, "read test defaults to kind 0 — accepted by purpose relays too")
+                assertNull(sent.ids, "read test must query real events, not the impossible id")
+                client.listener!!.onEose(fast, null)
+            }
+            collector.join()
+        }
+
+    @Test
+    fun writeTestEventIsEphemeralAndSelfExpiring() {
+        val template = RelayProbeWriteTest.build(createdAt = 5000)
+
+        assertEquals(20166, template.kind)
+        assertTrue(template.kind in 20000..29999, "the write probe must be an ephemeral kind")
+        assertTrue(listOf("expiration", "5060") in template.tags.map { it.toList() })
+    }
+
+    // ------------------------------------------------------------------
+    // readWriteCheck — honest read + write measurements, nothing claimed
+    // ------------------------------------------------------------------
+
+    private suspend fun ScriptedClient.playReadThenWrite(
+        relay: NormalizedRelayUrl,
+        ok: Boolean?,
+        okMessage: String = "",
+    ) {
+        delay(50)
+        listener!!.onEose(relay, null) // read phase answers
+        while (published == null) delay(10) // write phase begins
+        if (ok != null) answerOk(relay, ok, okMessage)
+    }
+
+    @Test
+    fun readWriteCheckMeasuresBothSides() =
+        runTest {
+            val client = ScriptedClient()
+            val signer = NostrSignerInternal(KeyPair())
+            var result: Map<NormalizedRelayUrl, RelayProber.ReadWriteVerdict>? = null
+
+            val check =
+                launch {
+                    result = RelayProber(client).readWriteCheck(listOf(fast), signer, timeoutMs = 5_000)
+                }
+            launch { client.playReadThenWrite(fast, ok = true) }
+            check.join()
+
+            val verdict = result!![fast]!!
+            assertTrue(verdict.rttReadMs >= 0, "an answered read must be measured")
+            assertTrue(verdict.rttWriteMs >= 0, "an answered write must be measured")
+            assertEquals(true, verdict.writeAccepted)
+            assertEquals(20166, client.published!!.kind, "the write test must use the ephemeral probe event")
+        }
+
+    @Test
+    fun writeRejectionIsAnAnswerNotAFailure() =
+        runTest {
+            val client = ScriptedClient()
+            val signer = NostrSignerInternal(KeyPair())
+            var result: Map<NormalizedRelayUrl, RelayProber.ReadWriteVerdict>? = null
+
+            val check =
+                launch {
+                    result = RelayProber(client).readWriteCheck(listOf(walled), signer, timeoutMs = 5_000)
+                }
+            launch { client.playReadThenWrite(walled, ok = false, okMessage = "pow: 28 bits needed") }
+            check.join()
+
+            val verdict = result!![walled]!!
+            assertEquals(false, verdict.writeAccepted, "OK false is a measured policy answer")
+            assertEquals("pow: 28 bits needed", verdict.writeMessage)
+            assertTrue(verdict.rttWriteMs >= 0, "a rejection is still a round trip")
+        }
+
+    @Test
+    fun foreignRelayOkDoesNotEndTheWriteConfirmationEarly() =
+        runTest {
+            // A relay OUTSIDE the checked set answering with the same event id (a
+            // straggler from an earlier wave that got the same probe event) must not
+            // count toward the confirmation window — before the relayList guard in
+            // publishAndCollectResults, it ended the wait early and misreported the
+            // real relay as silent.
+            val client = ScriptedClient()
+            val signer = NostrSignerInternal(KeyPair())
+            val foreign = RelayUrlNormalizer.normalize("wss://foreign.example.com")
+            var result: Map<NormalizedRelayUrl, RelayProber.ReadWriteVerdict>? = null
+
+            val check =
+                launch {
+                    result = RelayProber(client).readWriteCheck(listOf(fast), signer, timeoutMs = 5_000)
+                }
+            launch {
+                delay(50)
+                client.listener!!.onEose(fast, null)
+                while (client.published == null) delay(10)
+                client.answerOk(foreign, true, "")
+                delay(100)
+                client.answerOk(fast, true, "")
+            }
+            check.join()
+
+            val verdict = result!![fast]!!
+            assertEquals(true, verdict.writeAccepted, "the listed relay's OK must still be awaited and recorded")
+            assertNull(result!![foreign], "the foreign relay must not appear in the result")
+        }
+
+    @Test
+    fun silentWriteLeavesTheWriteSideUnobserved() =
+        runTest {
+            val client = ScriptedClient()
+            val signer = NostrSignerInternal(KeyPair())
+            var result: Map<NormalizedRelayUrl, RelayProber.ReadWriteVerdict>? = null
+
+            val check =
+                launch {
+                    result = RelayProber(client).readWriteCheck(listOf(fast), signer, timeoutMs = 2_000)
+                }
+            launch { client.playReadThenWrite(fast, ok = null) }
+            check.join()
+
+            val verdict = result!![fast]!!
+            assertTrue(verdict.rttReadMs >= 0)
+            assertNull(verdict.writeAccepted, "silence is not evidence about the write path")
+            assertEquals(-1, verdict.rttWriteMs)
+        }
+
+    // ------------------------------------------------------------------
+    // toDiscoveryEventTemplate — only observed facts become tags
+    // ------------------------------------------------------------------
+
+    private fun tagsOf(template: EventTemplate<*>) = template.tags.map { it.toList() }
+
+    @Test
+    fun reachableVerdictTemplateCarriesLivenessAndNetwork() {
+        val template =
+            RelayProber
+                .Verdict(fast, reachable = true, rttOpenMs = 150, rttEoseMs = 480, error = null)
+                .toDiscoveryEventTemplate(createdAt = 1000)
+
+        val tags = tagsOf(template)
+        assertEquals(30166, template.kind)
+        assertEquals(1000, template.createdAt)
+        assertTrue(listOf("d", fast.url) in tags)
+        assertTrue(listOf("n", "clearnet") in tags)
+        assertTrue(listOf("rtt-open", "150") in tags)
+        // rtt-eose is wave-relative (dial + queue + read) — never published as rtt-read.
+        assertNull(tags.firstOrNull { it[0] == "rtt-read" })
+    }
+
+    @Test
+    fun deadVerdictTemplateHasNoRttOpen() {
+        val template =
+            RelayProber
+                .Verdict(silent, reachable = false, rttOpenMs = -1, rttEoseMs = -1, error = "cannot:timeout")
+                .toDiscoveryEventTemplate()
+
+        val tags = tagsOf(template)
+        assertTrue(listOf("d", silent.url) in tags)
+        // Liveness is the PRESENCE of rtt-open; a dead record must not carry one.
+        assertNull(tags.firstOrNull { it[0] == "rtt-open" })
+    }
+
+    @Test
+    fun reachableWithoutMeasuredLatencyWritesZeroFlag() {
+        val template =
+            RelayProber
+                .Verdict(fast, reachable = true, rttOpenMs = -1, rttEoseMs = 300, error = null)
+                .toDiscoveryEventTemplate()
+
+        // 0 = "reachable, latency not observed": the flag form, never an invented number.
+        assertTrue(listOf("rtt-open", "0") in tagsOf(template))
+    }
+
+    @Test
+    fun observedAuthWallBecomesARequirementTag() {
+        val template =
+            RelayProber
+                .Verdict(walled, reachable = true, rttOpenMs = 90, rttEoseMs = -1, error = "closed:auth-required: sign in")
+                .toDiscoveryEventTemplate()
+
+        assertTrue(listOf("R", "auth") in tagsOf(template))
+    }
+
+    @Test
+    fun policyClosedIsNotAnAuthRequirement() {
+        val template =
+            RelayProber
+                .Verdict(walled, reachable = true, rttOpenMs = 90, rttEoseMs = -1, error = "closed:blocked: not welcome")
+                .toDiscoveryEventTemplate()
+
+        assertNull(tagsOf(template).firstOrNull { it[0] == "R" })
+    }
+
+    @Test
+    fun readWriteResultsBecomeRttTags() {
+        val verdict = RelayProber.Verdict(fast, reachable = true, rttOpenMs = 100, rttEoseMs = 300, error = null)
+        val readWrite = RelayProber.ReadWriteVerdict(fast, rttReadMs = 40, rttWriteMs = 55, writeAccepted = true, writeMessage = "")
+
+        val tags = tagsOf(verdict.toDiscoveryEventTemplate(readWrite = readWrite))
+        assertTrue(listOf("rtt-read", "40") in tags)
+        assertTrue(listOf("rtt-write", "55") in tags)
+    }
+
+    @Test
+    fun unobservedReadWriteSidesStayUntagged() {
+        val verdict = RelayProber.Verdict(fast, reachable = true, rttOpenMs = 100, rttEoseMs = -1, error = null)
+        val readWrite = RelayProber.ReadWriteVerdict(fast, rttReadMs = -1, rttWriteMs = -1, writeAccepted = null, writeMessage = null)
+
+        val tags = tagsOf(verdict.toDiscoveryEventTemplate(readWrite = readWrite))
+        assertNull(tags.firstOrNull { it[0] == "rtt-read" })
+        assertNull(tags.firstOrNull { it[0] == "rtt-write" })
+    }
+
+    @Test
+    fun writeRejectionReasonsBecomeRequirementTags() {
+        val verdict = RelayProber.Verdict(walled, reachable = true, rttOpenMs = 100, rttEoseMs = -1, error = null)
+
+        val pow = RelayProber.ReadWriteVerdict(walled, -1, 30, writeAccepted = false, writeMessage = "pow: 28 bits needed")
+        assertTrue(listOf("R", "pow") in tagsOf(verdict.toDiscoveryEventTemplate(readWrite = pow)))
+
+        val auth = RelayProber.ReadWriteVerdict(walled, -1, 30, writeAccepted = false, writeMessage = "auth-required: sign in")
+        assertTrue(listOf("R", "auth") in tagsOf(verdict.toDiscoveryEventTemplate(readWrite = auth)))
+
+        val blocked = RelayProber.ReadWriteVerdict(walled, -1, 30, writeAccepted = false, writeMessage = "blocked: not welcome")
+        assertNull(tagsOf(verdict.toDiscoveryEventTemplate(readWrite = blocked)).firstOrNull { it[0] == "R" })
+    }
+
+    @Test
+    fun onionRelayIsTaggedTor() {
+        val onion = RelayUrlNormalizer.normalize("ws://someonionaddressabcdefghijklmnop.onion")
+        val template =
+            RelayProber
+                .Verdict(onion, reachable = true, rttOpenMs = 900, rttEoseMs = -1, error = null)
+                .toDiscoveryEventTemplate()
+
+        assertTrue(listOf("n", "tor") in tagsOf(template))
+    }
+}

quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip11RelayInfo/OkHttpNip11Fetcher.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.nip11RelayInfo
+
+import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
+import com.vitorpamplona.quartz.nip01Core.relay.normalizer.toHttp
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.withContext
+import okhttp3.OkHttpClient
+import okhttp3.Request
+import okhttp3.coroutines.executeAsync
+
+/**
+ * OkHttp-backed [Nip11Fetcher]: GETs the relay's https url with
+ * `Accept: application/nostr+json` and parses the document. The client is
+ * resolved per relay so callers can route Tor/proxy relays through a different
+ * OkHttp instance (the same seam Amethyst's Nip11Retriever uses).
+ */
+class OkHttpNip11Fetcher(
+    private val okHttpClient: (NormalizedRelayUrl) -> OkHttpClient,
+) : Nip11Fetcher {
+    override suspend fun fetch(relay: NormalizedRelayUrl): Nip11RelayInformation =
+        withContext(Dispatchers.IO) {
+            val request =
+                Request
+                    .Builder()
+                    .header("Accept", "application/nostr+json")
+                    .url(relay.toHttp())
+                    .build()
+
+            okHttpClient(relay).newCall(request).executeAsync().use { response ->
+                if (!response.isSuccessful) {
+                    throw Nip11FetchException("HTTP ${response.code} fetching NIP-11 from ${relay.url}")
+                }
+                val body = response.body.string()
+                if (!body.startsWith("{")) {
+                    throw Nip11FetchException("Not a NIP-11 document from ${relay.url}")
+                }
+                Nip11RelayInformation.fromJson(body)
+            }
+        }
+}