From 5f65405b08f5251e34586fdac8e1fe81254bb139 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 17 Jul 2026 19:33:46 +0000 Subject: [PATCH] feat(graperank): add reverse follower crawl (amy graperank followers) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The outbox model can't find an observer's followers — you don't know a follower exists until you've seen their kind:3, so you can't route to their outbox first. FollowerCrawler casts a wide net instead: it asks as many relays as possible for kind:3 lists that #p-tag the observer, paging each relay past its per-REQ cap via fetchAllPagesFromPool, verifies with ParallelEventVerifier, keeps only lists that genuinely tag the observer, dedups by id, and group-commits to the store. Each follower's list is a full contact list, so persisting it also enriches the graph a later `graperank score` builds — every follower becomes a FOLLOW edge into the observer. CLI: `amy graperank followers [OBSERVER]` assembles "all possible relays" from the reachability-cache live set + every kind:10002/30166 relay in the store + the index/aggregator relays, skipping proven-dead relays. Runs anonymously (no signing) when given an explicit observer. Tunable via --relay/--page-limit/--timeout/--relay-concurrency/--insert-batch. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Xc3Wm4qVCrAGvSAotTUVt4 --- cli/README.md | 1 + .../amethyst/cli/commands/GrapeRankCommand.kt | 119 +++++++++++ .../experimental/graperank/FollowerCrawler.kt | 189 ++++++++++++++++++ 3 files changed, 309 insertions(+) create mode 100644 quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/experimental/graperank/FollowerCrawler.kt diff --git a/cli/README.md b/cli/README.md index c78efa9d56..257e5af825 100644 --- a/cli/README.md +++ b/cli/README.md @@ -391,6 +391,7 @@ HTTP endpoint. Reuses quartz's `Nip86Client` and the shared `Nip86Retriever` | `amy follow USER` / `amy unfollow USER` | Add/remove USER from your kind:3 contact list (fetches the freshest list first). | | `amy graperank [OBSERVER] [--offline] [--min-rank N]` | Crawl + score: compute GrapeRank web-of-trust scores (0..1) over the follow/mute/report graph, then persist the result. Exhaustively crawls each user's kind:10002 outbox for their latest kind:3/10000/1984 until every discovered user is checked (no user cap), dropping reports the author retracted via NIP-09. **Every score run persists its result locally**: the cards (rank cutoff `--min-rank`, default 2) are reconciled into the shared store as NIP-85 kind:30382 cards signed by a per-observer **service key** — each carries `rank`, `followers` (trusted-follower count, cutoff `--followers-threshold`, default 0.02) and `hops` (follow distance from the observer); changed cards re-signed, unchanged skipped (no event-id churn), dropped targets retracted (kind:5). `--offline` skips the crawl. | | `amy graperank crawl [OBSERVER] [--max-hops N] [--no-preconnect]` | Pipeline stage 1 — network only: crawl the follow/mute/report graph (kind 3/10000/1984/10002) into the local store, no scoring. Idempotent and cumulative: run it a few times to load everything, then `score`. | +| `amy graperank followers [OBSERVER] [--relay URL[,URL…]]` | The **reverse** crawl: find every user who *follows* the observer by asking as many relays as possible for kind:3 lists that `#p`-tag them (paged past each relay's cap). The outbox model can't find followers — you don't know one exists until you've seen their list — so this casts a wide net over the whole relay universe (reachability-cache live set + every kind:10002/30166 relay in the store + index/aggregator relays), skipping proven-dead relays. Persists each follower's list, enriching the graph a later `score` builds. Idempotent and cumulative. | | `amy graperank score [OBSERVER]` | Pipeline stage 2 — local only: score from the store and persist the cards (identical to bare `--offline`; same scoring flags). No network, so re-run with different `--rigor`/`--attenuation`/`--min-rank` without re-crawling. | | `amy graperank publish [OBSERVER] [--relay URL[,URL…]]` | Pipeline stage 3 — transport only: make the operator relay(s) converge to the locally persisted card set — one NIP-77 up-only reconcile per relay over the service key's kind:30382 + kind:5 (nothing is re-scored or re-signed; a relay that can't reconcile gets the full set published instead). Also refreshes the observer's kind:10040 pointer when we hold their key. | | `amy graperank rank USER [--provider PUBKEY] [--refresh]` | The consumer side: read the kind:30382 cards about USER — one rank per provider, newest card each. Local store first; `--refresh` (or a miss) drains the operator relays, the relays your kind:10040 declares, and the bootstrap set. | diff --git a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/GrapeRankCommand.kt b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/GrapeRankCommand.kt index 36a5aa6b36..e028d0335e 100644 --- a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/GrapeRankCommand.kt +++ b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/GrapeRankCommand.kt @@ -26,6 +26,7 @@ import com.vitorpamplona.amethyst.cli.DataDir import com.vitorpamplona.amethyst.cli.Output import com.vitorpamplona.amethyst.commons.defaults.Constants import com.vitorpamplona.amethyst.commons.defaults.DefaultIndexerRelayList +import com.vitorpamplona.quartz.experimental.graperank.FollowerCrawler import com.vitorpamplona.quartz.experimental.graperank.GrapeRank import com.vitorpamplona.quartz.experimental.graperank.GrapeRankCrawler import com.vitorpamplona.quartz.experimental.graperank.GrapeRankParams @@ -213,6 +214,7 @@ object GrapeRankCommand { "providers" -> providers(dataDir, tail.drop(1).toTypedArray()) "operator" -> operator(dataDir, tail.drop(1).toTypedArray()) "crawl" -> crawl(dataDir, tail.drop(1).toTypedArray()) + "followers" -> followers(dataDir, tail.drop(1).toTypedArray()) "status" -> status(dataDir) // The relay census outgrew graperank (it feeds the shared NIP-66 // reachability cache every command reads) and moved to `amy relay @@ -586,6 +588,123 @@ object GrapeRankCommand { return 0 } + /** + * `amy graperank followers [OBSERVER] [flags]` — the reverse crawl: find every + * user who FOLLOWS the observer and persist their kind:3 contact lists. + * + * The outbox model (what `crawl` uses) walks follows *outward* and can't find + * followers — you don't know a follower exists until you've seen their list, so + * you can't route to their outbox first. This casts a wide net instead: it asks + * as many relays as possible for kind:3 events that `#p`-tag the observer, paging + * each relay past its per-REQ cap. The relay universe is "all possible relays": + * the reachability-cache live set + every kind:10002 read/write relay and every + * kind:30166 monitored relay in the store + the index/aggregator relays that hold + * reverse-follow data for the network. Proven-dead relays are skipped + * (`--no-reachability-cache` to query them anyway). + * + * Each follower's list lands in the store, so it also enriches the graph a later + * `graperank score` builds — every follower becomes a FOLLOW edge into the + * observer. Idempotent and cumulative; run it a few times for completeness. + * + * Flags: `--relay URL[,URL…]` (query only these instead of the whole universe), + * `--page-limit N` (per-page REQ limit, default 500), `--timeout SECS` (per-page + * EOSE watchdog, default 15), `--relay-concurrency N` (relays paged at once, + * default 16), `--insert-batch N` (events per store commit, default 500). + */ + private suspend fun followers( + dataDir: DataDir, + rest: Array, + ): Int { + val args = Args(rest) + val observerArg = args.positionalOrNull(0) + val relayArg = args.flag("relay") + val relayConcurrency = args.intFlag("relay-concurrency", args.intFlag("concurrency", 16)) + + // Read-only + never signs, so it runs anonymously — but then an OBSERVER + // positional is required (there's no account to default to). + Context.openOrAnonymous(dataDir).use { ctx -> + ctx.prepare() + if (ctx.anonymous && observerArg == null) { + return Output.error("bad_args", "usage: amy graperank followers OBSERVER [flags] (no account — pass an observer)") + } + val observer = observerArg?.let { ctx.requireUserHex(it) } ?: ctx.identity.pubKeyHex + + val relays = + relayArg + ?.split(",") + ?.mapNotNull { RelayUrlNormalizer.normalizeOrNull(it.trim()) } + ?.toSet() + ?.takeIf { it.isNotEmpty() } + ?: allKnownRelays(ctx, args) + + if (relays.isEmpty()) { + return Output.error("no_relays", "no relays to query — run `amy relay probe` / `amy graperank crawl` first, or pass --relay") + } + + System.err.println("[followers] querying ${relays.size} relays for kind:3 #p=${observer.take(8)}…") + val crawler = + FollowerCrawler( + client = ctx.client, + store = ctx.store, + config = + FollowerCrawler.Config( + relays = relays, + pageLimit = args.intFlag("page-limit", 500), + timeoutMs = args.longFlag("timeout", 15L) * 1000, + maxConcurrentRelays = relayConcurrency, + insertBatchSize = args.intFlag("insert-batch", 500), + ), + log = { System.err.println(it) }, + ) + val stats = crawler.crawl(observer) + + Output.emit( + linkedMapOf( + "observer" to observer, + "relays_queried" to stats.relaysQueried, + "relays_answered" to stats.relaysAnswered, + "followers_found" to stats.followersFound, + "events_stored" to stats.eventsStored, + "download_ms" to stats.downloadMs, + ), + ) + } + return 0 + } + + /** + * "All possible relays" for the follower crawl: the reachability-cache live set, + * every kind:10002 read/write relay and every kind:30166 monitored relay the + * store knows, and the index/aggregator relays that hold reverse-follow data — + * minus the ones a prior probe proved dead (unless `--no-reachability-cache`). + */ + private suspend fun allKnownRelays( + ctx: Context, + args: Args, + ): Set { + val reach = ctx.reachability.snapshot() + val relays = LinkedHashSet() + relays.addAll(reach.live) + // Every advertised outbox/inbox relay in the store — the homes of the users + // whose followers we're hunting are exactly where those followers publish too. + for (event in ctx.store.query(Filter(kinds = listOf(AdvertisedRelayListEvent.KIND)))) { + if (event is AdvertisedRelayListEvent) relays.addAll(event.relaysNorm()) + } + // Relays a NIP-66 monitor has ever seen (kind:30166) — the widest census. + for (event in ctx.store.query(Filter(kinds = listOf(RelayDiscoveryEvent.KIND)))) { + if (event is RelayDiscoveryEvent) event.relay()?.let { relays.add(it) } + } + // Aggregators/indexers hold reverse-follow data for users whose own outbox + // we've never learned — the biggest completeness lever for followers. + relays.addAll(CONTENT_AGGREGATOR_RELAYS) + relays.addAll(EXTRA_DISCOVERY_RELAYS) + relays.addAll(ctx.bootstrapRelays()) + relays.addAll(Constants.eventFinderRelays) + + if (!args.bool(NO_REACHABILITY_CACHE_FLAG)) relays.removeAll(reach.dead) + return relays + } + /** * `amy graperank status` — read-only inventory of everything a GrapeRank run * depends on, straight from the local store: WoT record counts (the "do I diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/experimental/graperank/FollowerCrawler.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/experimental/graperank/FollowerCrawler.kt new file mode 100644 index 0000000000..4456015dae --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/experimental/graperank/FollowerCrawler.kt @@ -0,0 +1,189 @@ +/* + * 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.experimental.graperank + +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.core.HexKey +import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.ParallelEventVerifier +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.fetchAllPagesFromPool +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import com.vitorpamplona.quartz.nip01Core.store.IEventStore +import com.vitorpamplona.quartz.nip01Core.tags.people.isTaggedUser +import com.vitorpamplona.quartz.nip02FollowList.ContactListEvent +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.launch +import kotlinx.coroutines.withContext +import kotlin.time.TimeSource + +/** + * The reverse of [GrapeRankCrawler]: finds every user who **follows** the observer + * by asking as many relays as possible for kind:3 contact lists that `#p`-tag the + * observer, and persists each one to [store]. The outbox model can't find these — + * you don't know a follower until you've seen their list, so you can't route to + * their outbox first — so this casts a wide net across the whole relay universe + * ([Config.relays], assembled by the caller from the reachability cache, the + * kind:10002 relays in the store, and the index/aggregator relays that hold + * reverse-follow data for the network). + * + * Each discovered kind:3 is a follower's full contact list, so persisting it also + * enriches the graph a later `graperank score` builds from the store — the + * follower becomes a FOLLOW edge into the observer (and into everyone else it + * follows). A popular observer can have far more followers than a relay's per-REQ + * cap, so every relay is paged on its own `created_at` cursor via + * [fetchAllPagesFromPool]. + * + * Transport-agnostic within quartz: it takes an [INostrClient] and an + * [IEventStore]; relay *policy* (which relays make up "all possible relays") is the + * caller's, injected through [Config]. Progress is emitted through [log]. + */ +class FollowerCrawler( + private val client: INostrClient, + private val store: IEventStore, + private val config: Config, + private val log: (String) -> Unit = {}, +) { + /** + * @param relays the relay universe to query — "all possible relays" is the + * caller's policy (reachability-cache live set + every kind:10002 relay in the + * store + the index/aggregator relays). Empty means nothing to do. + * @param pageLimit per-page `limit` handed to each relay's paged REQ; the cursor + * walks past it, so this only bounds one page, not the total. + * @param timeoutMs per-page EOSE timeout for a relay before its next page fires. + * @param maxConcurrentRelays how many relays page at once (a global fan-out cap). + * @param insertBatchSize verified events group-committed per [IEventStore.batchInsert]. + */ + class Config( + val relays: Set, + val pageLimit: Int = 500, + val timeoutMs: Long = 15_000, + val maxConcurrentRelays: Int = 16, + val insertBatchSize: Int = 500, + ) + + /** What the follower crawl found. */ + class Stats( + val relaysQueried: Int, + /** Relays that returned at least one valid follower list. */ + val relaysAnswered: Int, + /** Distinct followers (kind:3 authors that `#p`-tag the observer). */ + val followersFound: Int, + /** Verified kind:3 events handed to the store (superseded versions included). */ + val eventsStored: Long, + val downloadMs: Long, + ) + + /** + * Find and persist every follower of [observer] across [Config.relays]. Verifies + * each event (a relay can lie), keeps only kind:3 that actually `#p`-tag the + * observer (a relay can over-return), dedups by event id, and group-commits to + * the store. Returns [Stats]. + */ + suspend fun crawl(observer: HexKey): Stats = + withContext(Dispatchers.IO) { + if (config.relays.isEmpty()) return@withContext Stats(0, 0, 0, 0L, 0L) + + val mark = TimeSource.Monotonic.markNow() + val filter = Filter(kinds = FOLLOW_KINDS, tags = mapOf("p" to listOf(observer)), limit = config.pageLimit) + val perRelay = config.relays.associateWith { listOf(filter) } + + // These three are read/written ONLY by the verifier's single drain + // coroutine (ParallelEventVerifier dispatches onVerified in order from one + // coroutine), so plain collections are safe — no concurrent access. + val followers = HashSet() + val seen = HashSet() + val answered = HashSet() + var eventsStored = 0L + + // Verified follower lists cross to a single persister coroutine that can + // suspend on the store; onVerified itself must not (it's a plain callback). + val toPersist = Channel(Channel.UNLIMITED) + + coroutineScope { + val persister = + launch(Dispatchers.IO) { + val flushAt = config.insertBatchSize.coerceAtLeast(1) + val buffer = ArrayList(flushAt) + + suspend fun flush() { + if (buffer.isEmpty()) return + store.batchInsert(buffer) + eventsStored += buffer.size + buffer.clear() + } + for (event in toPersist) { + buffer.add(event) + if (buffer.size >= flushAt) flush() + } + flush() + } + + val verifier = + ParallelEventVerifier( + scope = this, + onVerified = { event, relay -> + // A relay can over-return; keep only genuine followers, and + // dedup by id so a list mirrored across relays is stored once. + if (event is ContactListEvent && event.isTaggedUser(observer) && seen.add(event.id)) { + followers.add(event.pubKey) + answered.add(relay) + toPersist.trySend(event) + } + }, + ) + + client.fetchAllPagesFromPool( + filters = perRelay, + timeoutMs = config.timeoutMs, + maxConcurrentRelays = config.maxConcurrentRelays, + onRelayComplete = { relay, total -> + if (total > 0) log("[followers] ${relay.url}: $total kind:3 pages drained") + }, + ) { event, relay -> + // onEvent runs on the relay reader thread and must not suspend; + // submit() is a non-suspending channel send that backpressures. + if (event.kind == ContactListEvent.KIND) verifier.submit(event, relay) + } + + // No more events will arrive: drain the verifier, then the persister. + verifier.close() + verifier.join() + toPersist.close() + persister.join() + } + + Stats( + relaysQueried = config.relays.size, + relaysAnswered = answered.size, + followersFound = followers.size, + eventsStored = eventsStored, + downloadMs = mark.elapsedNow().inWholeMilliseconds, + ) + } + + companion object { + /** Reverse-follow lookup is kind:3 only — a follower IS the author of a kind:3 that `#p`-tags the observer. */ + private val FOLLOW_KINDS = listOf(ContactListEvent.KIND) + } +}