From 58e018c87777413c1c4e4d44ebd2352f126a0de8 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 14 Jul 2026 20:29:09 +0000 Subject: [PATCH] feat(cli): graperank scores always persist locally; add publish + rank verbs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every `amy graperank` / `graperank score` run now reconciles its result into the local event store as NIP-85 kind:30382 cards signed by the per-observer service key (cutoff --min-rank, default 2): changed ranks are re-signed, unchanged ones skipped (no event-id churn), dropped targets retracted with a kind:5 the store applies on insert. The store is the source of truth, so the score of a run is durable and reusable instead of ephemeral stdout. New verbs complete the crawl -> score -> publish pipeline: - `graperank publish [OBSERVER] [--relay URL[,URL...]]` — transport only: one NIP-77 up-only reconcile per operator relay over the provider key's kind:30382 + kind:5, converging the relay to the local set (deletions propagate, lost cards restored, full-set publish fallback for relays without negentropy); also refreshes the observer's kind:10040 pointer. - `graperank rank USER [--provider P] [--refresh]` — the consumer side: newest card per provider from the local store, with a relay drain on miss. GrapeRankPublisher (quartz) is reworked accordingly: reconcileAndPublish is replaced by reconcileLocal (sign+insert+retract against the store) and syncToRelays (NegentropyStoreSync up-only + blastPublish fallback), with a commonTest covering upsert/skip/retract and re-carding a retracted target. BREAKING (--json / flags): `--publish`, `--publish-limit`, `--publish-relay` and `--bench-sign` are removed from the score command. Its JSON drops published/publish_rejected/deleted/delete_rejected/skipped_unchanged/ publish_truncated/published_kind/published_to/publish_error/observer_10040/ bench_signed/bench_sign_ms and gains provider_pubkey/min_rank/cards_total/ cards_signed/cards_unchanged/cards_retracted/cards_ms. Publishing moved to the new `graperank publish` verb. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_013WSzVX9RoUxyV3nT3dfc56 --- cli/README.md | 43 +- cli/ROADMAP.md | 2 +- .../com/vitorpamplona/amethyst/cli/Main.kt | 29 +- .../amethyst/cli/commands/GrapeRankCommand.kt | 377 ++++++++++++------ .../graperank/GrapeRankPublisher.kt | 315 ++++++++++----- .../graperank/GrapeRankPublisherTest.kt | 128 ++++++ 6 files changed, 645 insertions(+), 249 deletions(-) create mode 100644 quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/experimental/graperank/GrapeRankPublisherTest.kt diff --git a/cli/README.md b/cli/README.md index 928009c996..bf393da508 100644 --- a/cli/README.md +++ b/cli/README.md @@ -389,14 +389,16 @@ HTTP endpoint. Reuses quartz's `Nip86Client` and the shared `Nip86Retriever` | `amy profile show [USER]` | Print kind:0 metadata. USER accepts npub/nprofile/hex/NIP-05; defaults to self. | | `amy profile edit --name … --about … --picture URL …` | Patch and re-publish your kind:0. | | `amy follow USER` / `amy unfollow USER` | Add/remove USER from your kind:3 contact list (fetches the freshest list first). | -| `amy graperank [OBSERVER] [--offline] [--publish] [--min-rank N] [--publish-relay URL]` | Compute GrapeRank web-of-trust scores (0..1) over the follow/mute/report graph. 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. With `--publish`, reconciles NIP-85 kind:30382 cards signed by a per-observer **service key**: publishes changed/new ranks (cutoff `--min-rank`, default 2), skips unchanged, and **retracts** (kind:5) any card whose target left the graph or fell below the cutoff. | -| `amy graperank operator [status \| relay … \| providers]` | Manage the machine's operator keys (independent of any account, under `~/.amy/operator/`). `relay` sets where cards + retractions publish; `status` shows the master pubkey and relays; `providers` lists the observer → service-pubkey map. | +| `amy graperank [OBSERVER] [--offline] [--min-rank N]` | Compute GrapeRank web-of-trust scores (0..1) over the follow/mute/report graph. 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 ranks (cutoff `--min-rank`, default 2) are reconciled into the shared store as NIP-85 kind:30382 cards signed by a per-observer **service key** — changed ranks re-signed, unchanged skipped (no event-id churn), dropped targets retracted (kind:5). `graperank score` is the local-only variant (same as `--offline`). | +| `amy graperank publish [OBSERVER] [--relay URL[,URL…]]` | 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. | +| `amy graperank operator [status \| relay … \| providers]` | Manage the machine's operator keys (independent of any account, under `~/.amy/operator/`). `relay` sets where `publish` sends cards + retractions; `status` shows the master pubkey and relays; `providers` lists the observer → service-pubkey map. | | `amy graperank register [PROVIDER] [--service KIND:TAG] [--relay URL]` | Declare a NIP-85 provider in your kind:10040 so clients can discover it (default: self as the `30382:rank` provider). | | `amy graperank providers [USER]` | List a user's declared NIP-85 trusted providers (public + your own private entries). | -#### Publishing GrapeRank scores (NIP-85) +#### GrapeRank scores are persisted locally, then published (NIP-85) -Ranks are published as kind:30382 cards, but **not** under your account key. A +Ranks are signed as kind:30382 cards, but **not** under your account key. A machine holds one **operator master** seed (`~/.amy/operator/`, stored via the same `--secret-backend` as accounts, independent of any account). From it a distinct, deterministic **service key** is derived per observer: @@ -406,23 +408,32 @@ serviceKey(observer) = sha256(masterPriv ‖ "graperank-provider:" ‖ observerH ``` Because kind:30382 is addressable (`pubkey + d-tag`), the stable per-observer key -means re-publishing **replaces** a target's card instead of orphaning it — and -losing everything but the master seed still re-derives every key. Set up once and -publish: +means re-signing **replaces** a target's card instead of orphaning it — and +losing everything but the master seed still re-derives every key. + +**Every score run persists its cards.** After scoring, Amy reconciles the result +into the local store: new or changed ranks (≥ `--min-rank`, default 2) are +signed; unchanged ranks are skipped (no new event id); and any card whose target +dropped out of the graph or fell below the cutoff is **retracted** with a kind:5 +(the store applies it; the tombstone is kept). The local store is the source of +truth — `graperank rank USER` reads it offline, and `graperank publish` mirrors +it out: ```bash amy graperank operator relay wss://relay.example.com # where all cards live -amy graperank --publish # sign with the observer's service key +amy graperank # crawl + score + persist cards locally +amy graperank publish # make the operator relay match the local set ``` -Each publish **reconciles** against what the service key already published: new or -changed ranks (≥ `--min-rank`, default 2) are signed and sent; unchanged ranks are -skipped (no new event id); and any card whose target dropped out of the graph or -fell below the cutoff is **retracted** with a kind:5. When the observer is your -own account (we hold the key), Amy also writes their kind:10040 pointing -`30382:rank → serviceKey @ operator relay` to their outbox, so clients can find -the cards. For a third-party observer, `graperank operator providers` prints the -`observer → service-pubkey` mapping to wire their kind:10040 out-of-band. +`publish` never re-scores or re-signs: it runs one NIP-77 up-only reconcile per +relay over the service key's kind:30382 + kind:5, so the relay converges to the +local card set (deletions included, lost cards restored); a relay that can't +negentropy-reconcile gets the full set published event-by-event instead. When +the observer is your own account (we hold the key), `publish` also writes their +kind:10040 pointing `30382:rank → serviceKey @ operator relay` to their outbox, +so clients can find the cards. For a third-party observer, `graperank operator +providers` prints the `observer → service-pubkey` mapping to wire their +kind:10040 out-of-band. ### Direct messages (NIP-17) diff --git a/cli/ROADMAP.md b/cli/ROADMAP.md index d0fc574d7c..6144320f7b 100644 --- a/cli/ROADMAP.md +++ b/cli/ROADMAP.md @@ -61,7 +61,7 @@ Status legend: ✅ shipped · 📦 logic lives in `commons/`, needs a command · | NIP-51 lists (bookmarks, mute, follow sets) | 🆕 | `amethyst/model/nip51Lists/` | | NIP-57 zaps (send + verify) | 🆕 | Needs LN-URL plumbing; `amethyst/service/lnurl/`. | | NIP-65 outbox model queries | 🆕 | | -| NIP-85 GrapeRank web-of-trust (`amy graperank`) | ✅ | `GrapeRankCommand` — outbox-model crawl + scoring engine in `commons/wot/` (`GrapeRank`, `TrustGraph`, `TrustGraphBuilder`); publishes kind:30382 `ContactCardEvent` (diffed against prior ranks), plus `register` / `providers` for the kind:10040 `TrustProviderListEvent` discovery layer. | +| NIP-85 GrapeRank web-of-trust (`amy graperank`) | ✅ | `GrapeRankCommand` — outbox-model crawl + scoring engine in `commons/wot/` (`GrapeRank`, `TrustGraph`, `TrustGraphBuilder`); every score run persists kind:30382 `ContactCardEvent` cards to the local store (diffed against prior ranks, kind:5 retractions), `publish` mirrors that set to the operator relays via NIP-77 up-sync, `rank` reads cards back, plus `register` / `providers` for the kind:10040 `TrustProviderListEvent` discovery layer. | | NIP-72 communities | 🆕 | | | NIP-78 app-specific data (settings sync) | 🆕 | | | Long-form (NIP-23) publish / read | 🆕 | | diff --git a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Main.kt b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Main.kt index f8c304711d..d7410753bc 100644 --- a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Main.kt +++ b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Main.kt @@ -611,17 +611,30 @@ private fun printUsage() { | [--rigor X] [--attenuation X] Exhaustively crawls each user's kind:10002 outbox | [--max-rounds N] [--max-hops N] for their latest kind:3/10000/1984 until every | [--offline] [--timeout SECS] discovered user has been checked (no user cap; - | [--diagnose] --max-hops bounds follow distance, e.g. 8; + | [--diagnose] [--min-rank N] --max-hops bounds follow distance, e.g. 8; | --diagnose dumps per-relay telemetry: outcome | mix, yield, latency, and a LIVE/DEAD + limits | classification table of every relay contacted). - | [--publish] [--min-rank N] OBSERVER: npub|nprofile|hex|name@domain (default: - | [--publish-limit N] [--publish-relay URL] active account). --offline scores from the local - | store only. --publish reconciles NIP-85 kind:30382 - | cards signed by a per-observer service key: sends - | new/changed ranks >= --min-rank (default 2), skips - | unchanged, and retracts (kind:5) any card whose - | target left the graph or fell below the cutoff. + | OBSERVER: npub|nprofile|hex|name@domain (default: + | active account). --offline scores from the local + | store only. EVERY score run persists its result + | locally as NIP-85 kind:30382 cards signed by a + | per-observer service key (ranks >= --min-rank, + | default 2): changed ranks re-signed, unchanged + | skipped, dropped targets retracted (kind:5). + | `graperank publish` pushes that set to relays. + | graperank score [OBSERVER] local only: build the graph from the store and + | score (same as bare --offline; same flags). Fast + | and param-tunable without re-crawling. + | graperank publish [OBSERVER] transport only: make the operator relay(s) match + | [--relay URL[,URL…]] [--timeout SECS] the local card set via one NIP-77 up-only + | [--relay-concurrency N] reconcile per relay (nothing re-scored or + | re-signed; full-set publish fallback when a relay + | can't reconcile). Also refreshes the observer's + | kind:10040 pointer when we hold their key. + | graperank rank USER [--provider PUBKEY] read the kind:30382 cards about USER — one rank + | [--refresh] [--timeout SECS] per provider, local store first, --refresh drains + | the operator/declared/bootstrap relays. | graperank crawl [OBSERVER] network only: crawl the WoT graph (kind 3/10000/ | [--max-hops N] [--preconnect-cap N] 1984/10002) into the local store without scoring. | [--no-preconnect] Pre-connects every known-live relay in one parallel 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 243b88264d..124756afae 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 @@ -35,11 +35,9 @@ import com.vitorpamplona.quartz.experimental.graperank.TrustGraphBuilder import com.vitorpamplona.quartz.nip01Core.core.Event import com.vitorpamplona.quartz.nip01Core.core.HexKey import com.vitorpamplona.quartz.nip01Core.core.toHexKey -import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair 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.NostrSigner import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerInternal import com.vitorpamplona.quartz.nip02FollowList.ContactListEvent import com.vitorpamplona.quartz.nip09Deletions.DeletionEvent @@ -53,13 +51,8 @@ import com.vitorpamplona.quartz.nip85TrustedAssertions.list.tags.ProviderTypes import com.vitorpamplona.quartz.nip85TrustedAssertions.list.tags.ServiceProviderTag import com.vitorpamplona.quartz.nip85TrustedAssertions.list.tags.ServiceType import com.vitorpamplona.quartz.nip85TrustedAssertions.users.ContactCardEvent -import com.vitorpamplona.quartz.nip85TrustedAssertions.users.tags.RankTag import kotlinx.coroutines.CancellationException -import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.asCoroutineDispatcher -import kotlinx.coroutines.async -import kotlinx.coroutines.awaitAll -import kotlinx.coroutines.coroutineScope import kotlinx.coroutines.withContext import java.net.InetSocketAddress import java.net.Socket @@ -80,26 +73,33 @@ import kotlin.math.roundToInt * unreachable outbox is retried a few times), then runs the scoring engine in * `commons/wot`. * - * Prints a ranked list (text, or one JSON object under `--json`). With - * `--publish`, results are also published as NIP-85 kind:30382 `ContactCardEvent` - * trusted assertions (one per scored user, `rank = round(score*100)`). - * - * The crawl and the computation are separable, because the crawl persists every - * event it fetches to the store and the score is a pure function over it: + * The pipeline is three separable stages, each with its own verb, and the local + * store is the source of truth between them (the crawl persists every event it + * fetches; the score is a pure function over the store; every score run persists + * its result as locally-signed NIP-85 kind:30382 cards): * - `amy graperank crawl [OBSERVER]` — network only: crawl the reachable graph's * kind 3/10000/1984/10002 into the local store (aliased as the former `sync`). * Idempotent and cumulative, so run it a few times to make sure everything is * loaded. Scores nothing. - * - `amy graperank score [OBSERVER]` — local only: build the graph from the store - * and score (same as bare `--offline`). Instant and param-tunable; repeat with - * different `--rigor`/`--attenuation`/cutoffs without re-crawling. + * - `amy graperank score [OBSERVER]` — local only: build the graph from the store, + * score (same as bare `--offline`), and ALWAYS reconcile the result into the + * store as kind:30382 [ContactCardEvent] cards signed by the observer's + * per-observer service key (`rank = round(score*100)`, cutoff `--min-rank`): + * changed ranks are re-signed, unchanged ones skipped, dropped targets + * retracted with a kind:5. That persisted card set is what `publish` and + * `rank` reuse — scores are never ephemeral. + * - `amy graperank publish [OBSERVER]` — transport only: make the operator + * relay(s) converge to the local card set via a NIP-77 up-only reconcile + * (nothing is re-signed or re-scored), and refresh the observer's kind:10040 + * pointer when we hold their key. * - `amy graperank probe` — the relay census: mass-connect every relay the store * knows and record live/dead + measured RTT into the reachability cache, so the * next crawl skips the dead and pre-connects the living in one parallel storm. * - bare `amy graperank [OBSERVER]` — the convenience combo: crawl then score. * - * Sub-verbs complete the NIP-85 provider experience — the discovery layer that - * lets clients find and consume those assertions: + * Sub-verbs complete the NIP-85 experience — discovery and consumption: + * - `amy graperank rank USER` — read the kind:30382 cards about USER (local + * store first, `--refresh` to drain providers' relays): the consumer side. * - `amy graperank register` — advertise a `30382:rank` provider in the * account's kind:10040 [TrustProviderListEvent] (defaults to self, so a * provider publishing ranks announces where to find them). @@ -204,6 +204,8 @@ object GrapeRankCommand { "probe" -> probe(dataDir, tail.drop(1).toTypedArray()) "update" -> update(dataDir, tail.drop(1).toTypedArray()) "score" -> run(dataDir, tail.drop(1).toTypedArray(), forceOffline = true) + "publish" -> publish(dataDir, tail.drop(1).toTypedArray()) + "rank" -> rank(dataDir, tail.drop(1).toTypedArray()) else -> run(dataDir, tail) } @@ -227,17 +229,11 @@ object GrapeRankCommand { // the result JSON, so keep local copies for that. val parkTimeoutMs = args.longFlag("park-timeout", 40L) * 1000 val insertBatch = args.intFlag("insert-batch", 500) - val doPublish = args.bool("publish") - // Publish cutoff: only cards with rank >= this are published; existing - // cards for targets below it (or gone from the graph) are retracted. Rank - // is round(score*100), so 2 drops the ~0.015-and-below barely-trusted tail. + // Card cutoff: only scores with rank >= this get a local kind:30382 card; + // existing cards for targets below it (or gone from the graph) are + // retracted. Rank is round(score*100), so 2 drops the ~0.015-and-below + // barely-trusted tail. val minRank = args.intFlag("min-rank", 2) - val publishLimit = args.intFlag("publish-limit", 500) - val publishRelaysArg = args.flag("publish-relay") - // Benchmark: build + sign one kind:30382 card per scored user (rank >= - // --min-rank) with a throwaway key and time it, WITHOUT publishing. - // Measures the id-hash + Schnorr-sign cost of emitting the full card set. - val benchSign = args.bool("bench-sign") val params = GrapeRankParams( @@ -348,84 +344,43 @@ object GrapeRankCommand { }, ) - if (doPublish) { - // The cards for THIS observer are signed by a dedicated, stable - // per-observer service key derived from the machine's operator - // master (see OperatorKeys) — not the account key. Same key across - // runs means re-signing a card replaces the addressable prior one. - val opKeys = ctx.dataDir.operatorKeys() - val serviceKey = opKeys.serviceKey(observer) - val serviceSigner = NostrSignerInternal(serviceKey) - val providerPubkey = serviceKey.pubKey.toHexKey() - result["provider_pubkey"] = providerPubkey + // Every score run persists its result: the desired card set (every user + // at or above the rank cutoff) is reconciled into the LOCAL store as + // kind:30382 cards — signed by a dedicated, stable per-observer service + // key derived from the machine's operator master (see OperatorKeys), + // not the account key. Changed ranks are re-signed (the addressable + // card is replaced), unchanged ones skipped, dropped targets retracted + // with a kind:5. `graperank publish` and `graperank rank` reuse this + // set; no relay is touched here. + val opKeys = ctx.dataDir.operatorKeys() + val serviceKey = opKeys.serviceKey(observer) + val providerPubkey = serviceKey.pubKey.toHexKey() + val desiredCards = + rankedIds + .filter { rankOf(scores[it]) >= minRank } + .map { graph.pubkeyOf(it) to rankOf(scores[it]) } - // Cards go to the operator's own relay(s), where the whole - // trusted-assertion set lives; --publish-relay overrides. - val relays = - publishRelaysArg - ?.split(",") - ?.mapNotNull { RelayUrlNormalizer.normalizeOrNull(it.trim()) } - ?.toSet() - ?.takeIf { it.isNotEmpty() } - ?: opKeys.operatorRelays() + val cardsStart = System.nanoTime() + val local = + GrapeRankPublisher(ctx.store) { System.err.println(it) } + .reconcileLocal( + providerSigner = NostrSignerInternal(serviceKey), + providerPubkey = providerPubkey, + scored = desiredCards, + ) + val cardsMs = (System.nanoTime() - cardsStart) / 1_000_000 + System.err.println( + "[graperank] local cards: ${local.signed} signed, ${local.unchanged} unchanged, " + + "${local.retracted} retracted in $cardsMs ms — `amy graperank publish` pushes them to the operator relay", + ) - if (relays.isEmpty()) { - result["published"] = 0 - result["publish_error"] = "no operator relay configured — run `amy graperank operator relay ` or pass --publish-relay" - } else { - // The scorer's desired card set: every user at or above the rank - // cutoff, as (target, rank). GrapeRankPublisher reconciles this - // against what this provider key already published and upserts / - // retracts the difference. - val publishable = - rankedIds - .filter { rankOf(scores[it]) >= minRank } - .map { graph.pubkeyOf(it) to rankOf(scores[it]) } - - val publisher = GrapeRankPublisher(ctx.store) { event, to -> ctx.publish(event, to) } - val pub = - publisher.reconcileAndPublish( - providerSigner = serviceSigner, - providerPubkey = providerPubkey, - scored = publishable, - relays = relays, - publishLimit = publishLimit, - ) - - result["skipped_unchanged"] = pub.skippedUnchanged - if (pub.truncated > 0) result["publish_truncated"] = pub.truncated - result["published"] = pub.published - result["publish_rejected"] = pub.publishRejected - result["deleted"] = pub.deleted - result["delete_rejected"] = pub.deleteRejected - result["published_kind"] = ContactCardEvent.KIND - result["published_to"] = relays.map { it.url } - - // Help the observer point clients at this provider: publish their - // kind:10040 (30382:rank -> providerPubkey @ operator relay) to - // their outbox — but only when we actually hold their key. - maybePublishObserverProviderList(ctx, observer, providerPubkey, relays.first())?.let { - result["observer_10040"] = it - } - } - } - - if (benchSign) { - // Throwaway key — these cards are for timing only and never leave - // the process, so no real identity signs them. - val tempSigner = NostrSignerInternal(KeyPair()) - val cards = - rankedIds - .filter { rankOf(scores[it]) >= minRank } - .map { graph.pubkeyOf(it) to rankOf(scores[it]) } - val signStart = System.nanoTime() - val signed = signCards(cards, tempSigner) - val signMs = (System.nanoTime() - signStart) / 1_000_000 - val perSec = if (signMs > 0) signed * 1000L / signMs else 0 - System.err.println("[graperank] signed $signed kind:30382 cards in $signMs ms ($perSec/s, temp key, not published)") - result["bench_signed"] = signed - result["bench_sign_ms"] = signMs - } + result["provider_pubkey"] = providerPubkey + result["min_rank"] = minRank + result["cards_total"] = desiredCards.size + result["cards_signed"] = local.signed + result["cards_unchanged"] = local.unchanged + result["cards_retracted"] = local.retracted + result["cards_ms"] = cardsMs Output.emit(result) return 0 @@ -780,37 +735,201 @@ object GrapeRankCommand { } /** - * Build + sign one kind:30382 [ContactCardEvent] per (target, rank), fanned - * out across CPU cores (id-hash + Schnorr sign is CPU-bound). The signed - * events are discarded — this only exists to time card generation. Returns - * the number signed. + * `amy graperank publish [OBSERVER] [--relay URL[,URL…]] [--relay-concurrency N] [--timeout SECS]` + * + * Transport only: make the operator relay(s) converge to the local card set + * that `graperank score` persisted for OBSERVER (default: the active account). + * One NIP-77 up-only reconcile per relay over the provider service key's + * kind:30382 cards + kind:5 retractions — nothing is re-scored or re-signed, + * and a card the relay lost is restored. A relay that can't reconcile gets the + * full local set blast-published instead. Also refreshes the observer's + * kind:10040 provider pointer when we hold their key. */ - private suspend fun signCards( - cards: List>, - signer: NostrSigner, + private suspend fun publish( + dataDir: DataDir, + rest: Array, ): Int { - if (cards.isEmpty()) return 0 - val cores = Runtime.getRuntime().availableProcessors().coerceAtLeast(1) - val chunkSize = ((cards.size + cores - 1) / cores).coerceAtLeast(1) - return coroutineScope { - cards - .chunked(chunkSize) - .map { chunk -> - async(Dispatchers.Default) { - for ((target, rank) in chunk) { - ContactCardEvent.create( - targetUser = target, - signer = signer, - publicInitializer = { add(RankTag.assemble(rank)) }, + val args = Args(rest) + val observerArg = args.positionalOrNull(0) + val relayArg = args.flag("relay") + val relayConcurrency = args.intFlag("relay-concurrency", 4) + val idleTimeoutMs = args.longFlag("timeout", 30L) * 1000 + + Context.open(dataDir).use { ctx -> + ctx.prepare() + val observer = observerArg?.let { ctx.requireUserHex(it) } ?: ctx.identity.pubKeyHex + val opKeys = ctx.dataDir.operatorKeys() + val providerPubkey = opKeys.serviceKey(observer).pubKey.toHexKey() + + // Cards live on the operator's own relay(s); --relay overrides. + val relays = + relayArg + ?.split(",") + ?.mapNotNull { RelayUrlNormalizer.normalizeOrNull(it.trim()) } + ?.toSet() + ?.takeIf { it.isNotEmpty() } + ?: opKeys.operatorRelays() + if (relays.isEmpty()) { + return Output.error("no_relays", "no operator relay configured — run `amy graperank operator relay ` or pass --relay") + } + + val publisher = GrapeRankPublisher(ctx.store) { System.err.println(it) } + val sync = + publisher.syncToRelays( + client = ctx.client, + providerPubkey = providerPubkey, + relays = relays, + relayConcurrency = relayConcurrency, + idleTimeoutMs = idleTimeoutMs, + ) + + if (sync.cards == 0 && sync.deletions == 0) { + Output.emit( + linkedMapOf( + "observer" to observer, + "provider_pubkey" to providerPubkey, + "cards" to 0, + "note" to "no local cards for this observer — run `amy graperank score` first", + ), + ) + return 0 + } + + // Help the observer point clients at this provider: publish their + // kind:10040 (30382:rank -> providerPubkey @ operator relay) to + // their outbox — but only when we actually hold their key. + val observer10040 = maybePublishObserverProviderList(ctx, observer, providerPubkey, relays.first()) + + Output.emit( + linkedMapOf( + "observer" to observer, + "provider_pubkey" to providerPubkey, + "cards" to sync.cards, + "deletions" to sync.deletions, + "relays" to sync.perRelay.size, + "relays_ok" to sync.perRelay.count { it.ok }, + "relays_failed" to sync.perRelay.count { !it.ok }, + "uploaded" to sync.perRelay.sumOf { it.uploaded }, + "fallback_published" to sync.perRelay.sumOf { it.fallbackPublished }, + "per_relay" to + sync.perRelay.map { + linkedMapOf( + "relay" to it.relay.url, + "uploaded" to it.uploaded, + "fallback_published" to it.fallbackPublished, + "fallback_rejected" to it.fallbackRejected, + "error" to it.error, ) - } - chunk.size - } - }.awaitAll() - .sum() + }, + "observer_10040" to observer10040, + ), + ) + return 0 } } + /** + * `amy graperank rank USER [--provider PUBKEY] [--refresh] [--timeout SECS]` + * + * The consumer side of NIP-85: read the kind:30382 cards about USER and print + * one rank per provider (newest card each). Cache-first — a `graperank score` + * run on this machine already left its cards in the store — falling back to a + * relay drain on a miss or with `--refresh` (sources: the operator relays, the + * relays declared in the account's kind:10040, and the bootstrap set). + * `--provider` narrows to one provider key. + */ + private suspend fun rank( + dataDir: DataDir, + rest: Array, + ): Int { + val args = Args(rest) + val userArg = + args.positionalOrNull(0) + ?: return Output.error("bad_args", "usage: amy graperank rank USER [--provider PUBKEY] [--refresh] [--timeout SECS]") + val providerArg = args.flag("provider") + val refresh = args.bool("refresh") + val timeoutMs = args.longFlag("timeout", 8L) * 1000 + + Context.openOrAnonymous(dataDir).use { ctx -> + ctx.prepare() + val user = ctx.requireUserHex(userArg) + val provider = providerArg?.let { ctx.requireUserHex(it) } + val cardFilter = + Filter( + kinds = listOf(ContactCardEvent.KIND), + tags = mapOf("d" to listOf(user)), + authors = provider?.let { listOf(it) }, + ) + + suspend fun localCards(): List = ctx.store.query(cardFilter).filterIsInstance() + + var cards = localCards() + if (refresh || cards.isEmpty()) { + val relays = rankSourceRelays(ctx, provider) + if (relays.isNotEmpty()) { + ctx.drain(relays.associateWith { listOf(cardFilter.copy(limit = 50)) }, timeoutMs) + cards = localCards() + } + } + + // Newest card per provider key, strongest assertion first. + val newest = + cards + .groupBy { it.pubKey } + .mapNotNull { (_, list) -> list.maxByOrNull { it.createdAt } } + .sortedWith(compareByDescending { it.rank() ?: -1 }.thenByDescending { it.createdAt }) + + // A provider key this machine's operator master derived maps back to + // the observer whose subjective view the rank expresses. + val providerToObserver = + ctx.dataDir + .operatorKeys() + .providers() + .entries + .associate { (observer, rec) -> rec.providerPubKey to observer } + + Output.emit( + linkedMapOf( + "user" to user, + "found" to newest.isNotEmpty(), + "cards" to + newest.map { card -> + linkedMapOf( + "provider" to card.pubKey, + "rank" to card.rank(), + "observer" to providerToObserver[card.pubKey], + "created_at" to card.createdAt, + "event_id" to card.id, + ) + }, + ), + ) + return 0 + } + } + + /** + * Relays worth draining for someone's kind:30382 cards: the machine's own + * operator relay(s), every relay the account's kind:10040 declares for a + * 30382 service (narrowed to [provider] when given), and the bootstrap set. + */ + private suspend fun rankSourceRelays( + ctx: Context, + provider: HexKey?, + ): Set { + val declared = + if (!ctx.anonymous) { + providerListOf(ctx, ctx.identity.pubKeyHex) + ?.serviceProviders() + ?.filter { it.service.kind == ContactCardEvent.KIND && (provider == null || it.pubkey == provider) } + ?.map { it.relayUrl } + .orEmpty() + } else { + emptyList() + } + return ctx.dataDir.operatorKeys().operatorRelays() + declared + ctx.bootstrapRelays() + Constants.eventFinderRelays + } + /** * `amy graperank operator [status | relay … | providers]` * diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/experimental/graperank/GrapeRankPublisher.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/experimental/graperank/GrapeRankPublisher.kt index 1ccc26cea2..9ff48d83ef 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/experimental/graperank/GrapeRankPublisher.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/experimental/graperank/GrapeRankPublisher.kt @@ -22,6 +22,9 @@ 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.NegentropyStoreSync +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.publishAndConfirmDetailed import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl import com.vitorpamplona.quartz.nip01Core.signers.NostrSigner @@ -29,92 +32,165 @@ import com.vitorpamplona.quartz.nip01Core.store.IEventStore import com.vitorpamplona.quartz.nip09Deletions.DeletionEvent import com.vitorpamplona.quartz.nip85TrustedAssertions.users.ContactCardEvent import com.vitorpamplona.quartz.nip85TrustedAssertions.users.tags.RankTag +import com.vitorpamplona.quartz.utils.TimeUtils +import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.async import kotlinx.coroutines.awaitAll import kotlinx.coroutines.coroutineScope +import kotlin.coroutines.cancellation.CancellationException /** - * Publishes a set of GrapeRank scores as NIP-85 kind:30382 [ContactCardEvent] - * trusted assertions (one `rank` card per scored user), reconciled against what - * this provider key has already published so a repeat run only writes what moved. + * Persists a set of GrapeRank scores as NIP-85 kind:30382 [ContactCardEvent] + * trusted assertions in the local [IEventStore] — the durable, reusable form of a + * score run — and pushes that local card set out to the operator's relays on + * demand. The store is the source of truth; the two halves are separable: * - * Reconciliation, given the desired `(target, rank)` set the scorer produced: - * - **skip** a target whose stored card already carries the same rank string — - * re-signing an unchanged card would churn a new event id for no client benefit; - * - **upsert** a target whose rank changed (or that has no card yet), up to a - * publish limit; - * - **retract** every stored card whose target is no longer in the desired set - * (it fell below the caller's cutoff, or dropped out of the graph) with a NIP-09 - * kind:5 deletion, batched so the frame stays under the ~64KB event cap. - * - * Transport-agnostic like [GrapeRankCrawler]: it reads prior cards from an - * [IEventStore] and emits through an injected [publish] function (event + relays → - * per-relay ack), so the store/relay wiring stays in the application while the - * reconcile + card-construction logic is reusable (e.g. by the Android app). + * - [reconcileLocal] runs after every scoring pass. It diffs the desired + * `(target, rank)` set against the cards this provider key already holds in the + * store, signs + inserts only what moved (an unchanged rank is skipped so no new + * event id churns), and retracts every card whose target dropped out with a + * NIP-09 kind:5. The store applies kind:5 on insert, so a retracted card + * disappears locally while the deletion event remains as the durable tombstone + * to propagate. + * - [syncToRelays] is pure transport: a NIP-77 up-only reconcile of everything the + * provider key authored (kind:30382 cards + kind:5 retractions) against each + * relay, so the relay converges to the local set — nothing is re-signed, and a + * card the relay lost (or never had) is restored. A relay that can't reconcile + * gets the full local set blast-published instead (relays dedup by id). */ class GrapeRankPublisher( private val store: IEventStore, - private val publish: suspend (Event, Set) -> Map, + private val log: (String) -> Unit = {}, ) { - /** Outcome counts for one reconcile: what was written, retracted, and skipped. */ - class Result( - val published: Int, - val publishRejected: Int, - val deleted: Int, - val deleteRejected: Int, - val skippedUnchanged: Int, - /** Changed cards beyond [publishLimit] that were not upserted this run. */ - val truncated: Int, + /** Outcome of one [reconcileLocal]: what was signed, retracted, and left alone. */ + class LocalResult( + /** New or rank-changed cards signed and inserted into the store. */ + val signed: Int, + /** Stale cards retracted (kind:5) because their target left the desired set. */ + val retracted: Int, + /** Desired cards whose stored rank already matched — no new signature. */ + val unchanged: Int, ) /** * Reconcile the desired [scored] `(target, rank)` set (the caller has already - * applied any rank cutoff) against the cards [providerPubkey] previously - * published, then upsert the changes and retract the stale cards, all signed by - * [providerSigner]. At most [publishLimit] changed cards are upserted per run. + * applied any rank cutoff) into the local store, signed by [providerSigner]: + * upsert the changed cards, retract the stale ones, skip the unchanged rest. + * Every card and retraction is stamped [createdAt], so a replacement is + * strictly newer than what it displaces. */ - suspend fun reconcileAndPublish( + suspend fun reconcileLocal( providerSigner: NostrSigner, providerPubkey: HexKey, scored: List>, - relays: Set, - publishLimit: Int, - publishConcurrency: Int = PUBLISH_CONCURRENCY, - ): Result { - // Newest card per target this provider already published (read back from - // the store, which every published card was persisted to). + createdAt: Long = TimeUtils.now(), + ): LocalResult { val existing = existingCards(providerPubkey) - val publishableTargets = scored.mapTo(HashSet()) { it.first } + val desiredTargets = scored.mapTo(HashSet()) { it.first } - // Upsert publishable targets whose rank tag STRING would change (or that - // have no card yet). RankTag.assemble writes rank.toString(), so we diff - // that exact string — an unchanged score is skipped so clients only sync - // ranks that moved. + // Upsert targets whose rank tag STRING would change (or that have no card + // yet). RankTag.assemble writes rank.toString(), so we diff that exact + // string — an unchanged score never produces a new signature. val changed = scored.filter { (target, rank) -> existing[target]?.let(::rankTagValue) != rank.toString() } - val toUpsert = changed.take(publishLimit) - // Retract existing cards whose target is no longer publishable — it dropped - // out of the graph, or fell below the caller's cutoff. We won't leave a - // stale assertion standing. - val toDelete = existing.filterKeys { it !in publishableTargets }.values.toList() + // Retract cards whose target is no longer desired — it dropped out of the + // graph, or fell below the caller's cutoff. No stale assertion is left standing. + val toRetract = existing.filterKeys { it !in desiredTargets }.values.toList() - val (ok, rejected) = publishCards(providerSigner, toUpsert, relays, publishConcurrency) - val (deleted, deleteRejected) = publishDeletions(providerSigner, toDelete, relays) + val signed = signAndInsertCards(providerSigner, changed, createdAt) + val retracted = insertRetractions(providerSigner, toRetract, createdAt) - return Result( - published = ok, - publishRejected = rejected, - deleted = deleted, - deleteRejected = deleteRejected, - skippedUnchanged = scored.size - changed.size, - truncated = (changed.size - toUpsert.size).coerceAtLeast(0), + return LocalResult( + signed = signed, + retracted = retracted, + unchanged = scored.size - changed.size, ) } + /** Per-relay outcome of a [syncToRelays]. */ + class RelaySyncResult( + val relay: NormalizedRelayUrl, + /** Events the NIP-77 reconcile found missing on the relay and uploaded. */ + val uploaded: Int, + /** Events blast-published because the relay couldn't reconcile. */ + val fallbackPublished: Int, + val fallbackRejected: Int, + /** The negentropy failure that triggered the fallback, or null when it reconciled. */ + val error: String?, + ) { + /** True when the relay now converged to the local set by either path. */ + val ok: Boolean get() = error == null || (fallbackPublished > 0 && fallbackRejected == 0) + } + + /** Aggregate outcome of a [syncToRelays]. */ + class SyncResult( + /** kind:30382 cards this provider key holds locally (the set being mirrored). */ + val cards: Int, + /** kind:5 retraction events this provider key holds locally. */ + val deletions: Int, + val perRelay: List, + ) + /** - * The newest kind:30382 card [providerPubkey] published per target, read from - * the store (every card [publish] sends is persisted first, so on repeat runs - * this reflects what is already out there). + * Make every relay in [relays] converge to the local card set of + * [providerPubkey]: one NIP-77 up-only reconcile per relay over the provider's + * kind:30382 + kind:5 (nothing is downloaded — the store is the source of + * truth), falling back to blast-publishing the full local set when a relay + * can't reconcile. Best-effort per relay; a failure never aborts the others. + */ + suspend fun syncToRelays( + client: INostrClient, + providerPubkey: HexKey, + relays: Set, + relayConcurrency: Int = 4, + idleTimeoutMs: Long = 30_000L, + publishTimeoutSecs: Long = 15, + ): SyncResult { + val cards = store.query(Filter(kinds = listOf(ContactCardEvent.KIND), authors = listOf(providerPubkey))) + val deletions = store.query(Filter(kinds = listOf(DeletionEvent.KIND), authors = listOf(providerPubkey))) + + val filter = Filter(kinds = listOf(ContactCardEvent.KIND, DeletionEvent.KIND), authors = listOf(providerPubkey)) + val groups = + NegentropyStoreSync( + client = client, + store = store, + config = + NegentropyStoreSync.Config( + // Up-only: the relay must converge to the store, never the + // reverse — pulling a stale card back down would resurrect a + // locally-retracted assertion. The kind:5s ride the same + // filter, so deletions propagate as ordinary uploads. + down = false, + up = true, + syncDeletions = false, + // The paged fallback DOWNLOADS the filter — wrong direction + // for a push. Failed groups get [blastPublish] instead. + pageFallback = false, + concurrency = relayConcurrency, + idleTimeoutMs = idleTimeoutMs, + publishTimeoutSecs = publishTimeoutSecs, + ), + log = log, + ).sync(relays.associateWith { listOf(filter) }) + + val perRelay = + groups.map { g -> + if (g.error == null) { + RelaySyncResult(g.relay, uploaded = g.uploaded, fallbackPublished = 0, fallbackRejected = 0, error = null) + } else { + log("[graperank] ${g.relay.url}: negentropy failed (${g.error}) — publishing the full local set instead") + val (ok, rejected) = blastPublish(client, cards + deletions, g.relay, publishTimeoutSecs) + RelaySyncResult(g.relay, uploaded = g.uploaded, fallbackPublished = ok, fallbackRejected = rejected, error = g.error) + } + } + + return SyncResult(cards = cards.size, deletions = deletions.size, perRelay = perRelay) + } + + /** + * The newest kind:30382 card [providerPubkey] holds per target in the local + * store. Locally-retracted cards never show up — inserting their kind:5 + * removed them — so a target can be re-carded later without fighting a ghost. */ private suspend fun existingCards(providerPubkey: HexKey): Map = store @@ -129,7 +205,7 @@ class GrapeRankPublisher( /** * The raw `rank` tag value string on a card — exactly what a client diffs, so an * unchanged score never produces a new signature. Our cards carry only a `rank` - * tag (plus the d-tag target), so this one value decides whether a re-publish + * tag (plus the d-tag target), so this one value decides whether a re-sign * would differ. */ private fun rankTagValue(card: ContactCardEvent): String? = @@ -137,63 +213,112 @@ class GrapeRankPublisher( if (tag.size > 1 && tag[0] == RankTag.TAG_NAME) tag[1] else null } - /** Build + publish one kind:30382 card per (target, rank), bounded-concurrently. */ - private suspend fun publishCards( + /** + * Build + sign one kind:30382 card per (target, rank) — fanned out on + * [Dispatchers.Default], since id-hash + Schnorr sign is CPU-bound — and insert + * each into the store. Batched so a whole-network card set never materializes + * in memory at once. Returns how many the store accepted. + */ + private suspend fun signAndInsertCards( signer: NostrSigner, cards: List>, - relays: Set, - concurrency: Int, - ): Pair { - var published = 0 - var rejected = 0 - for (batch in cards.chunked(concurrency)) { - val acks = + createdAt: Long, + ): Int { + if (cards.isEmpty()) return 0 + var inserted = 0 + for (batch in cards.chunked(SIGN_BATCH)) { + val signedBatch = coroutineScope { batch - .map { (pubkey, rank) -> - async { - val card = - ContactCardEvent.create( - targetUser = pubkey, - signer = signer, - publicInitializer = { add(RankTag.assemble(rank)) }, - ) - publish(card, relays) + .map { (target, rank) -> + async(Dispatchers.Default) { + ContactCardEvent.create( + targetUser = target, + signer = signer, + createdAt = createdAt, + publicInitializer = { add(RankTag.assemble(rank)) }, + ) } }.awaitAll() } - for (ack in acks) { - if (ack.values.any { it }) published++ else rejected++ + for (card in signedBatch) { + if (insertQuietly(card)) inserted++ } } - return published to rejected + return inserted } /** * Retract stale cards with NIP-09 kind:5 deletions signed by [signer] (the same - * key that signed the cards). Batches [DELETE_PER_EVENT] addressable coordinates - * per deletion so the kind:5 frame stays under the ~64KB event cap; each carries - * the card's `a` tag (30382:provider:target), so re-publishing a newer version - * later isn't blocked. Returns (deleted, rejected) card counts. + * key that signed the cards), inserted into the store — which applies them, so + * the retracted cards vanish locally and only the tombstone remains to sync. + * Batches [DELETE_PER_EVENT] addressable coordinates per deletion so the kind:5 + * frame stays under the ~64KB event cap; each carries the card's `a` tag + * (30382:provider:target), so re-publishing a newer version later isn't blocked. + * Returns how many cards were retracted. */ - private suspend fun publishDeletions( + private suspend fun insertRetractions( signer: NostrSigner, cards: List, - relays: Set, - ): Pair { - if (cards.isEmpty()) return 0 to 0 - var deleted = 0 - var rejected = 0 + createdAt: Long, + ): Int { + if (cards.isEmpty()) return 0 + var retracted = 0 for (chunk in cards.chunked(DELETE_PER_EVENT)) { - val event = signer.sign(DeletionEvent.build(chunk)) - val ack = publish(event, relays) - if (ack.values.any { it }) deleted += chunk.size else rejected += chunk.size + val deletion = signer.sign(DeletionEvent.build(chunk, createdAt)) + if (insertQuietly(deletion)) retracted += chunk.size } - return deleted to rejected + return retracted + } + + /** + * Insert without letting a single-row failure abort the whole reconcile. The + * one expected rejection is the store's own deletion guard (a re-carded target + * whose kind:5 landed in the same second); anything else is logged. Returns + * whether the store accepted the event. + */ + private suspend fun insertQuietly(event: Event): Boolean = + try { + store.insert(event) + true + } catch (e: CancellationException) { + throw e + } catch (e: Exception) { + log("[graperank] store rejected ${event.kind}/${event.id.take(8)}: ${e.message}") + false + } + + /** + * Fallback for a relay without working NIP-77: publish every event of the local + * set individually (the relay dedups by id), bounded-concurrently. Returns + * (accepted, rejected) event counts. + */ + private suspend fun blastPublish( + client: INostrClient, + events: List, + relay: NormalizedRelayUrl, + publishTimeoutSecs: Long, + ): Pair { + var ok = 0 + var rejected = 0 + val relaySet = setOf(relay) + for (batch in events.chunked(PUBLISH_CONCURRENCY)) { + val acks = + coroutineScope { + batch.map { event -> async { client.publishAndConfirmDetailed(event, relaySet, publishTimeoutSecs) } }.awaitAll() + } + for (ack in acks) { + if (ack.values.any { it }) ok++ else rejected++ + } + } + return ok to rejected } companion object { - /** Concurrent card publishes when upserting. */ + /** Cards signed per batch — bounds live event objects, not parallelism. */ + const val SIGN_BATCH = 1024 + + /** Concurrent per-event publishes in the [blastPublish] fallback. */ const val PUBLISH_CONCURRENCY = 16 /** diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/experimental/graperank/GrapeRankPublisherTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/experimental/graperank/GrapeRankPublisherTest.kt new file mode 100644 index 0000000000..cbd3fbb6de --- /dev/null +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/experimental/graperank/GrapeRankPublisherTest.kt @@ -0,0 +1,128 @@ +/* + * 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.crypto.KeyPair +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerInternal +import com.vitorpamplona.quartz.nip01Core.store.IEventStore +import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore +import com.vitorpamplona.quartz.nip09Deletions.DeletionEvent +import com.vitorpamplona.quartz.nip85TrustedAssertions.users.ContactCardEvent +import kotlinx.coroutines.runBlocking +import kotlin.test.Test +import kotlin.test.assertEquals + +/** + * [GrapeRankPublisher.reconcileLocal] is what makes a score run durable: the + * local store must end up holding exactly the desired card set — changed ranks + * re-signed, unchanged ranks untouched (no event-id churn), dropped targets + * retracted via kind:5 (which the store applies on insert) with the tombstone + * kept for a later relay sync. + */ +class GrapeRankPublisherTest { + private fun hexKey(n: Int): String = n.toString(16).padStart(64, '0') + + private suspend fun cardsByTarget( + store: IEventStore, + provider: String, + ): Map = + store + .query(Filter(kinds = listOf(ContactCardEvent.KIND), authors = listOf(provider))) + .filterIsInstance() + .associate { it.aboutUser() to it.rank() } + + @Test + fun reconcileLocalUpsertsSkipsAndRetracts() = + runBlocking { + val store = EventStore(null) + val signer = NostrSignerInternal(KeyPair()) + val provider = signer.pubKey + val publisher = GrapeRankPublisher(store) + + val a = hexKey(0xA) + val b = hexKey(0xB) + val c = hexKey(0xC) + + // First run: every card is new. + val r1 = publisher.reconcileLocal(signer, provider, listOf(a to 50, b to 10), createdAt = 1_000L) + assertEquals(2, r1.signed) + assertEquals(0, r1.unchanged) + assertEquals(0, r1.retracted) + assertEquals(mapOf(a to 50, b to 10), cardsByTarget(store, provider)) + val firstIds = + store + .query(Filter(kinds = listOf(ContactCardEvent.KIND), authors = listOf(provider))) + .map { it.id } + .toSet() + + // Same ranks again: nothing is re-signed, no event id churns. + val r2 = publisher.reconcileLocal(signer, provider, listOf(a to 50, b to 10), createdAt = 2_000L) + assertEquals(0, r2.signed) + assertEquals(2, r2.unchanged) + assertEquals(0, r2.retracted) + val secondIds = + store + .query(Filter(kinds = listOf(ContactCardEvent.KIND), authors = listOf(provider))) + .map { it.id } + .toSet() + assertEquals(firstIds, secondIds) + + // a's rank moved, b dropped out, c is new: a + c signed, b retracted. + val r3 = publisher.reconcileLocal(signer, provider, listOf(a to 60, c to 5), createdAt = 3_000L) + assertEquals(2, r3.signed) + assertEquals(0, r3.unchanged) + assertEquals(1, r3.retracted) + + // The store converged to exactly the desired set — b's card is gone… + assertEquals(mapOf(a to 60, c to 5), cardsByTarget(store, provider)) + + // …but its kind:5 tombstone remains, ready to sync to relays. + val deletions = store.query(Filter(kinds = listOf(DeletionEvent.KIND), authors = listOf(provider))) + assertEquals(1, deletions.size) + + store.close() + } + + @Test + fun retractedTargetCanBeReCardedLater() = + runBlocking { + val store = EventStore(null) + val signer = NostrSignerInternal(KeyPair()) + val provider = signer.pubKey + val publisher = GrapeRankPublisher(store) + + val a = hexKey(0xA) + + publisher.reconcileLocal(signer, provider, listOf(a to 40), createdAt = 1_000L) + publisher.reconcileLocal(signer, provider, emptyList(), createdAt = 2_000L) + assertEquals(emptyMap(), cardsByTarget(store, provider)) + + // A NEWER card outranks the older kind:5 (NIP-09 deletions only cover + // versions up to their created_at), so the target comes back cleanly. + val r = publisher.reconcileLocal(signer, provider, listOf(a to 45), createdAt = 3_000L) + assertEquals(1, r.signed) + assertEquals(mapOf(a to 45), cardsByTarget(store, provider)) + + store.close() + } +}