mirror of
https://github.com/vitorpamplona/amethyst.git
synced 2026-08-09 08:04:45 +00:00
feat(cli): graperank scores always persist locally; add publish + rank verbs
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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013WSzVX9RoUxyV3nT3dfc56
This commit is contained in:
Claude
+27
-16
@@ -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 <url>… \| 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 <url>… \| 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 <observer> --publish # sign with the observer's service key
|
||||
amy graperank <observer> # crawl + score + persist cards locally
|
||||
amy graperank publish <observer> # 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)
|
||||
|
||||
|
||||
+1
-1
@@ -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 | 🆕 | |
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 <url>` 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<Pair<HexKey, Int>>,
|
||||
signer: NostrSigner,
|
||||
private suspend fun publish(
|
||||
dataDir: DataDir,
|
||||
rest: Array<String>,
|
||||
): 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 <url>` 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<String, Any?>(
|
||||
"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<String, Any?>(
|
||||
"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<String, Any?>(
|
||||
"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<String>,
|
||||
): 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<ContactCardEvent> = ctx.store.query<Event>(cardFilter).filterIsInstance<ContactCardEvent>()
|
||||
|
||||
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<ContactCardEvent> { 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<String, Any?>(
|
||||
"user" to user,
|
||||
"found" to newest.isNotEmpty(),
|
||||
"cards" to
|
||||
newest.map { card ->
|
||||
linkedMapOf<String, Any?>(
|
||||
"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<NormalizedRelayUrl> {
|
||||
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 <url>… | providers]`
|
||||
*
|
||||
|
||||
+220
-95
@@ -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<NormalizedRelayUrl>) -> Map<NormalizedRelayUrl, Boolean>,
|
||||
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<Pair<HexKey, Int>>,
|
||||
relays: Set<NormalizedRelayUrl>,
|
||||
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<RelaySyncResult>,
|
||||
)
|
||||
|
||||
/**
|
||||
* 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<NormalizedRelayUrl>,
|
||||
relayConcurrency: Int = 4,
|
||||
idleTimeoutMs: Long = 30_000L,
|
||||
publishTimeoutSecs: Long = 15,
|
||||
): SyncResult {
|
||||
val cards = store.query<Event>(Filter(kinds = listOf(ContactCardEvent.KIND), authors = listOf(providerPubkey)))
|
||||
val deletions = store.query<Event>(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<HexKey, ContactCardEvent> =
|
||||
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<Pair<HexKey, Int>>,
|
||||
relays: Set<NormalizedRelayUrl>,
|
||||
concurrency: Int,
|
||||
): Pair<Int, Int> {
|
||||
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<ContactCardEvent>,
|
||||
relays: Set<NormalizedRelayUrl>,
|
||||
): Pair<Int, Int> {
|
||||
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<Event>,
|
||||
relay: NormalizedRelayUrl,
|
||||
publishTimeoutSecs: Long,
|
||||
): Pair<Int, Int> {
|
||||
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
|
||||
|
||||
/**
|
||||
|
||||
+128
@@ -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<String, Int?> =
|
||||
store
|
||||
.query<Event>(Filter(kinds = listOf(ContactCardEvent.KIND), authors = listOf(provider)))
|
||||
.filterIsInstance<ContactCardEvent>()
|
||||
.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<Event>(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<Event>(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<Event>(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()
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user