Merge pull request #3494 from vitorpamplona/claude/graperank-wot-cli-qreg2a

GrapeRank web-of-trust for amy: sync → score → publish (NIP-85)
This commit is contained in:
Vitor Pamplona
2026-07-07 21:02:28 -04:00
committed by GitHub
31 changed files with 4680 additions and 125 deletions
+35
View File
@@ -385,6 +385,41 @@ HTTP endpoint. Reuses quartz's `Nip86Client` and the shared `Nip86Retriever`
| `amy notes feed [--author USER \| --following] [--limit N]` | Read recent kind:1 notes (yours, one user's, or your follow set). |
| `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 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)
Ranks are published 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:
```
serviceKey(observer) = sha256(masterPriv ‖ "graperank-provider:" ‖ observerHex)
```
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:
```bash
amy graperank operator relay wss://relay.example.com # where all cards live
amy graperank <observer> --publish # sign with the observer's service key
```
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.
### Direct messages (NIP-17)
+1
View File
@@ -60,6 +60,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-72 communities | 🆕 | |
| NIP-78 app-specific data (settings sync) | 🆕 | |
| Long-form (NIP-23) publish / read | 🆕 | |
@@ -0,0 +1,134 @@
# GrapeRank score parity with NosFabrica Brainstorm
Goal: `amy graperank` should output scores **numerically very close** to
NosFabrica's Brainstorm service, the reference GrapeRank implementation.
Sources analysed:
- `NosFabrica/brainstorm_graperank_algorithm` — the Java scoring worker.
- `NosFabrica/brainstorm_server` — the Python orchestration server.
## How Brainstorm builds its service
A four-stage pipeline:
1. **Ingest.** `app/nostr_event_transferer/nostr_event_transferer.py` copies raw
social-graph events — **kinds 0, 3, 10000, 1984** (profiles, follows, mutes,
reports) — from a strfry relay into the server. Same four kinds we crawl.
2. **Graph.** Events land in **Neo4j** as a directed graph of follow / mute /
report edges between pubkeys. Redis + Postgres back the job queue and config.
3. **Score.** The Java worker (`grape/GrapeRankAlgorithm.java`) runs GrapeRank
from an observer, producing a **`ScoreCard`** per user
(`rank/ScoreCard.java`): `observer, observee, hops, averageScore, input,
confidence, influence, verified, trustedFollowers, trustedReporters`.
**There is no `rank` field — the trust value is `influence` ∈ [0,1].**
4. **Serve / publish.** Presets are tunable per deployment
(`DEFAULT` / `PERMISSIVE` / `RESTRICTIVE`, `graperank_preset` table, validated
by `GrapeRankPresetParams`). Java `GrapeRankParams` mirrors the Python model
field-for-field; the README states Python is the source of truth and both
repos must stay in sync.
## The algorithm (their `grape/GrapeRankAlgorithm.java`)
```
rigority = -log(rigor)
confidence(sumWeights) = 1 - exp(-sumWeights * rigority) # weight -> confidence
per edge: weight = edgeConfidence * influenceOfRater * attenuationFactor
wxr = weight * edgeRating
averageScore = sumWxR / sumWeights (0 if sumWeights == 0)
influence = max(averageScore * confidence(sumWeights), 0)
```
- Observer seeded at `influence = 1.0` (fixed authority).
- Non-observers seeded by hop distance, then **iterated until every user's
influence delta < 0.0001** (`loopBreakDelta`). Seeding only affects the
starting guess; attenuation < 1 makes the update a contraction, so the fixed
point is unique.
- The rater weight uses the rater's **`influence`**, and
`influence = max(weightToConfidence(sumW) * sumWR/sumW, 0)`.
## Side-by-side: Brainstorm DEFAULT vs `commons/wot`
`Constants.java` `DEFAULT_PARAMS` (== the Pydantic `GrapeRankPresetParams`
DEFAULT) against our `GrapeRankParams` defaults:
| Brainstorm field | value | our field | value | match |
|---|---|---|---|---|
| `attenuationFactor` | 0.85 | `attenuation` | 0.85 | ✅ |
| `rigor` | 0.5 | `rigor` | 0.5 | ✅ |
| `followRating` | 1.0 | `FOLLOW.rating` | 1.0 | ✅ |
| `muteRating` | -0.1 | `MUTE.rating` | -0.1 | ✅ |
| `reportRating` | -0.1 | `REPORT.rating` | -0.1 | ✅ |
| `followConfidenceOfObserver` | 0.5 | `directFollowConfidence` | 0.5 | ✅ |
| `followConfidence` | 0.03 | `indirectFollowConfidence` | 0.03 | ✅ |
| `muteConfidence` | 0.5 | `muteConfidence` | 0.5 | ✅ |
| `reportConfidence` | 0.5 | `reportConfidence` | 0.5 | ✅ |
| `loopBreakDelta` | 0.0001 | `convergence` | 0.0001 | ✅ |
The three `verified*InfluenceCutoff`s (followers 0.02, reporters 0.1,
muters 0.01) only flag a derived `verified` boolean; they do **not** affect the
score.
**Conclusion: our formula is identical and every scoring parameter matches
DEFAULT.** Our `score` *is* their `influence`
(`max(weightToConfidence(sumW) * sumWR/sumW, 0)`), propagated as the rater
weight — the exact same quantity. On the same input graph the two produce the
same influence to floating-point precision. Our published `rank = round(score *
100)` is a presentation choice on top of that influence (their `ScoreCard`
exposes `influence` as a raw float via the API).
## Where divergence can still come from — and why it's small
It is **data**, not math:
1. **Graph completeness.** Brainstorm ingests the whole strfry graph into Neo4j;
we crawl outward from the observer via the outbox model. **This matters less
than it seems:** a mute/report contributes `confidence * influenceOfRater *
attenuation`, so a signal from a user with **zero influence** (someone outside
the observer's trust graph) contributes **zero**. Only follows/mutes/reports
authored by users *inside* the follow graph move a score — and those are
exactly the users our crawl discovers and whose kind 3/10000/1984 we fetch.
So the effective scoring input is the same, as long as the crawl actually
checks every discovered user's outbox — which it now does exhaustively (no
user cap, retrying an unreachable outbox a few times).
2. **Fringe users / crawl gaps.** A relay timeout that drops a contact list
removes edges and shifts nearby scores. The injector mitigates this with a
two-stage model mirroring the app's `pickRelaysToLoadUsers`, plus a
completeness loop that retries until every user's outbox has been checked:
- **Relay-list discovery** (kind:10002) queries the account's relays +
bootstrap + event-finder + **indexer relays** (purplepag.es, coracle, …).
Indexers aggregate kind:10002 (and kind:0) for the whole network, so this is
where a stranger's outbox is found — the biggest completeness lever.
- **Content** (kind:3/10000/1984/0) is fetched from each user's **own outbox**
write relays, with harvested **relay hints** (from the `p`-tag hints in
contact lists we crawl) and general-purpose relays as a best-effort fallback
when the outbox is unknown/down. **Indexers are not used for content** — they
don't serve those kinds; kind:3/mutes/reports live only on the user's outbox.
The crawl loops round by round, retrying any member whose contact list still
didn't arrive (a few times), until every discovered user's outbox
has been checked. Remaining mitigation lever: a generous `--timeout`.
3. **Convergence precision.** Both stop at delta 0.0001; residual error is
< ~0.0001 in influence ⇒ < ~0.01 rank points ⇒ identical integer `rank`.
4. **Seeding.** Their hop-distance seed vs our zero seed — same fixed point, no
effect on the result.
## Recommendations
- **Keep the current DEFAULT params** — they are byte-for-byte the Brainstorm
DEFAULT preset. No change needed for parity.
- **The crawl is exhaustive by default** (no user cap; every reachable user's
outbox is checked, unreachable outboxes retried a few times). An
incomplete crawl is the single biggest source of drift, so avoid capping it.
- **Optional, for fuller parity (not required for close scores):**
- Add `--preset default|permissive|restrictive`. DEFAULT is confirmed; the
PERMISSIVE / RESTRICTIVE numbers are DB-seeded in `brainstorm_server` (an
alembic seed migration) and were not extractable from the public tree —
pull them from a running instance before hard-coding.
- Optionally expose `influence` as a raw float alongside `rank` in `--json`,
and compute the `verified` flag from the cutoffs, to mirror their
`ScoreCard` shape for interop diffing.
## Verification idea
Point `amy graperank <observer> --offline` at a store seeded from the same
strfry snapshot Brainstorm ingested, and diff our `score` against their
`ScoreCard.influence` for the same observer. Expect agreement to ~1e-4.
@@ -234,6 +234,22 @@ class DataDir(
val groupsDir = File(marmotDir, "groups")
val keyPackageBundleFile = File(marmotDir, "keypackages.bundle")
/**
* SQLite event-store DB file, a sibling of [eventsDir] under
* `<root>/shared/`. Used when the store backend is SQLite (the
* default — see [StoreFactory]); the FS backend uses [eventsDir]
* instead. Kept alongside the FS store so switching backends never
* clobbers the other's data.
*/
val eventsDbFile: File = File(eventsDir.parentFile ?: root, "events.db")
/**
* Machine-level operator keys for GrapeRank trusted-assertion publishing,
* rooted at `~/.amy/operator/` (the account root's parent) so a single
* operator master is shared across accounts. See [OperatorKeys].
*/
fun operatorKeys(): OperatorKeys = OperatorKeys(root.parentFile ?: root, secrets)
init {
SecureFileIO.secureMkdirs(root)
// The accountless dir only ever exposes the shared event store; don't
@@ -38,12 +38,14 @@ import com.vitorpamplona.quartz.marmot.mip03GroupMessages.GroupEvent
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.core.HexKey
import com.vitorpamplona.quartz.nip01Core.core.hexToByteArray
import com.vitorpamplona.quartz.nip01Core.crypto.verify
import com.vitorpamplona.quartz.nip01Core.jackson.JacksonMapper
import com.vitorpamplona.quartz.nip01Core.metadata.MetadataEvent
import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.AdaptiveRelayLimiter
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.DrainFailure
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.classifyDrainFailure
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.fetchAllPagesFromPool
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.publishAndConfirmDetailed
import com.vitorpamplona.quartz.nip01Core.relay.client.auth.RelayAuthenticator
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.newSubId
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.CachingEventDecoder
@@ -55,7 +57,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.sockets.okhttp.TcpNoDelaySocketF
import com.vitorpamplona.quartz.nip01Core.signers.NostrSigner
import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerInternal
import com.vitorpamplona.quartz.nip01Core.store.IEventStore
import com.vitorpamplona.quartz.nip01Core.store.fs.FsEventStore
import com.vitorpamplona.quartz.nip01Core.store.verifyAndInsert
import com.vitorpamplona.quartz.nip02FollowList.ContactListEvent
import com.vitorpamplona.quartz.nip17Dm.settings.ChatMessageRelayListEvent
import com.vitorpamplona.quartz.nip46RemoteSigner.signer.NostrSignerRemote
@@ -78,7 +80,9 @@ import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.launch
import kotlinx.coroutines.selects.select
import kotlinx.coroutines.withTimeoutOrNull
import okhttp3.Dispatcher
import okhttp3.OkHttpClient
import java.util.concurrent.TimeUnit
/**
* Per-invocation wiring. Each CLI run constructs a Context, does its work,
@@ -98,9 +102,10 @@ import okhttp3.OkHttpClient
* Every Nostr event Amy observes — whether received from a relay
* subscription, unwrapped from a NIP-59 gift wrap, or generated locally
* before publish — is verified (NIP-01 signature + id check via
* [Event.verify]) and persisted to the file-backed [IEventStore] at
* `<data-dir>/events-store/`. Malformed events are dropped before
* reaching command code.
* [Event.verify]) and persisted to the shared [IEventStore] under
* `<data-dir>/shared/` (a SQLite DB by default, or the FS tree when
* `AMY_STORE=fs` — see [StoreFactory]). Malformed events are dropped
* before reaching command code.
*
* This makes [store] the authoritative cache of everything Amy has ever
* seen: profile metadata, relay lists, contact lists, gift wraps,
@@ -124,7 +129,31 @@ class Context(
*/
val anonymous: Boolean = false,
) : AutoCloseable {
private val okhttp = OkHttpClient.Builder().socketFactory(TcpNoDelaySocketFactory).build()
private val okhttp =
OkHttpClient
.Builder()
.socketFactory(TcpNoDelaySocketFactory)
// The crawl opens WebSockets to thousands of relays. Each WS-upgrade
// handshake is an async call through OkHttp's shared Dispatcher, whose
// default cap (maxRequests=64) throttles the connection ramp — worse,
// a dead relay holds a slot for the whole connectTimeout, starving live
// relays queued behind it. Widen the dispatcher so handshakes fan out,
// and keep connectTimeout tight-ish so an unreachable relay frees its
// slot fast. This is orthogonal to REQ concurrency (that runs on
// already-open sockets, bounded by AdaptiveRelayLimiter), so it can't
// trip a relay's REQ rate-limit — it only speeds connection setup. The
// executor thread pool is unbounded on demand, so raising maxRequests
// just lets more of those short-lived handshakes proceed at once. 7s
// (not 5s): a 5s cap struck too many merely-busy relays as connect
// failures — the crawl treats a connect *timeout* as retryable anyway,
// but the extra headroom lets slow-but-alive relays finish the handshake.
.connectTimeout(7, TimeUnit.SECONDS)
.dispatcher(
Dispatcher().apply {
maxRequests = 256
maxRequestsPerHost = 16
},
).build()
val client: NostrClient =
NostrClient(
@@ -154,6 +183,44 @@ class Context(
)
} ?: NostrSignerInternal(identity.keyPair())
/**
* Client-wide tally of relay feedback — NOTICE frames, CLOSED reasons
* (auth-required / rate-limited / restricted / …), and NIP-42 AUTH
* challenges — so a failed REQ can be explained instead of guessed at.
* Registered on [client] for the life of this run.
*/
val relayDiagnostics: RelayDiagnostics = RelayDiagnostics().also { client.addConnectionListener(it) }
/**
* Adaptive per-relay concurrent-subscription cap. Starts every relay
* generous (100) and demotes only the ones that complain about concurrency
* (100 → 20 → 10), driven straight off the NOTICE/CLOSED frames it observes
* as a connection listener. [drain]'s `gatePerRelay` path holds a relay's
* permit for the life of that relay's subscription, so we never exceed the
* cap the relay itself asked for. Idle for commands that don't opt in.
*/
val relayLimiter: AdaptiveRelayLimiter = AdaptiveRelayLimiter().also { client.addConnectionListener(it) }
/**
* NIP-42 responder: answers a relay's AUTH challenge by signing with the
* account key, so auth-gated relays serve our reads instead of CLOSing the
* subscription. Constructing it registers its own listener on [client].
* Only a local key auto-signs — a remote bunker signer is skipped, since a
* per-relay remote round-trip during a crawl would stall it (and signing an
* auth event with any key still unlocks relays that just want *some* auth).
*/
private val relayAuth: RelayAuthenticator =
RelayAuthenticator(
client = client,
signWithAllLoggedInUsers = { _, template ->
if (signer is NostrSignerInternal) {
runCatching { listOf(signer.sign(template)) }.getOrElse { emptyList() }
} else {
emptyList()
}
},
)
/**
* NIP-05 resolver for turning `alice@damus.io`-style identifiers into pubkeys.
* Uses the same OkHttp instance as the WebSocket client so we share connection
@@ -174,23 +241,13 @@ class Context(
private val messageStore by lazy { FileMarmotMessageStore(dataDir.groupsDir) }
/**
* Filesystem-backed Nostr event store, rooted at [DataDir.eventsDir].
* Lazy so commands that don't touch persistent event state pay zero
* open cost (no `.lock` file, no seed allocation). Closed by
* [close] when this Context shuts down.
*
* Files are written pretty-printed (not the compact NIP-01 canonical
* form) so `cat`, `jq`, `git diff` are useful out of the box —
* humans inspect these files. Verification always re-canonicalises,
* so the stored bytes never feed back into a signature check.
* Shared Nostr event store for this run, opened via [StoreFactory]
* (SQLite by default, or the FS tree when `AMY_STORE=fs`). Lazy so
* commands that don't touch persistent event state pay zero open cost
* (no DB file / `.lock`, no seed allocation). Closed by [close] when
* this Context shuts down.
*/
private val storeDelegate: Lazy<IEventStore> =
lazy {
FsEventStore(
root = dataDir.eventsDir.toPath(),
eventToJson = JacksonMapper::toJsonPretty,
)
}
private val storeDelegate: Lazy<IEventStore> = lazy { StoreFactory.open(dataDir) }
val store: IEventStore by storeDelegate
/** Fully-wired manager. Call [prepare] once before use to load persisted state. */
@@ -424,15 +481,26 @@ class Context(
* Subscribe to the given filters across the given relays, drain all events
* until either every relay has sent EOSE or the timeout elapses, and
* return them. Used for one-shot catch-up queries — not live subscriptions.
*
* When [deadOut] is provided, every relay that reported it could not be
* connected to (`onCannotConnect`) is added to it, so callers can prune
* proven-dead relays from future routing instead of paying the full
* [timeoutMs] on them again. Slow-but-connected relays are NOT reported —
* only hard connect failures, so a temporarily-busy relay isn't discarded.
*/
suspend fun drain(
filters: Map<NormalizedRelayUrl, List<Filter>>,
timeoutMs: Long = 8_000,
diagnoseSlow: Boolean = false,
deadOut: MutableMap<NormalizedRelayUrl, DrainFailure>? = null,
): List<Pair<NormalizedRelayUrl, Event>> {
if (filters.isEmpty()) return emptyList()
val eventChannel = Channel<Pair<NormalizedRelayUrl, Event>>(UNLIMITED)
val doneChannel = Channel<NormalizedRelayUrl>(UNLIMITED)
// Carries the terminal reason per relay so a timeout can distinguish a slow
// relay (never terminal) from a connect failure / CLOSED.
val doneChannel = Channel<Pair<NormalizedRelayUrl, String>>(UNLIMITED)
val remaining = filters.keys.toMutableSet()
val doneReasons = HashMap<NormalizedRelayUrl, String>()
val subId = newSubId()
val listener =
object : SubscriptionListener {
@@ -449,7 +517,7 @@ class Context(
relay: NormalizedRelayUrl,
forFilters: List<Filter>?,
) {
doneChannel.trySend(relay)
doneChannel.trySend(relay to "eose")
}
override fun onClosed(
@@ -457,7 +525,7 @@ class Context(
relay: NormalizedRelayUrl,
forFilters: List<Filter>?,
) {
doneChannel.trySend(relay)
doneChannel.trySend(relay to "closed:$message")
}
override fun onCannotConnect(
@@ -465,37 +533,75 @@ class Context(
message: String,
forFilters: List<Filter>?,
) {
doneChannel.trySend(relay)
doneChannel.trySend(relay to "cannot:$message")
}
}
val collected = mutableListOf<Pair<NormalizedRelayUrl, Event>>()
try {
client.subscribe(subId, filters, listener)
withTimeoutOrNull(timeoutMs) {
while (remaining.isNotEmpty()) {
select {
eventChannel.onReceive { pair ->
if (verifyAndStore(pair.second)) collected.add(pair)
val completed =
withTimeoutOrNull(timeoutMs) {
while (remaining.isNotEmpty()) {
select {
eventChannel.onReceive { pair ->
if (verifyAndStore(pair.second)) collected.add(pair)
}
doneChannel.onReceive { (relay, reason) ->
remaining.remove(relay)
doneReasons[relay] = reason
}
}
doneChannel.onReceive { r -> remaining.remove(r) }
}
// Drain any events that landed after EOSE but before cancel
while (true) {
val r = eventChannel.tryReceive()
if (!r.isSuccess) break
val pair = r.getOrThrow()
if (verifyAndStore(pair.second)) collected.add(pair)
}
true
}
// Drain any events that landed after EOSE but before cancel
while (true) {
val r = eventChannel.tryReceive()
if (!r.isSuccess) break
val pair = r.getOrThrow()
if (verifyAndStore(pair.second)) collected.add(pair)
}
if (diagnoseSlow && completed == null && remaining.isNotEmpty()) {
logSlowDrain(timeoutMs, remaining, doneReasons, collected)
}
} finally {
client.unsubscribe(subId)
eventChannel.close()
doneChannel.close()
}
deadOut?.let { out ->
for ((relay, reason) in doneReasons) {
classifyDrainFailure(reason)?.let { out[relay] = it }
}
}
return collected
}
/**
* On a [drain] timeout, report which relays stalled and why — a relay that
* never sent EOSE (slow, possibly still streaming) vs one that couldn't be
* reached (CANNOT-CONNECT, which points at our side / the network) vs one
* that CLOSED the sub. Includes how many events each slow relay did send, so
* "relay is slow" and "we never connected" are easy to tell apart.
*/
private fun logSlowDrain(
timeoutMs: Long,
stalled: Set<NormalizedRelayUrl>,
doneReasons: Map<NormalizedRelayUrl, String>,
collected: List<Pair<NormalizedRelayUrl, Event>>,
) {
val eventsPer = collected.groupingBy { it.first }.eachCount()
val cannot = doneReasons.filterValues { it.startsWith("cannot") }
val closed = doneReasons.filterValues { it.startsWith("closed") }
val slowDetail = stalled.take(12).joinToString(", ") { "${it.url}(${eventsPer[it] ?: 0}ev)" }
val cannotDetail = cannot.entries.take(8).joinToString(", ") { "${it.key.url}=${it.value.removePrefix("cannot:").take(40)}" }
System.err.println(
"[drain] timeout ${timeoutMs}ms: ${stalled.size} slow(no EOSE), ${cannot.size} cannot-connect, ${closed.size} closed" +
(if (slowDetail.isNotEmpty()) " | slow: $slowDetail" else "") +
(if (cannotDetail.isNotEmpty()) " | cannot: $cannotDetail" else ""),
)
}
/**
* Like [drain], but paginates every relay to completion via
* [fetchAllPagesFromPool] instead of stopping at the first EOSE — so a query
@@ -595,26 +701,17 @@ class Context(
}
/**
* Verify [event]'s NIP-01 id+signature and, if valid, persist it
* to [store]. Returns `true` when the event was accepted (and
* therefore should be surfaced to callers). Persistence failures
* (I/O errors, full disk) are logged but do not propagate.
* Verify [event]'s NIP-01 id+signature and, if valid, persist it to [store].
* Returns `true` when the event was accepted (and therefore should be surfaced
* to callers). Persistence failures (I/O errors, full disk) are logged but do
* not propagate; a UNIQUE-constraint rejection is normal and swallowed quietly.
*
* Every event-arrival path in the CLI funnels through this method
* so that [store] is the authoritative cache of what Amy has seen.
* Every event-arrival path in the CLI funnels through this so that [store] is
* the authoritative cache of what Amy has seen. Delegates to the shared quartz
* [verifyAndInsert] sink so the CLI and the GrapeRank crawler apply the exact
* same verify-then-store policy.
*/
suspend fun verifyAndStore(event: Event): Boolean {
if (!event.verify()) {
System.err.println("[cli] dropped event ${event.id.take(8)} kind=${event.kind} — bad signature")
return false
}
try {
store.insert(event)
} catch (t: Throwable) {
System.err.println("[cli] store insert failed for ${event.id.take(8)}: ${t.message}")
}
return true
}
suspend fun verifyAndStore(event: Event): Boolean = store.verifyAndInsert(event)
// ------------------------------------------------------------------
// Cache-first reads from [store]
@@ -38,6 +38,7 @@ import com.vitorpamplona.amethyst.cli.commands.FilterCommand
import com.vitorpamplona.amethyst.cli.commands.FollowCommand
import com.vitorpamplona.amethyst.cli.commands.GiftCommands
import com.vitorpamplona.amethyst.cli.commands.GitCommands
import com.vitorpamplona.amethyst.cli.commands.GrapeRankCommand
import com.vitorpamplona.amethyst.cli.commands.GroupCommands
import com.vitorpamplona.amethyst.cli.commands.InitCommands
import com.vitorpamplona.amethyst.cli.commands.KeyCommands
@@ -72,6 +73,8 @@ import com.vitorpamplona.amethyst.cli.commands.cashu.CashuCommands
import com.vitorpamplona.amethyst.cli.commands.cashu.CashuMintCommands
import com.vitorpamplona.amethyst.cli.commands.route
import com.vitorpamplona.amethyst.cli.secrets.SecretStore
import com.vitorpamplona.quartz.utils.Log
import com.vitorpamplona.quartz.utils.LogLevel
import kotlinx.coroutines.runBlocking
import kotlin.system.exitProcess
@@ -105,6 +108,12 @@ fun main(argv: Array<String>) {
// braces guard for invocations that bypass the launcher scripts.
System.setProperty("java.awt.headless", "true")
// Quiet quartz's internal DEBUG chatter (relay auth, MLS restore, URL
// rejection, throttle notices) by default so it doesn't drown a command's
// own output; --verbose / -v restores full DEBUG. Set before dispatch so
// even startup logging is gated.
Log.minLevel = if (argv.any { it == "--verbose" || it == "-v" }) LogLevel.DEBUG else LogLevel.WARN
// Set output mode before dispatch so even argument-parsing errors
// honour --json.
if (argv.any { it == "--json" || it == "--json=true" }) {
@@ -161,6 +170,7 @@ private suspend fun dispatch(argv: Array<String>): Int {
GlobalFlag.SECRET_BACKEND -> secretBackendFlag = consumed.value
GlobalFlag.PASSPHRASE_FILE -> passphraseFileFlag = consumed.value
GlobalFlag.JSON -> Output.mode = Output.Mode.JSON
GlobalFlag.VERBOSE -> Unit // level already applied in main(); just strip it here
null -> filteredArgs.add(a)
}
i += consumed.tokensConsumed
@@ -254,6 +264,7 @@ private suspend fun dispatch(argv: Array<String>): Int {
"store" -> StoreCommands.dispatch(dataDir, tail)
"follow" -> FollowCommand.follow(dataDir, tail)
"unfollow" -> FollowCommand.unfollow(dataDir, tail)
"graperank" -> GrapeRankCommand.dispatch(dataDir, tail)
"search" -> SearchCommand.dispatch(dataDir, tail)
"zap" -> ZapCommand.dispatch(dataDir, tail)
"offer" -> OfferCommands.dispatch(dataDir, tail)
@@ -305,11 +316,13 @@ private suspend fun marmotDispatch(
private enum class GlobalFlag(
val long: String,
val takesValue: Boolean = true,
val short: String? = null,
) {
ACCOUNT("--account"),
SECRET_BACKEND("--secret-backend"),
PASSPHRASE_FILE("--passphrase-file"),
JSON("--json", takesValue = false),
VERBOSE("--verbose", takesValue = false, short = "-v"),
}
private data class ConsumedFlag(
@@ -328,7 +341,7 @@ private fun extractGlobalFlag(
idx: Int,
): Pair<GlobalFlag?, ConsumedFlag> {
for (flag in GlobalFlag.values()) {
if (token == flag.long) {
if (token == flag.long || token == flag.short) {
return if (flag.takesValue) {
flag to ConsumedFlag(argv.getOrNull(idx + 1), 2)
} else {
@@ -353,13 +366,15 @@ private fun printUsage() {
| [--secret-backend auto|keychain|ncryptsec|plaintext]
| [--passphrase-file PATH]
| [--json]
| [--verbose|-v]
| <cmd> [args...]
|
|Account selection:
| All state lives under ~/.amy/. Per-account directories
| ~/.amy/<account>/ hold identity, cursors, MLS state, and
| aliases; every observed Nostr event lands in the shared
| ~/.amy/shared/events-store/. ACCOUNT must match
| store under ~/.amy/shared/ (a SQLite `events.db` by default, or
| the `events-store/` tree when AMY_STORE=fs). ACCOUNT must match
| [a-zA-Z0-9_-]{1,64} (no spaces, no slashes).
|
| Resolution order:
@@ -575,6 +590,31 @@ private fun printUsage() {
| unfollow USER [--timeout SECS] remove USER from your contact list
| (USER: npub|nprofile|hex|name@domain)
|
|Web of Trust (GrapeRank):
| graperank [OBSERVER] compute subjective trust scores (0..1) for every
| [--limit N] [--min-score X] user reachable in the follow/mute/report graph.
| [--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 logs slow/failed relays on timeout).
| [--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.
| graperank operator [status|relay <url>… manage the machine's operator keys (~/.amy/operator/,
| |providers] independent of accounts): relay sets where cards +
| retractions publish; status shows master + relays;
| providers lists observer -> service-pubkey.
| graperank register [PROVIDER] declare a NIP-85 provider in your kind:10040 so
| [--service KIND:TAG] [--relay URL] clients can discover it (default: self as the
| [--private] 30382:rank provider at your first outbox relay).
| graperank providers [USER] [--refresh] list a user's declared NIP-85 trusted providers
| [--timeout SECS] (default: active account).
|
|Zaps (NIP-57):
| zap user USER SATS build a profile zap-request, fetch a BOLT11
| [--comment X] [--anon|--private] invoice from the recipient's LN service
@@ -649,11 +689,15 @@ private fun printUsage() {
|
| marmot reset [--yes] wipe all local MLS/KeyPackage state (destructive)
|
|Local event store (`<data-dir>/events-store/`):
| store stat event count, kind histogram, disk usage
|Local event store (shared, under `<data-dir>/shared/`):
| Backend selected by AMY_STORE: sqlite (default; `shared/events.db`)
| or fs (`AMY_STORE=fs`; the `shared/events-store/` tree). SQLite is
| far more compact at scale — the FS tree spends one file per index
| posting, so large crawls balloon on disk.
| store stat event count + disk usage (kind histogram/mtime on fs)
| store sweep-expired delete events past their NIP-40 expiration
| store scrub rebuild idx/ from canonical events (after edits / crashes)
| store compact drop dangling idx entries (canonical gone)
| store scrub fs: rebuild idx/ from canonical events; sqlite: no-op
| store compact fs: drop dangling idx entries; sqlite: VACUUM
| store reindex-fts rebuild the NIP-50 search index (after a searchable-kinds change)
""".trimMargin(),
)
@@ -0,0 +1,156 @@
/*
* 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.amethyst.cli
import com.fasterxml.jackson.module.kotlin.readValue
import com.vitorpamplona.amethyst.cli.secrets.IdentitySecret
import com.vitorpamplona.amethyst.cli.secrets.SecretStore
import com.vitorpamplona.quartz.nip01Core.core.HexKey
import com.vitorpamplona.quartz.nip01Core.core.hexToByteArray
import com.vitorpamplona.quartz.nip01Core.core.toHexKey
import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer
import com.vitorpamplona.quartz.utils.sha256.sha256
import java.io.File
/**
* Operator-level signing keys for GrapeRank trusted-assertion publishing.
*
* A machine holds ONE operator master seed, **independent of any amy account**,
* stored under `~/.amy/operator/` through the same [SecretStore] backend the
* accounts use (OS keychain / NIP-49 ncryptsec / plaintext). From it we
* deterministically derive ONE service key per observer:
*
* ```
* serviceKey(observer) = sha256(masterPriv ‖ "graperank-provider:" ‖ observerHex ‖ counter)
* ```
*
* That service key signs the observer's kind:30382 rank cards (and their kind:5
* retractions). Deterministic derivation buys two things:
* - **Stable identity** — the same observer always maps to the same key, so
* re-signing a card *replaces* the prior one (kind:30382 is addressable)
* instead of orphaning it and spamming clients with duplicates.
* - **One-secret backup** — back up only the master seed; every service key is
* re-derivable even if the [providers] manifest is lost.
*
* The manifest (`~/.amy/operator/operator.json`) records the master pubkey, the
* configured operator relay(s), and the observer → provider-pubkey mapping. Only
* the master itself is a secret; it rides the [SecretStore] descriptor, so the
* manifest holds public data.
*/
class OperatorKeys(
amyHome: File,
private val secrets: SecretStore,
) {
private val dir = File(amyHome, DIR_NAME)
private val configFile = File(dir, CONFIG_NAME)
data class ProviderRecord(
val providerPubKey: HexKey = "",
)
data class Config(
val masterPubKey: HexKey = "",
val master: IdentitySecret? = null,
val relays: List<String> = emptyList(),
val providers: MutableMap<HexKey, ProviderRecord> = mutableMapOf(),
)
private fun load(): Config? = if (configFile.exists()) Output.mapper.readValue<Config>(configFile.readText()) else null
private fun save(cfg: Config) {
SecureFileIO.secureMkdirs(dir)
configFile.writeText(Output.mapper.writeValueAsString(cfg))
SecureFileIO.tighten(configFile)
}
/** True once an operator master exists on this machine. */
fun exists(): Boolean = load()?.master != null
/** Load (or, on first use, create + persist) the operator master private key. */
private fun masterPriv(): ByteArray {
load()?.master?.let { return secrets.resolve(it).hexToByteArray() }
val kp = KeyPair()
val pub = kp.pubKey.toHexKey()
val secret = secrets.store(pub, kp.privKey!!.toHexKey())
save(Config(masterPubKey = pub, master = secret))
System.err.println("[operator] created operator master ${pub.take(8)}… at ${configFile.path}")
return kp.privKey!!
}
/** The operator master pubkey, creating the master on first use. */
fun masterPubKey(): HexKey {
masterPriv()
return load()!!.masterPubKey
}
/**
* The deterministic service key for [observerHex], recording the observer →
* provider-pubkey mapping in the manifest. The counter loop only ever runs
* once in practice — it's a guard for the ~2^-128 chance a sha256 output isn't
* a valid secp256k1 scalar.
*/
fun serviceKey(observerHex: HexKey): KeyPair {
val master = masterPriv()
var counter = 0
while (true) {
val material = master + "$DERIVATION_LABEL$observerHex:$counter".encodeToByteArray()
val kp = runCatching { KeyPair(privKey = sha256(material)) }.getOrNull()
if (kp?.privKey != null) {
recordProvider(observerHex, kp.pubKey.toHexKey())
return kp
}
counter++
}
}
private fun recordProvider(
observerHex: HexKey,
providerPubKey: HexKey,
) {
val cfg = load() ?: return
if (cfg.providers[observerHex]?.providerPubKey == providerPubKey) return
cfg.providers[observerHex] = ProviderRecord(providerPubKey)
save(cfg)
}
/** Relays the operator publishes all its 30382 cards + retractions to. */
fun operatorRelays(): Set<NormalizedRelayUrl> =
load()
?.relays
.orEmpty()
.mapNotNull { RelayUrlNormalizer.normalizeOrNull(it) }
.toSet()
fun setRelays(urls: List<String>) {
masterPriv() // make sure the config (and master) exists first
save(load()!!.copy(relays = urls))
}
fun providers(): Map<HexKey, ProviderRecord> = load()?.providers.orEmpty()
companion object {
private const val DIR_NAME = "operator"
private const val CONFIG_NAME = "operator.json"
private const val DERIVATION_LABEL = "graperank-provider:"
}
}
@@ -0,0 +1,96 @@
/*
* 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.amethyst.cli
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.AuthMessage
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.ClosedMessage
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.NoticeMessage
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.atomic.AtomicLong
/**
* Client-wide tally of the relay feedback the crawl would otherwise never see:
* `NOTICE` frames, `CLOSED` reasons (`auth-required` / `rate-limited` /
* `restricted` / …), and NIP-42 `AUTH` challenges. Registered as a
* [RelayConnectionListener] on the shared client, so every incoming message
* during a run is counted and a REQ failure can be explained instead of
* guessed at.
*
* Callbacks fire on the per-relay socket threads, so all state is concurrent.
*/
class RelayDiagnostics : RelayConnectionListener {
private val closedByReason = ConcurrentHashMap<String, AtomicLong>()
private val noticeSamples = ConcurrentHashMap<String, AtomicLong>()
private val authChallenges = AtomicLong()
override fun onIncomingMessage(
relay: IRelayClient,
msgStr: String,
msg: Message,
) {
when (msg) {
// CLOSED reasons follow the NIP-01 machine-readable "word: text"
// convention, so the prefix categorises the failure.
is ClosedMessage -> bump(closedByReason, prefix(msg.message))
// NOTICE is free-form; keep the (truncated) text so recurring
// relay complaints ("too many concurrent REQs", …) are visible.
is NoticeMessage -> if (noticeSamples.size < MAX_DISTINCT_NOTICES) bump(noticeSamples, msg.message.trim().take(80))
is AuthMessage -> authChallenges.incrementAndGet()
else -> Unit
}
}
private fun bump(
map: ConcurrentHashMap<String, AtomicLong>,
key: String,
) {
map.getOrPut(key) { AtomicLong() }.incrementAndGet()
}
/** The NIP-01 machine-readable prefix (`word` before `:`), or `other`. */
private fun prefix(message: String): String {
val head = message.substringBefore(':').trim().lowercase()
return head.ifEmpty { "other" }.take(24)
}
fun hadFeedback(): Boolean = authChallenges.get() > 0 || closedByReason.isNotEmpty() || noticeSamples.isNotEmpty()
/** JSON-friendly summary for the command output. */
fun snapshot(): Map<String, Any?> =
mapOf(
"auth_challenges" to authChallenges.get(),
"closed_by_reason" to closedByReason.entries.associate { it.key to it.value.get() }.toSortedMap(),
"notices" to noticeSamples.values.sumOf { it.get() },
"notice_top" to
noticeSamples.entries
.sortedByDescending { it.value.get() }
.take(TOP_NOTICES)
.map { "${it.key} (${it.value.get()})" },
)
companion object {
private const val MAX_DISTINCT_NOTICES = 500
private const val TOP_NOTICES = 8
}
}
@@ -0,0 +1,89 @@
/*
* 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.amethyst.cli
import com.vitorpamplona.quartz.nip01Core.jackson.JacksonMapper
import com.vitorpamplona.quartz.nip01Core.store.IEventStore
import com.vitorpamplona.quartz.nip01Core.store.fs.FsEventStore
import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore
import kotlin.io.path.Path
/** On-disk backend for the shared event store. */
enum class StoreBackend {
/**
* Single SQLite database file at [DataDir.eventsDbFile]. Postings live
* in shared B-tree pages, so an event's kind/author/tag indexes cost a
* handful of rows — not one 4 KB-block file each, the way the FS store
* lays them out. For crawl-scale corpora (hundreds of thousands of
* follow lists) this is several times smaller on disk and the default.
*/
SQLITE,
/**
* Filesystem tree at [DataDir.eventsDir] — one pretty-printed JSON file
* per event plus one file per index posting. Human-inspectable with
* `cat`/`jq`/`git diff`, but every posting rounds up to a filesystem
* block, so a large corpus balloons. Opt in with `AMY_STORE=fs`.
*/
FS,
}
/**
* Chooses and opens the event-store backend for `amy`. The backend is
* selected by the `AMY_STORE` environment variable and defaults to
* [StoreBackend.SQLITE]; set `AMY_STORE=fs` for the legacy filesystem
* store. Both backends implement [IEventStore], so every command works
* unchanged regardless of the choice — the only user-visible difference
* is where bytes land ([DataDir.eventsDbFile] vs [DataDir.eventsDir]) and
* how much disk they take.
*/
object StoreFactory {
const val ENV = "AMY_STORE"
/** Resolve the configured backend. Unrecognised values fall back to the default. */
fun backend(): StoreBackend =
when (System.getenv(ENV)?.trim()?.lowercase()) {
"fs", "file", "files", "filesystem" -> StoreBackend.FS
else -> StoreBackend.SQLITE
}
/**
* Open the store for [dataDir] using the configured [backend]. Events
* are written pretty-printed on the FS backend so the on-disk JSON stays
* inspection-friendly; the SQLite backend stores the compact NIP-01
* form internally. Neither is re-used for signature checks (verification
* always re-canonicalises), so the stored representation is purely an
* implementation detail. Callers own [IEventStore.close].
*/
fun open(dataDir: DataDir): IEventStore =
when (backend()) {
StoreBackend.SQLITE -> {
// BundledSQLiteDriver won't create parent directories.
dataDir.eventsDbFile.parentFile?.mkdirs()
EventStore(dbName = dataDir.eventsDbFile.absolutePath, relay = null)
}
StoreBackend.FS ->
FsEventStore(
root = Path(dataDir.eventsDir.absolutePath),
eventToJson = JacksonMapper::toJsonPretty,
)
}
}
@@ -0,0 +1,755 @@
/*
* 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.amethyst.cli.commands
import com.vitorpamplona.amethyst.cli.Args
import com.vitorpamplona.amethyst.cli.Context
import com.vitorpamplona.amethyst.cli.DataDir
import com.vitorpamplona.amethyst.cli.Output
import com.vitorpamplona.amethyst.commons.defaults.Constants
import com.vitorpamplona.amethyst.commons.defaults.DefaultIndexerRelayList
import com.vitorpamplona.quartz.experimental.graperank.GrapeRank
import com.vitorpamplona.quartz.experimental.graperank.GrapeRankDataCrawler
import com.vitorpamplona.quartz.experimental.graperank.GrapeRankParams
import com.vitorpamplona.quartz.experimental.graperank.GrapeRankPublisher
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
import com.vitorpamplona.quartz.nip09Deletions.DeletionIndex
import com.vitorpamplona.quartz.nip51Lists.muteList.MuteListEvent
import com.vitorpamplona.quartz.nip56Reports.ReportEvent
import com.vitorpamplona.quartz.nip85TrustedAssertions.list.TrustProviderListEvent
import com.vitorpamplona.quartz.nip85TrustedAssertions.list.serviceProviders
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.Dispatchers
import kotlinx.coroutines.async
import kotlinx.coroutines.awaitAll
import kotlinx.coroutines.coroutineScope
import kotlin.math.roundToInt
/**
* `amy graperank [OBSERVER] [flags]` — compute GrapeRank web-of-trust scores.
*
* GrapeRank assigns every user reachable in the follow/mute/report graph a
* subjective trust score in `[0, 1]` from the observer's point of view (the
* observer has full self-trust). It crawls the follow graph outward using the
* outbox model — each user's kind:10002 write relays are located first, then
* their kind:3 / kind:10000 / kind:1984 events are fetched from *their own*
* relays. The crawl is exhaustive: it keeps going, with no user cap, until every
* discovered user's outbox has been checked and their contact list pulled (an
* 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:
* - `amy graperank sync [OBSERVER]` — network only: crawl the reachable graph's
* kind 3/10000/1984/10002 into the local store. 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.
* - bare `amy graperank [OBSERVER]` — the convenience combo: sync then score.
*
* Sub-verbs complete the NIP-85 provider experience — the discovery layer that
* lets clients find and consume those assertions:
* - `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).
* - `amy graperank providers [USER]` — list a user's trusted providers.
*/
object GrapeRankCommand {
// Broad, big general relays that carry kind:10002 for many users, added to the
// crawler's discovery set to raise the odds of resolving a stranger's outbox.
private val EXTRA_DISCOVERY_RELAYS: Set<NormalizedRelayUrl> =
listOf(
"wss://relay.damus.io",
"wss://relay.snort.social",
"wss://offchain.pub",
"wss://nostr.land",
"wss://eden.nostr.land",
).mapNotNull { RelayUrlNormalizer.normalizeOrNull(it) }.toSet()
suspend fun dispatch(
dataDir: DataDir,
tail: Array<String>,
): Int =
// Sub-verbs are explicit words; anything else (npub / hex / nprofile /
// NIP-05, or nothing) is the OBSERVER positional for a score computation.
when (tail.firstOrNull()) {
"register" -> register(dataDir, tail.drop(1).toTypedArray())
"providers" -> providers(dataDir, tail.drop(1).toTypedArray())
"operator" -> operator(dataDir, tail.drop(1).toTypedArray())
"sync" -> sync(dataDir, tail.drop(1).toTypedArray())
"score" -> run(dataDir, tail.drop(1).toTypedArray(), forceOffline = true)
else -> run(dataDir, tail)
}
suspend fun run(
dataDir: DataDir,
rest: Array<String>,
forceOffline: Boolean = false,
): Int {
val args = Args(rest)
val observerArg = args.positionalOrNull(0)
// Crawl to full convergence by default (every reachable user's outbox
// checked). --max-rounds is only a safety backstop; --max-hops bounds the
// follow-graph distance from the observer that we crawl (Brainstorm uses 8).
val limit = args.intFlag("limit", 100)
val minScore = args.flag("min-score")?.toDoubleOrNull() ?: 0.0
// `graperank score` forces the local (no-network) path; `--offline` does the
// same on the bare command. Either way we build + score from the store only.
val offline = forceOffline || args.bool("offline")
// Crawl tuning (--max-rounds/--max-hops/--timeout/--diagnose/--drain-concurrency)
// is read straight from args by [newCrawler]; only these two are surfaced in
// 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.
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(
attenuation = args.flag("attenuation")?.toDoubleOrNull() ?: GrapeRankParams().attenuation,
rigor = args.flag("rigor")?.toDoubleOrNull() ?: GrapeRankParams().rigor,
)
Context.open(dataDir).use { ctx ->
ctx.prepare()
val observer = observerArg?.let { ctx.requireUserHex(it) } ?: ctx.identity.pubKeyHex
// Contact lists stream straight into a compact int-CSR structure as the
// crawl finds them and the Event is discarded, so the whole network fits
// in memory without holding millions of kind:3 objects.
val builder = TrustGraphBuilder()
var contactListsFed = 0
// Wall time to read + deserialize the contact lists out of the store
// (offline path only; online streams them in during the crawl).
var storeLoadMs: Long? = null
// Crawl telemetry (online path only): rounds, relays contacted, the
// per-hop histogram, and the network-bound download time that dominates a
// from-scratch run. Null on the offline path.
var crawlStats: GrapeRankDataCrawler.Stats? = null
if (!offline) {
val stats = newCrawler(ctx, args).crawl(observer, builder)
crawlStats = stats
contactListsFed = stats.contactListsFed
reportRelayFeedback(ctx)
} else {
// Offline: stream contact lists from the local store into the graph.
val loadStart = System.nanoTime()
for (event in ctx.store.query<Event>(Filter(kinds = listOf(ContactListEvent.KIND)))) {
if (event is ContactListEvent) {
builder.addFollows(event.pubKey, event.verifiedFollowKeySet())
contactListsFed++
}
}
storeLoadMs = (System.nanoTime() - loadStart) / 1_000_000
System.err.println("[graperank] offline: $contactListsFed contact lists from local store in $storeLoadMs ms")
}
// Mutes + reports come from the store (both paths). Far fewer than contact
// lists, so materialising them is cheap.
for (event in ctx.store.query<Event>(Filter(kinds = listOf(MuteListEvent.KIND)))) {
if (event is MuteListEvent) builder.addMutes(event.pubKey, event.linkedPubKeys())
}
val reportsDeleted = materializeReports(ctx, builder)
val buildStart = System.nanoTime()
val graph = builder.build()
val buildMs = (System.nanoTime() - buildStart) / 1_000_000
System.err.println("[graperank] graph built: ${graph.nodeCount} users, ${graph.edgeCount()} edges in $buildMs ms; scoring…")
// Live scoring progress: fires once per Gauss-Seidel sweep with the
// running node-update count and how many nodes still moved more than the
// convergence delta this sweep — that second number trends to 0, so a
// large graph shows convergence instead of hanging silently.
val scoreStart = System.nanoTime()
var sweeps = 0
val scores =
GrapeRank(params).compute(graph, observer) { visited, stillMoving ->
sweeps++
System.err.println("[graperank] scoring sweep $sweeps: $visited node-updates, $stillMoving still moving")
}
fun rankOf(score: Double) = (score * 100).roundToInt()
val observerId = graph.idOf(observer)
// Reachable users with positive trust at or above --min-score, high→low.
val rankedIds = ArrayList<Int>()
for (id in 0 until graph.nodeCount) {
if (id != observerId && scores[id] > 0.0 && scores[id] >= minScore) rankedIds.add(id)
}
rankedIds.sortByDescending { scores[it] }
val scoringMs = (System.nanoTime() - scoreStart) / 1_000_000
System.err.println("[graperank] scored ${rankedIds.size} users in $scoringMs ms")
val hopHistogram = crawlStats?.hopHistogram.orEmpty()
val result =
linkedMapOf<String, Any?>(
"observer" to observer,
"crawl_rounds" to (crawlStats?.rounds ?: 0),
"relays_contacted" to (crawlStats?.relaysContacted ?: 0),
"relay_feedback" to if (ctx.relayDiagnostics.hadFeedback()) ctx.relayDiagnostics.snapshot() else null,
"relay_throttling" to if (ctx.relayLimiter.hadThrottling()) ctx.relayLimiter.snapshot() else null,
"max_hop_reached" to (hopHistogram.keys.maxOrNull() ?: 0),
"users_by_hop" to hopHistogram.mapKeys { it.key.toString() },
"graph_users" to graph.nodeCount,
"graph_edges" to graph.edgeCount(),
"reports_deleted" to reportsDeleted,
"users_scored" to rankedIds.size,
"download_ms" to crawlStats?.downloadMs,
"verify_ms" to crawlStats?.verifyMs,
"insert_ms" to crawlStats?.insertMs,
"events_stored" to crawlStats?.eventsStored,
"insert_batch" to insertBatch,
"park_timeout_ms" to parkTimeoutMs,
"store_load_ms" to storeLoadMs,
"graph_build_ms" to buildMs,
"scoring_ms" to scoringMs,
"scoring_sweeps" to sweeps,
"scores" to
rankedIds.take(limit).map {
mapOf("pubkey" to graph.pubkeyOf(it), "score" to scores[it], "rank" to rankOf(scores[it]))
},
)
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
// 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()
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
}
Output.emit(result)
return 0
}
}
/**
* Configure the outbox-model crawler from the crawl flags on [args] plus the
* account's relay policy. Shared by the bare command and `graperank sync`.
* Relay policy — where a stranger's kind:10002 is found (index/discovery
* aggregators + general defaults) and best-effort general relays that might
* hold content when an outbox is unknown — lives in app code, so the quartz
* crawler takes it injected.
*/
private suspend fun newCrawler(
ctx: Context,
args: Args,
): GrapeRankDataCrawler {
val discoveryRelays =
ctx.bootstrapRelays() + Constants.eventFinderRelays + DefaultIndexerRelayList + EXTRA_DISCOVERY_RELAYS
val contentFallback = ctx.bootstrapRelays() + Constants.eventFinderRelays
return GrapeRankDataCrawler(
client = ctx.client,
store = ctx.store,
limiter = ctx.relayLimiter,
config =
GrapeRankDataCrawler.Config(
relayListDiscoveryRelays = discoveryRelays,
contentFallbackRelays = contentFallback,
maxRounds = args.intFlag("max-rounds", Int.MAX_VALUE),
maxHops = args.intFlag("max-hops", Int.MAX_VALUE),
timeoutMs = args.longFlag("timeout", 10L) * 1000,
parkTimeoutMs = args.longFlag("park-timeout", 40L) * 1000,
diagnose = args.bool("diagnose"),
insertBatchSize = args.intFlag("insert-batch", 500),
drainConcurrency = args.intFlag("drain-concurrency", 24),
),
log = { System.err.println(it) },
)
}
/** Echo any relay NOTICE/CLOSED feedback + adaptive throttling the crawl saw. */
private fun reportRelayFeedback(ctx: Context) {
if (ctx.relayDiagnostics.hadFeedback()) {
System.err.println("[graperank] relay feedback: ${ctx.relayDiagnostics.snapshot()}")
}
if (ctx.relayLimiter.hadThrottling()) {
System.err.println("[graperank] relay throttling: ${ctx.relayLimiter.snapshot()}")
}
}
/**
* `amy graperank sync [OBSERVER]` — network-only WoT data sync. Crawls the
* reachable follow/mute/report graph into the local store (kind 3/10000/1984/
* 10002) and reports what it loaded, WITHOUT scoring. Idempotent + cumulative:
* run it a few times to make sure everything is loaded, then `graperank score`.
*/
private suspend fun sync(
dataDir: DataDir,
rest: Array<String>,
): Int {
val args = Args(rest)
val observerArg = args.positionalOrNull(0)
Context.open(dataDir).use { ctx ->
ctx.prepare()
val observer = observerArg?.let { ctx.requireUserHex(it) } ?: ctx.identity.pubKeyHex
// Persist-only crawl: no in-memory graph (null builder); every event
// still lands in the store for a later `score`.
val stats = newCrawler(ctx, args).crawl(observer, null)
reportRelayFeedback(ctx)
Output.emit(
linkedMapOf<String, Any?>(
"observer" to observer,
"crawl_rounds" to stats.rounds,
"relays_contacted" to stats.relaysContacted,
"relay_feedback" to if (ctx.relayDiagnostics.hadFeedback()) ctx.relayDiagnostics.snapshot() else null,
"relay_throttling" to if (ctx.relayLimiter.hadThrottling()) ctx.relayLimiter.snapshot() else null,
"max_hop_reached" to (stats.hopHistogram.keys.maxOrNull() ?: 0),
"users_by_hop" to stats.hopHistogram.mapKeys { it.key.toString() },
"users_discovered" to stats.hopHistogram.values.sum(),
"contact_lists_fed" to stats.contactListsFed,
"download_ms" to stats.downloadMs,
"verify_ms" to stats.verifyMs,
"insert_ms" to stats.insertMs,
"events_stored" to stats.eventsStored,
),
)
}
return 0
}
/**
* 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.
*/
private suspend fun signCards(
cards: List<Pair<HexKey, Int>>,
signer: NostrSigner,
): 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)) },
)
}
chunk.size
}
}.awaitAll()
.sum()
}
}
/**
* `amy graperank operator [status | relay <url>… | providers]`
*
* Manage the machine's operator keys used to sign trusted-assertion cards.
* - `status` (default): master pubkey, configured relay(s), provider count.
* - `relay <url>…`: set the operator relay(s) the cards + retractions publish
* to; creates the operator master on first use.
* - `providers`: the observer -> provider-pubkey mapping learned so far.
*/
private fun operator(
dataDir: DataDir,
rest: Array<String>,
): Int {
val opKeys = dataDir.operatorKeys()
return when (rest.firstOrNull()) {
"relay" -> {
val urls = rest.drop(1).filter { it.isNotBlank() }
val normalized = urls.mapNotNull { RelayUrlNormalizer.normalizeOrNull(it) }
if (normalized.isEmpty()) return Output.error("bad_args", "usage: amy graperank operator relay <wss://…> [<wss://…> …]")
opKeys.setRelays(urls)
Output.emit(mapOf("master_pubkey" to opKeys.masterPubKey(), "relays" to normalized.map { it.url }))
0
}
"providers" -> {
Output.emit(
mapOf(
"master_pubkey" to if (opKeys.exists()) opKeys.masterPubKey() else null,
"providers" to opKeys.providers().map { (observer, rec) -> mapOf("observer" to observer, "provider_pubkey" to rec.providerPubKey) },
),
)
0
}
null, "status" -> {
if (!opKeys.exists()) {
Output.emit(mapOf("initialized" to false))
} else {
Output.emit(
mapOf(
"initialized" to true,
"master_pubkey" to opKeys.masterPubKey(),
"relays" to opKeys.operatorRelays().map { it.url },
"providers" to opKeys.providers().size,
),
)
}
0
}
else -> Output.error("bad_args", "unknown operator subcommand '${rest.first()}' (status | relay | providers)")
}
}
/**
* `amy graperank register [PROVIDER] [--service KIND:TAG] [--relay URL] [--private]`
*
* Add a NIP-85 provider entry to the account's kind:10040
* [TrustProviderListEvent] — the declaration a client reads to discover which
* key publishes which assertion, and where. Defaults to declaring *self* as
* the `30382:rank` provider at the account's first outbox relay, which is the
* self-advertisement a GrapeRank provider makes so its followers can find the
* cards it publishes. Fetches the freshest list first so existing providers
* are preserved.
*/
private suspend fun register(
dataDir: DataDir,
rest: Array<String>,
): Int {
val args = Args(rest)
val providerArg = args.positionalOrNull(0) ?: args.flag("provider")
val serviceArg = args.flag("service")
val relayArg = args.flag("relay")
val isPrivate = args.bool("private")
val timeoutMs = args.longFlag("timeout", 8L) * 1000
val service =
serviceArg?.let {
ServiceType.parse(it) ?: return Output.error("bad_args", "--service must be KIND:TAG, e.g. 30382:rank")
} ?: ProviderTypes.rank
Context.open(dataDir).use { ctx ->
ctx.prepare()
val self = ctx.identity.pubKeyHex
val provider = providerArg?.let { ctx.requireUserHex(it) } ?: self
val outbox = ctx.outboxRelays()
val relay =
relayArg?.let { RelayUrlNormalizer.normalizeOrNull(it) }
?: outbox.firstOrNull()
?: return Output.error("no_relays", "no relay hint; pass --relay URL or configure outbox relays")
val latest = fetchLatestProviderList(ctx, self, outbox, timeoutMs)
val alreadyListed =
latest?.serviceProviders()?.any {
it.service == service && it.pubkey == provider && it.relayUrl == relay
} ?: false
if (alreadyListed) {
Output.emit(
mapOf(
"service" to service.toValue(),
"provider" to provider,
"relay" to relay.url,
"changed" to false,
"based_on" to latest?.id,
),
)
return 0
}
val tag = ServiceProviderTag(service, provider, relay)
val event =
if (latest == null) {
TrustProviderListEvent.create(tag, isPrivate = isPrivate, signer = ctx.signer)
} else {
TrustProviderListEvent.add(latest, tag, isPrivate = isPrivate, signer = ctx.signer)
}
val ack = ctx.publish(event, outbox)
Output.emit(
mapOf(
"service" to service.toValue(),
"provider" to provider,
"relay" to relay.url,
"private" to isPrivate,
"changed" to true,
"event_id" to event.id,
"based_on" to latest?.id,
"published_to" to ack.filterValues { it }.keys.map { it.url },
"rejected_by" to ack.filterValues { !it }.keys.map { it.url },
),
)
return 0
}
}
/**
* `amy graperank providers [USER] [--refresh] [--timeout SECS]`
*
* List the NIP-85 trusted providers a user declares in their kind:10040
* (default: the active account). Cache-first; falls back to a relay drain on
* a miss or with `--refresh`. For the active account, private (NIP-44)
* provider entries are decrypted and included too.
*/
private suspend fun providers(
dataDir: DataDir,
rest: Array<String>,
): Int {
val args = Args(rest)
val userArg = args.positionalOrNull(0)
val refresh = args.bool("refresh")
val timeoutMs = args.longFlag("timeout", 8L) * 1000
Context.open(dataDir).use { ctx ->
ctx.prepare()
val user = userArg?.let { ctx.requireUserHex(it) } ?: ctx.identity.pubKeyHex
val isSelf = user == ctx.identity.pubKeyHex
var event = if (refresh) null else providerListOf(ctx, user)
if (event == null) {
ctx.drain(
(ctx.bootstrapRelays() + Constants.eventFinderRelays).associateWith {
listOf(Filter(kinds = listOf(TrustProviderListEvent.KIND), authors = listOf(user), limit = 1))
},
timeoutMs,
)
event = providerListOf(ctx, user)
}
if (event == null) {
Output.emit(mapOf("user" to user, "found" to false, "providers" to emptyList<Any>()))
return 0
}
val public = event.serviceProviders()
val private = if (isSelf) event.privateTags(ctx.signer)?.serviceProviders().orEmpty() else emptyList()
fun render(
tag: ServiceProviderTag,
scope: String,
) = mapOf(
"service" to tag.service.toValue(),
"provider" to tag.pubkey,
"relay" to tag.relayUrl.url,
"scope" to scope,
)
Output.emit(
mapOf(
"user" to user,
"found" to true,
"event_id" to event.id,
"created_at" to event.createdAt,
"providers" to public.map { render(it, "public") } + private.map { render(it, "private") },
),
)
return 0
}
}
/** Latest known kind:10040 provider list for [pubKey] from the local store. */
private suspend fun providerListOf(
ctx: Context,
pubKey: HexKey,
): TrustProviderListEvent? =
ctx.store
.query<Event>(Filter(kinds = listOf(TrustProviderListEvent.KIND), authors = listOf(pubKey), limit = 1))
.firstOrNull() as? TrustProviderListEvent
/**
* Fetch the freshest kind:10040 for [pubKey] from [relays] so a register
* builds on top of the current provider set instead of clobbering it.
*/
private suspend fun fetchLatestProviderList(
ctx: Context,
pubKey: HexKey,
relays: Set<NormalizedRelayUrl>,
timeoutMs: Long,
): TrustProviderListEvent? {
if (relays.isEmpty()) return providerListOf(ctx, pubKey)
val filter = Filter(kinds = listOf(TrustProviderListEvent.KIND), authors = listOf(pubKey), limit = 1)
ctx.drain(relays.associateWith { listOf(filter) }, timeoutMs)
return providerListOf(ctx, pubKey)
}
/**
* Feed reports into [builder], dropping any that a valid NIP-09 deletion has
* retracted. Uses quartz's [DeletionIndex] — the same indexer the Android
* app's LocalCache runs — which keys each deletion under the DELETER's pubkey,
* so `hasBeenDeleted(report)` is true only when the report's own author
* deleted it (NIP-09: a deletion is authoritative only from the event's
* author). It also honours created_at ordering. Returns how many were dropped.
*/
private suspend fun materializeReports(
ctx: Context,
builder: TrustGraphBuilder,
): Int {
val reports = ctx.store.query<Event>(Filter(kinds = listOf(ReportEvent.KIND))).filterIsInstance<ReportEvent>()
if (reports.isEmpty()) return 0
// Everything in the store already passed verifyAndStore, so mark the
// deletions as verified and skip the redundant signature check.
val deletions = DeletionIndex()
for (ev in ctx.store.query<Event>(Filter(kinds = listOf(DeletionEvent.KIND)))) {
if (ev is DeletionEvent) deletions.add(ev, wasVerified = true)
}
var dropped = 0
for (r in reports) {
if (deletions.hasBeenDeleted(r)) {
dropped++
continue
}
builder.addReports(r.pubKey, r.reportedAuthor().map { it.pubkey })
}
if (dropped > 0) System.err.println("[graperank] dropped $dropped retracted reports (NIP-09 deletions)")
return dropped
}
/**
* If the active account IS the observer (so we hold their key), publish/refresh
* their kind:10040 declaring `30382:rank` -> [providerPubkey] at [relay], to
* their own outbox relays — the NIP-85 pointer a client follows to find these
* cards. Returns the 10040 event id, or null when we don't hold the key (a
* third-party observer must add the provider to their 10040 out-of-band).
*/
private suspend fun maybePublishObserverProviderList(
ctx: Context,
observer: HexKey,
providerPubkey: HexKey,
relay: NormalizedRelayUrl,
): String? {
if (observer != ctx.identity.pubKeyHex) return null
val service = ProviderTypes.rank
val outbox = ctx.outboxRelays()
val latest = fetchLatestProviderList(ctx, observer, outbox, 8_000)
val alreadyListed =
latest?.serviceProviders()?.any {
it.service == service && it.pubkey == providerPubkey && it.relayUrl == relay
} ?: false
if (alreadyListed) return latest?.id
val tag = ServiceProviderTag(service, providerPubkey, relay)
val event =
if (latest == null) {
TrustProviderListEvent.create(tag, isPrivate = false, signer = ctx.signer)
} else {
TrustProviderListEvent.add(latest, tag, isPrivate = false, signer = ctx.signer)
}
ctx.publish(event, outbox)
return event.id
}
}
@@ -22,28 +22,39 @@ package com.vitorpamplona.amethyst.cli.commands
import com.vitorpamplona.amethyst.cli.DataDir
import com.vitorpamplona.amethyst.cli.Output
import com.vitorpamplona.amethyst.cli.StoreStats
import com.vitorpamplona.quartz.nip01Core.jackson.JacksonMapper
import com.vitorpamplona.amethyst.cli.StoreBackend
import com.vitorpamplona.amethyst.cli.StoreFactory
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.store.IEventStore
import com.vitorpamplona.quartz.nip01Core.store.fs.FsEventStore
import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore
import java.io.File
import java.io.IOException
import java.nio.file.Files
import java.nio.file.Path
import java.util.concurrent.TimeUnit
import kotlin.io.path.exists
/**
* `amy store <stat|sweep-expired|scrub|compact>` — direct introspection
* and maintenance of the file-backed event store at
* `<data-dir>/events-store/`.
* and maintenance of the shared event store under `<data-dir>/shared/`.
*
* - `stat` total event count, kind histogram, disk bytes,
* mtime range — pure read, no relay traffic.
* The store backend is selected by `AMY_STORE` (SQLite by default, or the
* FS tree with `AMY_STORE=fs` — see [StoreFactory]); each verb adapts to
* whichever is active:
*
* - `stat` total event count, disk bytes, backend, plus (FS only)
* the per-kind histogram and mtime range — pure read,
* no relay traffic.
* - `sweep-expired` delete events whose NIP-40 `expiration` tag has
* passed (per the store's own sweep logic). Run
* from cron / scheduler / `amy` periodically.
* - `scrub` rebuild every `idx/` entry from the canonical
* events. Recovers from partial-write crashes or
* external edits.
* - `compact` drop dangling `idx/` entries whose canonical is
* gone. Cheaper than scrub.
* - `scrub` FS: rebuild every `idx/` entry from the canonical
* events, recovering from partial-write crashes or
* external edits. SQLite: a no-op (indexes are updated
* transactionally and can't drift).
* - `compact` FS: drop dangling `idx/` entries whose canonical is
* gone. SQLite: `VACUUM` the database to reclaim space.
* - `reindex-fts` wipe and rebuild only the NIP-50 full-text search
* index from the stored events. Run after a quartz
* upgrade that changes which kinds are searchable.
@@ -66,16 +77,116 @@ object StoreCommands {
),
)
private fun stat(dataDir: DataDir): Int {
val stats = StoreStats.of(dataDir.eventsDir.toPath())
private suspend fun stat(dataDir: DataDir): Int =
when (StoreFactory.backend()) {
StoreBackend.SQLITE -> sqliteStat(dataDir)
StoreBackend.FS -> fsStat(dataDir)
}
/**
* SQLite `stat`: total count via `COUNT(*)` and on-disk bytes from the
* DB file plus its `-wal`/`-shm` sidecars. The per-kind histogram and
* mtime range are FS-store concepts (they read the `idx/kind` tree and
* file mtimes), so they're omitted here.
*/
private suspend fun sqliteStat(dataDir: DataDir): Int {
val dbFile = dataDir.eventsDbFile
if (!dbFile.exists()) {
Output.emit(
mapOf(
"backend" to "sqlite",
"events" to 0,
"disk_bytes" to 0L,
"root" to dbFile.absolutePath,
),
)
return 0
}
val count =
EventStore(dbName = dbFile.absolutePath, relay = null).use { store ->
store.count(Filter())
}
val diskBytes =
listOf("", "-wal", "-shm").sumOf { suffix ->
val f = File(dbFile.absolutePath + suffix)
if (f.isFile) f.length() else 0L
}
Output.emit(
mapOf(
"events" to stats.events,
"by_kind" to stats.byKind,
"disk_bytes" to stats.diskBytes,
"oldest_at" to stats.oldestAt,
"newest_at" to stats.newestAt,
"root" to stats.root.toString(),
"backend" to "sqlite",
"events" to count,
"disk_bytes" to diskBytes,
"root" to dbFile.absolutePath,
),
)
return 0
}
private fun fsStat(dataDir: DataDir): Int {
val storeRoot = dataDir.eventsDir.toPath()
if (!storeRoot.exists()) {
Output.emit(
mapOf(
"events" to 0,
"by_kind" to emptyMap<String, Long>(),
"disk_bytes" to 0L,
"oldest_at" to null,
"newest_at" to null,
"root" to storeRoot.toAbsolutePath().toString(),
),
)
return 0
}
val eventsRoot = storeRoot.resolve("events")
var count = 0L
var oldest: Long? = null
var newest: Long? = null
if (Files.isDirectory(eventsRoot)) {
Files.walk(eventsRoot).use { stream ->
for (p in stream) {
if (!Files.isRegularFile(p)) continue
if (!p.fileName.toString().endsWith(".json")) continue
count++
val mt =
try {
Files.getLastModifiedTime(p).to(TimeUnit.SECONDS)
} catch (_: IOException) {
continue
}
val o = oldest
if (o == null || mt < o) oldest = mt
val n = newest
if (n == null || mt > n) newest = mt
}
}
}
// Histogram from idx/kind/<k>/ — for a healthy store this is
// exactly one entry per (kind, event), so summing matches `count`.
// Mismatch points at index drift; run `amy store scrub` to fix.
val kindRoot = storeRoot.resolve("idx/kind")
val byKind = sortedMapOf<String, Long>()
if (Files.isDirectory(kindRoot)) {
Files.list(kindRoot).use { stream ->
for (kindDir in stream) {
if (!Files.isDirectory(kindDir)) continue
val n = Files.list(kindDir).use { it.count() }
byKind[kindDir.fileName.toString()] = n
}
}
}
val diskBytes = walkSize(storeRoot)
Output.emit(
mapOf(
"events" to count,
"by_kind" to byKind,
"disk_bytes" to diskBytes,
"oldest_at" to oldest,
"newest_at" to newest,
"root" to storeRoot.toAbsolutePath().toString(),
),
)
return 0
@@ -83,37 +194,65 @@ object StoreCommands {
private suspend fun sweepExpired(dataDir: DataDir): Int =
withStore(dataDir) { store ->
val expiresAtDir = dataDir.eventsDir.toPath().resolve("idx/expires_at")
val before = countEntries(expiresAtDir)
store.deleteExpiredEvents()
val after = countEntries(expiresAtDir)
Output.emit(
mapOf(
"swept" to (before - after).coerceAtLeast(0L),
"remaining" to after,
),
)
if (store is FsEventStore) {
// The FS store exposes its expiration index as a directory,
// so we can report exactly how many entries the sweep cleared.
val expiresAtDir = dataDir.eventsDir.toPath().resolve("idx/expires_at")
val before = countEntries(expiresAtDir)
store.deleteExpiredEvents()
val after = countEntries(expiresAtDir)
Output.emit(
mapOf(
"swept" to (before - after).coerceAtLeast(0L),
"remaining" to after,
),
)
} else {
store.deleteExpiredEvents()
Output.emit(mapOf("ok" to true))
}
0
}
private fun scrub(dataDir: DataDir): Int =
private suspend fun scrub(dataDir: DataDir): Int =
withStore(dataDir) { store ->
store.scrub()
Output.emit(mapOf("ok" to true))
when (store) {
is FsEventStore -> {
store.scrub()
Output.emit(mapOf("ok" to true))
}
// SQLite indexes are written in the same transaction as the
// event, so they can't drift the way the FS `idx/` tree can —
// there is nothing to rebuild.
else ->
Output.emit(
mapOf(
"ok" to true,
"note" to "scrub is a no-op for the sqlite backend (indexes update transactionally)",
),
)
}
0
}
private fun compact(dataDir: DataDir): Int =
private suspend fun compact(dataDir: DataDir): Int =
withStore(dataDir) { store ->
store.compact()
when (store) {
// FS: drop dangling idx/ postings. SQLite: VACUUM to rebuild
// the file and hand freed pages back to the OS.
is FsEventStore -> store.compact()
is EventStore -> store.store.vacuum()
else -> Unit
}
Output.emit(mapOf("ok" to true))
0
}
private suspend fun reindexFts(dataDir: DataDir): Int =
withStore(dataDir) { store ->
val fsBacked = store is FsEventStore
val ftsDir = dataDir.eventsDir.toPath().resolve("idx/fts")
val before = countEntries(ftsDir)
val before = if (fsBacked) countEntries(ftsDir) else 0L
// Drive the resumable, batched path to completion so a huge
// store is processed without holding the writer lock for the
// whole pass. A real long-running caller would persist the
@@ -127,35 +266,34 @@ object StoreCommands {
processed += progress.processedThisBatch
batches++
} while (!progress.done)
val after = countEntries(ftsDir)
Output.emit(
mapOf(
val out =
linkedMapOf<String, Any?>(
"ok" to true,
"processed" to processed,
"batches" to batches,
"tokens_before" to before,
"tokens_after" to after,
),
)
)
if (fsBacked) {
// Token-file counts are an FS-store notion (idx/fts is a
// directory); the SQLite FTS index doesn't expose one.
out["tokens_before"] = before
out["tokens_after"] = countEntries(ftsDir)
}
Output.emit(out)
0
}
/**
* Maintenance verbs only need the store — not identity, not relays,
* not the signer. Skip [Context.open] (which throws if no identity
* has been bootstrapped) and construct the [FsEventStore] directly
* from [DataDir.eventsDir]. Pretty formatter matches what the rest
* of the CLI uses for inspection-friendly output.
* not the signer. Skip [Context.open] (which throws if no identity has
* been bootstrapped) and open the configured backend directly via
* [StoreFactory], so `amy store` acts on whichever store the rest of
* the CLI is using.
*/
private inline fun withStore(
private suspend fun withStore(
dataDir: DataDir,
body: (FsEventStore) -> Int,
body: suspend (IEventStore) -> Int,
): Int {
val store =
FsEventStore(
root = dataDir.eventsDir.toPath(),
eventToJson = JacksonMapper::toJsonPretty,
)
val store = StoreFactory.open(dataDir)
try {
return body(store)
} finally {
@@ -163,6 +301,23 @@ object StoreCommands {
}
}
private fun walkSize(root: Path): Long {
if (!Files.exists(root)) return 0L
var total = 0L
Files.walk(root).use { stream ->
for (p in stream) {
if (!Files.isRegularFile(p)) continue
total +=
try {
Files.size(p)
} catch (_: IOException) {
0L
}
}
}
return total
}
private fun countEntries(dir: Path): Long {
if (!Files.isDirectory(dir)) return 0L
return Files.list(dir).use { it.count() }
@@ -0,0 +1,168 @@
/*
* 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 androidx.compose.runtime.Immutable
import com.vitorpamplona.quartz.nip01Core.core.HexKey
import kotlin.math.abs
import kotlin.math.exp
import kotlin.math.ln
/**
* Tunable GrapeRank parameters. Defaults mirror the reference implementation at
* <https://github.com/vitorpamplona/graperank> and NosFabrica's Brainstorm
* `DEFAULT` preset.
*/
@Immutable
data class GrapeRankParams(
val attenuation: Double = 0.85,
val rigor: Double = 0.5,
val directFollowConfidence: Double = 0.5,
val indirectFollowConfidence: Double = 0.03,
val muteConfidence: Double = 0.5,
val reportConfidence: Double = 0.5,
val convergence: Double = 0.0001,
)
/**
* GrapeRank — a subjective, observer-centric web-of-trust score in `[0, 1]` for
* every user reachable from an observer in a [TrustGraph]. See the algorithm
* notes in `TrustGraph`/`GrapeRankTest`; this is the single-observer
* Gauss-Seidel form, operating on the compact int-CSR graph so it scales to the
* whole network.
*
* [compute] returns a `DoubleArray` indexed by node id (`graph.idOf(pubkey)`),
* not a map — at millions of nodes a boxed map would dwarf the graph itself. The
* observer's own entry stays pinned at `1.0`; callers rank the others.
*/
class GrapeRank(
val params: GrapeRankParams = GrapeRankParams(),
) {
private val rigidity = -ln(params.rigor)
/** Exponential saturation turning accumulated weight into a confidence in `[0, 1)`. */
private fun weightToConfidence(weight: Double): Double = 1.0 - exp(-weight * rigidity)
private fun confidence(
relationCode: Int,
sourceIsObserver: Boolean,
): Double =
when (relationCode) {
TrustRelation.FOLLOW.code -> if (sourceIsObserver) params.directFollowConfidence else params.indirectFollowConfidence
TrustRelation.MUTE.code -> params.muteConfidence
else -> params.reportConfidence
}
private fun rating(relationCode: Int): Double =
when (relationCode) {
TrustRelation.FOLLOW.code -> TrustRelation.FOLLOW.rating
TrustRelation.MUTE.code -> TrustRelation.MUTE.rating
else -> TrustRelation.REPORT.rating
}
/**
* Score every node reachable from [observer]. Returns scores by node id, or an
* all-zero array if the observer isn't in the graph. [onProgress] fires once
* per sweep with `(totalNodeUpdates, nodesStillMoving)` — the second value is
* how many nodes moved more than [GrapeRankParams.convergence] this sweep, so
* it trends to 0 as the graph settles.
*
* Iterates synchronous **Gauss-Seidel** sweeps over every node, updating scores
* in place so a value computed earlier in a sweep is already visible to nodes
* later in the same sweep (this converges faster than a double-buffered Jacobi
* pass). A sweep that moves no node by more than the convergence delta ends the
* loop — the same per-node threshold and fixed point as NosFabrica's Brainstorm
* reference. On a dense graph this is far less total work than a
* change-propagating worklist: a worklist re-visits a node once per rater whose
* score nudges, so its cost scales with the in-degree of the churning core,
* whereas a sweep touches each node exactly once per iteration. Attenuation < 1
* makes the update a contraction, so the fixed point is unique regardless of
* sweep order; ids run in roughly BFS order from the observer, which lets
* trust flow outward within a single sweep and keeps the iteration count low.
*
* Nodes unreachable from the observer settle to 0 for free: all of their raters
* stay at 0, so the inner loop's `sourceScore != 0.0` guard skips every edge.
*/
fun compute(
graph: TrustGraph,
observer: HexKey,
onProgress: ((visited: Long, queued: Int) -> Unit)? = null,
): DoubleArray {
val n = graph.nodeCount
val scores = DoubleArray(n)
val observerId = graph.idOf(observer)
if (observerId < 0) return scores
scores[observerId] = 1.0
val attenuation = params.attenuation
val convergence = params.convergence
val inOffsets = graph.inOffsets
val inPacked = graph.inPacked
var visited = 0L
while (true) {
var stillMoving = 0
var target = 0
while (target < n) {
if (target != observerId) {
var sumOfWeights = 0.0
var sumOfWeightedRatings = 0.0
var i = inOffsets[target]
val end = inOffsets[target + 1]
while (i < end) {
val packed = inPacked[i]
val source = packed and TrustGraph.SOURCE_MASK
val sourceScore = scores[source]
if (sourceScore != 0.0) {
val relationCode = packed ushr TrustGraph.SOURCE_BITS
val weight = confidence(relationCode, source == observerId) * sourceScore * attenuation
sumOfWeights += weight
sumOfWeightedRatings += weight * rating(relationCode)
}
i++
}
val newScore =
if (abs(sumOfWeights) < 0.00001) {
0.0
} else {
val s = weightToConfidence(sumOfWeights) * sumOfWeightedRatings / sumOfWeights
if (s > 0.0) s else 0.0
}
val oldScore = scores[target]
if (newScore != oldScore) {
scores[target] = newScore
if (abs(newScore - oldScore) > convergence) stillMoving++
}
visited++
}
target++
}
onProgress?.invoke(visited, stillMoving)
if (stillMoving == 0) break
}
return scores
}
}
@@ -0,0 +1,207 @@
/*
* Copyright (c) 2025 Vitor Pamplona
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of
* this software and associated documentation files (the "Software"), to deal in
* the Software without restriction, including without limitation the rights to use,
* copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
* Software, and to permit persons to whom the Software is furnished to do so,
* subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
* AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
package com.vitorpamplona.quartz.experimental.graperank
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.core.HexKey
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
import com.vitorpamplona.quartz.nip01Core.signers.NostrSigner
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 kotlinx.coroutines.async
import kotlinx.coroutines.awaitAll
import kotlinx.coroutines.coroutineScope
/**
* 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.
*
* 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 [GrapeRankDataCrawler]: 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).
*/
class GrapeRankPublisher(
private val store: IEventStore,
private val publish: suspend (Event, Set<NormalizedRelayUrl>) -> Map<NormalizedRelayUrl, Boolean>,
) {
/** 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,
)
/**
* 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.
*/
suspend fun reconcileAndPublish(
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).
val existing = existingCards(providerPubkey)
val publishableTargets = 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.
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()
val (ok, rejected) = publishCards(providerSigner, toUpsert, relays, publishConcurrency)
val (deleted, deleteRejected) = publishDeletions(providerSigner, toDelete, relays)
return Result(
published = ok,
publishRejected = rejected,
deleted = deleted,
deleteRejected = deleteRejected,
skippedUnchanged = scored.size - changed.size,
truncated = (changed.size - toUpsert.size).coerceAtLeast(0),
)
}
/**
* 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).
*/
private suspend fun existingCards(providerPubkey: HexKey): Map<HexKey, ContactCardEvent> =
store
.query<Event>(Filter(kinds = listOf(ContactCardEvent.KIND), authors = listOf(providerPubkey)))
.filterIsInstance<ContactCardEvent>()
.groupBy { it.aboutUser() }
.mapNotNull { (target, cards) ->
val t = target ?: return@mapNotNull null
t to (cards.maxByOrNull { it.createdAt } ?: return@mapNotNull null)
}.toMap()
/**
* 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
* would differ.
*/
private fun rankTagValue(card: ContactCardEvent): String? =
card.tags.firstNotNullOfOrNull { tag ->
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(
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 =
coroutineScope {
batch
.map { (pubkey, rank) ->
async {
val card =
ContactCardEvent.create(
targetUser = pubkey,
signer = signer,
publicInitializer = { add(RankTag.assemble(rank)) },
)
publish(card, relays)
}
}.awaitAll()
}
for (ack in acks) {
if (ack.values.any { it }) published++ else rejected++
}
}
return published to rejected
}
/**
* 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.
*/
private suspend fun publishDeletions(
signer: NostrSigner,
cards: List<ContactCardEvent>,
relays: Set<NormalizedRelayUrl>,
): Pair<Int, Int> {
if (cards.isEmpty()) return 0 to 0
var deleted = 0
var rejected = 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
}
return deleted to rejected
}
companion object {
/** Concurrent card publishes when upserting. */
const val PUBLISH_CONCURRENCY = 16
/**
* Addressable coordinates cited per kind:5 retraction. Each `a` tag is
* ~130 bytes (30382:<64hex>:<64hex>), so 400 keeps the whole event ~52KB —
* under the 64KB event-size cap many relays enforce (stricter than the
* 256KB message cap).
*/
const val DELETE_PER_EVENT = 400
}
}
@@ -0,0 +1,101 @@
/*
* 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.HexKey
/**
* A trust relationship kind and the GrapeRank rating it carries. [code] is the
* 2-bit tag packed alongside a source node id in the edge arrays (see
* [TrustGraph]); keep it in `0..3`.
*/
enum class TrustRelation(
val rating: Double,
val code: Int,
) {
FOLLOW(1.0, 0),
MUTE(-0.1, 1),
REPORT(-0.1, 2),
}
/**
* A web-of-trust graph over Nostr pubkeys, stored compactly so it scales to the
* whole network (millions of edges) without a `String`-keyed edge object per
* relationship.
*
* Pubkeys are interned to dense `Int` node ids. Edges live in two
* compressed-sparse-row (CSR) layouts backed by flat `IntArray`s — one indexed
* by target (what [GrapeRank] reads to score a node) and one by source (what the
* propagation worklist follows). Each incoming entry packs the source id in the
* low 29 bits and the [TrustRelation.code] in the top bits, so an edge is a
* single `int`. A 100M-edge graph is then ~0.8 GB of primitive arrays instead of
* tens of GB of objects.
*
* Build one with [TrustGraphBuilder], feeding contact lists / mutes / reports in
* as they stream off the relays.
*/
class TrustGraph internal constructor(
val nodeCount: Int,
private val pubkeys: Array<HexKey>,
private val ids: HashMap<HexKey, Int>,
// CSR by target: incoming edges of node t are inPacked[inOffsets[t] until inOffsets[t+1]],
// each packing source id (low 29 bits) + relation code (top bits).
internal val inOffsets: IntArray,
internal val inPacked: IntArray,
// CSR by source: out-neighbour targets of node s are outTargets[outOffsets[s] until outOffsets[s+1]].
internal val outOffsets: IntArray,
internal val outTargets: IntArray,
) {
/** Node id for [pubkey], or `-1` if it never appeared in the graph. */
fun idOf(pubkey: HexKey): Int = ids[pubkey] ?: -1
/** Pubkey for a node [id]. */
fun pubkeyOf(id: Int): HexKey = pubkeys[id]
fun edgeCount(): Int = inPacked.size
companion object {
const val SOURCE_BITS = 29
const val SOURCE_MASK = (1 shl SOURCE_BITS) - 1
const val MAX_NODES = SOURCE_MASK // ids must fit in the low 29 bits
}
}
/** A minimal growable `int[]` — avoids boxing `Int`s in an `ArrayList` at graph scale. */
internal class IntArrayList(
initialCapacity: Int = 16,
) {
var data: IntArray = IntArray(initialCapacity.coerceAtLeast(1))
private set
var size: Int = 0
private set
fun add(value: Int) {
if (size == data.size) data = data.copyOf(data.size * 2)
data[size++] = value
}
fun get(index: Int): Int = data[index]
fun removeLast(): Int = data[--size]
fun isNotEmpty(): Boolean = size > 0
}
@@ -0,0 +1,127 @@
/*
* 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.HexKey
/**
* Builds a [TrustGraph] incrementally so callers never have to hold every contact
* list in memory at once — feed each user's follows / mutes / reports as they
* stream off the relays (or out of the store), then call [build].
*
* Interns pubkeys to dense ids on the fly and accumulates edges in flat growable
* int arrays. Follows and mutes are replaceable (one list per author, deduped by
* the caller via latest-per-author + set-valued tags); reports are regular events,
* so `(reporter → reported)` report edges are deduped here. Self-edges are dropped.
*/
class TrustGraphBuilder {
private val ids = HashMap<HexKey, Int>()
private val pubkeys = ArrayList<HexKey>()
// Parallel edge arrays: edge i is source edgeSource[i] --relation--> edgeTarget[i],
// with the relation packed into the top bits of edgeSource[i].
private val edgeTargets = IntArrayList()
private val edgeSourcesPacked = IntArrayList()
// Dedup for report edges only (reporters can file many kind:1984 for one target).
private val reportSeen = HashSet<Long>()
private fun intern(pubkey: HexKey): Int =
ids.getOrPut(pubkey) {
val id = pubkeys.size
pubkeys.add(pubkey)
id
}
private fun addEdge(
source: HexKey,
target: HexKey,
relation: TrustRelation,
) {
if (source == target) return
val s = intern(source)
val t = intern(target)
if (relation == TrustRelation.REPORT) {
val key = (s.toLong() shl 32) or (t.toLong() and 0xFFFFFFFFL)
if (!reportSeen.add(key)) return
}
edgeTargets.add(t)
edgeSourcesPacked.add(s or (relation.code shl TrustGraph.SOURCE_BITS))
}
fun addFollows(
source: HexKey,
follows: Iterable<HexKey>,
) {
for (target in follows) addEdge(source, target, TrustRelation.FOLLOW)
}
fun addMutes(
source: HexKey,
muted: Iterable<HexKey>,
) {
for (target in muted) addEdge(source, target, TrustRelation.MUTE)
}
fun addReports(
source: HexKey,
reported: Iterable<HexKey>,
) {
for (target in reported) addEdge(source, target, TrustRelation.REPORT)
}
fun nodeCount(): Int = pubkeys.size
fun edgeCount(): Int = edgeTargets.size
/** Freeze the accumulated edges into the two CSR layouts. */
fun build(): TrustGraph {
val n = pubkeys.size
val m = edgeTargets.size
// Incoming CSR (by target).
val inOffsets = IntArray(n + 1)
for (i in 0 until m) inOffsets[edgeTargets.get(i) + 1]++
for (i in 1..n) inOffsets[i] += inOffsets[i - 1]
val inPacked = IntArray(m)
val inCursor = inOffsets.copyOf()
for (i in 0 until m) {
val t = edgeTargets.get(i)
inPacked[inCursor[t]++] = edgeSourcesPacked.get(i)
}
// Outgoing CSR (by source).
val outOffsets = IntArray(n + 1)
for (i in 0 until m) {
val s = edgeSourcesPacked.get(i) and TrustGraph.SOURCE_MASK
outOffsets[s + 1]++
}
for (i in 1..n) outOffsets[i] += outOffsets[i - 1]
val outTargets = IntArray(m)
val outCursor = outOffsets.copyOf()
for (i in 0 until m) {
val s = edgeSourcesPacked.get(i) and TrustGraph.SOURCE_MASK
outTargets[outCursor[s]++] = edgeTargets.get(i)
}
return TrustGraph(n, pubkeys.toTypedArray(), ids, inOffsets, inPacked, outOffsets, outTargets)
}
}
@@ -0,0 +1,275 @@
/*
* 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.nip01Core.relay.client.accessories
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.ClosedMessage
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.NoticeMessage
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
import com.vitorpamplona.quartz.utils.Log
import com.vitorpamplona.quartz.utils.TimeUtils
import com.vitorpamplona.quartz.utils.concurrent.ConcurrentMap
import kotlinx.coroutines.CompletableDeferred
import kotlinx.coroutines.delay
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
import kotlin.concurrent.atomics.AtomicInt
import kotlin.concurrent.atomics.AtomicLong
import kotlin.concurrent.atomics.ExperimentalAtomicApi
/**
* Adaptive per-relay back-pressure with TWO independent controls, because relays
* push back for two different reasons that need two different responses:
*
* 1. **Subscription-count limit** — a max on how many subscriptions may be OPEN
* at once ("too many subscriptions", "maximum concurrent subscription count",
* "number of subscriptions exceeds limit"). The fix is fewer *concurrent*
* subs, so we demote the relay's concurrency cap down [subLadder]
* (100 → 20 → 10).
* 2. **Rate limit** — too many subscription *changes per second* ("rate-limited:
* too many messages", "burst exhausted", "slow down"). Fewer concurrent subs
* wouldn't help; the fix is to *space the REQs out in time*, so we impose a
* minimum interval between opens to that relay, growing it up [rateLadder]
* (250ms → 500ms → 1s → 2s).
*
* Mixing the two mishandles the relay: capping concurrency does nothing for a
* rate limit, and slowing the rate does nothing for a subscription-count cap. So
* each complaint is routed to its own actuator by matching the notice text.
*
* A well-behaved relay starts at [startCap] concurrent subs with no rate delay,
* and only the ones that push back get throttled — each only as far, and in the
* dimension, they keep pushing.
*
* Registered as a [RelayConnectionListener] on the shared client, so both signals
* are driven straight off the incoming NOTICE/CLOSED frames (which fire on the
* per-relay socket threads — all state here is concurrent). Drains gate through
* [withPermit]: the gated-drain path holds a relay's permit for the lifetime of
* that relay's subscription, and passes the rate gate before it opens, so we
* respect both limits at once.
*/
@OptIn(ExperimentalAtomicApi::class)
class AdaptiveRelayLimiter(
private val startCap: Int = 100,
private val subLadder: List<Int> = listOf(20, 10),
private val rateLadder: List<Long> = listOf(250L, 500L, 1000L, 2000L),
) : RelayConnectionListener {
private val gates = ConcurrentMap<NormalizedRelayUrl, Gate>()
// Concurrency-cap demotions per relay (== index+1 into subLadder). Capped at
// subLadder.size: past the floor we stop demoting.
private val subDemotions = ConcurrentMap<NormalizedRelayUrl, Int>()
// Rate-limit state per relay: how far down rateLadder we've stepped, the
// current min interval between opens, and the next epoch-ms an open may fire.
private val rateSteps = ConcurrentMap<NormalizedRelayUrl, Int>()
private val rateDelayMs = ConcurrentMap<NormalizedRelayUrl, Long>()
private val nextAllowedAtMs = ConcurrentMap<NormalizedRelayUrl, AtomicLong>()
private fun gate(relay: NormalizedRelayUrl): Gate = gates.getOrPut(relay) { Gate(startCap) }
/**
* Run [block] against [relay] respecting both limits: first wait out any rate
* delay (spacing opens in time), then hold one of the relay's concurrency
* permits for the duration.
*/
suspend fun <T> withPermit(
relay: NormalizedRelayUrl,
block: suspend () -> T,
): T {
rateGate(relay)
val g = gate(relay)
g.acquire()
try {
return block()
} finally {
g.release()
}
}
/** If [relay] is rate-limited, reserve and wait for its next allowed open slot. */
private suspend fun rateGate(relay: NormalizedRelayUrl) {
val delayMs = rateDelayMs[relay] ?: return
if (delayMs <= 0L) return
val now = TimeUtils.nowMillis()
// Atomically claim the next slot: my turn is max(prevSlot, now); the next
// caller can't fire until delayMs after me. Serializes opens to this relay
// at one per delayMs, in arrival order.
val slot = nextAllowedAtMs.getOrPut(relay) { AtomicLong(now) }
var myTurn: Long
while (true) {
val prev = slot.load()
myTurn = maxOf(prev, now)
if (slot.compareAndSet(prev, myTurn + delayMs)) break
}
val wait = myTurn - now
if (wait > 0) delay(wait)
}
override fun onIncomingMessage(
relay: IRelayClient,
msgStr: String,
msg: Message,
) {
val text =
when (msg) {
is ClosedMessage -> msg.message
is NoticeMessage -> msg.message
else -> return
}
val t = text.lowercase()
// Route each complaint to the matching actuator. Not mutually exclusive:
// if a relay somehow reports both, we act on both (they don't conflict).
if (RATE_LIMIT_MARKERS.any { it in t }) throttleRate(relay.url)
if (SUB_LIMIT_MARKERS.any { it in t }) demoteConcurrency(relay.url)
}
/** Step [relay] one rung down the concurrency-cap ladder, unless already at the floor. */
private fun demoteConcurrency(relay: NormalizedRelayUrl) {
if ((subDemotions[relay] ?: 0) >= subLadder.size) return
val step = subDemotions.merge(relay, 1) { a, b -> a + b }
val cap = subLadder[(step - 1).coerceIn(0, subLadder.size - 1)]
gate(relay).lower(cap)
if (step <= subLadder.size) {
Log.d("AdaptiveRelayLimiter") { "${relay.url} concurrency capped at $cap subs (sub-limit #$step)" }
}
}
/** Step [relay] one rung down the rate ladder, unless already at the slowest. */
private fun throttleRate(relay: NormalizedRelayUrl) {
if ((rateSteps[relay] ?: 0) >= rateLadder.size) return
val step = rateSteps.merge(relay, 1) { a, b -> a + b }
val d = rateLadder[(step - 1).coerceIn(0, rateLadder.size - 1)]
rateDelayMs[relay] = d
if (step <= rateLadder.size) {
Log.d("AdaptiveRelayLimiter") { "${relay.url} rate-throttled to 1 REQ / ${d}ms (rate-limit #$step)" }
}
}
/** JSON-friendly view of which relays we throttled, in which dimension, how far. */
fun snapshot(): Map<String, Any?> {
val capCounts = HashMap<Int, Int>()
for ((_, step) in subDemotions.snapshot()) {
val cap = subLadder[(step - 1).coerceIn(0, subLadder.size - 1)]
capCounts[cap] = (capCounts[cap] ?: 0) + 1
}
val cappedAt = capCounts.toList().sortedBy { it.first }.toMap()
val rateCounts = HashMap<Long, Int>()
for ((_, step) in rateSteps.snapshot()) {
val d = rateLadder[(step - 1).coerceIn(0, rateLadder.size - 1)]
rateCounts[d] = (rateCounts[d] ?: 0) + 1
}
val rateAt = rateCounts.toList().sortedBy { it.first }.toMap()
return mapOf(
"start_cap" to startCap,
"sub_ladder" to subLadder,
"rate_ladder_ms" to rateLadder,
"concurrency_capped_relays" to subDemotions.size(),
"concurrency_capped_at" to cappedAt,
"rate_limited_relays" to rateSteps.size(),
"rate_limited_at_ms" to rateAt,
)
}
fun hadThrottling(): Boolean = subDemotions.size() > 0 || rateSteps.size() > 0
/**
* A bounded-concurrency gate whose limit can only ever be *lowered* (relays
* never earn their cap back within a run). Fair FIFO hand-off: a released
* permit goes to the longest-waiting acquirer. Lowering the limit below the
* in-use count doesn't cancel live holders — it just refuses to admit new
* ones until enough release that `inUse < limit` again, so the concurrency
* converges down to the new cap as the excess subscriptions finish.
*/
private class Gate(
initialLimit: Int,
) {
private val limit = AtomicInt(initialLimit)
private val mutex = Mutex()
private var inUse = 0
private val waiters = ArrayDeque<CompletableDeferred<Unit>>()
suspend fun acquire() {
val wait =
mutex.withLock {
if (inUse < limit.load()) {
inUse++
null
} else {
CompletableDeferred<Unit>().also { waiters.addLast(it) }
}
}
wait?.await()
}
suspend fun release() {
mutex.withLock {
inUse--
while (inUse < limit.load() && waiters.isNotEmpty()) {
waiters.removeFirst().complete(Unit)
inUse++
}
}
}
/** Monotonically shrink the cap. Safe to call from any thread. */
fun lower(newLimit: Int) {
while (true) {
val cur = limit.load()
if (newLimit >= cur) return
if (limit.compareAndSet(cur, newLimit)) return
}
}
}
companion object {
// A cap on how many subscriptions may be OPEN at once. Fix: fewer
// concurrent subs (demote the concurrency cap).
private val SUB_LIMIT_MARKERS =
listOf(
"too many concurrent",
"concurrent req",
"too many subscription",
"number of subscriptions",
"subscriptions exceeds",
"subscription limit",
"subscription count",
"maximum concurrent subscription",
"max subscription",
"too many req",
)
// Too many subscription CHANGES per second. Fix: space the REQs out in
// time (a per-relay min interval), not fewer concurrent subs.
private val RATE_LIMIT_MARKERS =
listOf(
"rate-limit",
"rate limit",
"ratelimit",
"too many messages",
"too many requests",
"burst exhausted",
"throttl",
"slow down",
)
}
}
@@ -0,0 +1,77 @@
/*
* 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.nip01Core.relay.client.accessories
/**
* Why a relay could not be used for a one-shot drain — when the reason is worth
* acting on (dropping the relay from further routing).
*
* - [HARD]: the relay answered wrong, or cannot exist. A bad HTTP upgrade (not a
* websocket / dead status code), an unresolvable domain, or a TLS misconfig.
* This will not fix itself, so one strike is enough to drop it.
* - [TRANSIENT]: a failure that might clear — connection refused / reset, host
* unreachable, or a temporary 429/5xx on the upgrade. Struck a few times
* before we give up.
*
* A pure connect **timeout** is neither. The relay is most likely just busy, so
* we retry it and never mark it dead — [classifyDrainFailure] returns null for
* it (and for any non-failure terminal reason).
*/
enum class DrainFailure { HARD, TRANSIENT }
/**
* Classify a drain per-relay terminal reason. Returns null when the relay should
* simply be retried (a timeout, or a non-failure like eose/closed). The reason
* shape is `cannot:<message>` for a connect failure (see
* `BasicRelayClient.onCannotConnect`), or `eose` / `closed:…` / `timeout`.
*/
fun classifyDrainFailure(reason: String): DrainFailure? {
if (!reason.startsWith("cannot")) return null
val m = reason.removePrefix("cannot:").lowercase()
// The message now carries the exception class name (see BasicRelayClient), so
// we can key on the stable *type* rather than localized message text.
// Busy, not dead: a connect/read timeout means the handshake just didn't
// finish in time. Retry it — the relay is probably fine, only slow or loaded.
if ("timeout" in m || "timed out" in m) return null // SocketTimeoutException, etc.
// Cannot ever work: unresolvable domain (DNS) or a TLS misconfiguration.
// Dead for good — one strike is enough.
if ("unknownhost" in m || // UnknownHostException
"unable to resolve host" in m ||
"no address associated" in m ||
"nodename nor servname" in m ||
"sslhandshake" in m || // SSLHandshakeException
"sslpeerunverified" in m ||
"sslexception" in m ||
"certificate" in m || // CertificateException
"trust anchor" in m ||
"certpath" in m
) {
return DrainFailure.HARD
}
// Wrong HTTP upgrade. Usually a misconfigured endpoint (not a relay), but
// 429 / 5xx mean "busy, come back later", so those stay transient.
if ("server misconfigured" in m || "not a websocket" in m || "expected http 101" in m) {
val transientCode = Regex("response: (429|500|502|503|504)").containsMatchIn(m)
return if (transientCode) DrainFailure.TRANSIENT else DrainFailure.HARD
}
// Refused / reset / unreachable / anything else: might clear — retry a few times.
return DrainFailure.TRANSIENT
}
@@ -136,7 +136,9 @@ open class BasicRelayClient(
socket?.connect()
} catch (e: Exception) {
if (e is CancellationException) throw e
listener.onCannotConnect(this, "Error when trying to connect: ${e.message ?: e::class.simpleName}")
val typeName = e::class.simpleName
val detail = e.message?.let { "$it ($typeName)" } ?: (typeName ?: "unknown error")
listener.onCannotConnect(this, "Error when trying to connect: $detail")
listener.onDisconnected(this)
dontTryAgainForALongTime()
markConnectionAsClosed()
@@ -187,9 +189,15 @@ open class BasicRelayClient(
} else {
socket?.disconnect()
// suppression rules below must match the raw message; displayMsg is for listener output only
// suppression rules below must match the raw message; displayMsg is for listener output only.
// Always include the exception's class name: message text is
// localized and inconsistent across platforms, but the type
// (SocketTimeoutException / UnknownHostException / SSLHandshakeException /
// ConnectException …) is stable and lets listeners classify a failure
// reliably — a busy relay (timeout) vs a dead one (bad domain / TLS).
val msg = t.message
val displayMsg = msg ?: t::class.simpleName
val typeName = t::class.simpleName
val displayMsg = if (msg != null) "$msg ($typeName)" else (typeName ?: "unknown error")
// checks if this is an actual failure. Closing the socket generates an onFailure as well.
// ignore tor errors.
@@ -156,7 +156,7 @@ class RelayUrlNormalizer {
if (trimmed.contains("://")) {
// some other scheme we cannot connect to.
Log.w("RelayUrlNormalizer") { "Rejected $url" }
Log.d("RelayUrlNormalizer") { "Rejected $url" }
return null
}
@@ -189,14 +189,14 @@ class RelayUrlNormalizer {
normalizedUrls.put(url, NormalizationResult.Success(normalized))
normalized
} else {
Log.w("NormalizedRelayUrl") { "Rejected $url" }
Log.d("NormalizedRelayUrl") { "Rejected $url" }
normalizedUrls.put(url, NormalizationResult.Error)
null
}
} catch (e: Exception) {
if (e is CancellationException) throw e
normalizedUrls.put(url, NormalizationResult.Error)
Log.w("NormalizedRelayUrl") { "Rejected $url" }
Log.d("NormalizedRelayUrl") { "Rejected $url" }
null
}
}
@@ -0,0 +1,55 @@
/*
* 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.nip01Core.store
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.crypto.verify
import com.vitorpamplona.quartz.utils.Log
/**
* Verify [event]'s NIP-01 id + signature and, if valid, persist it to this store.
* Returns `true` when the event was accepted (verified) — even if the insert was a
* no-op — so callers can gate "surface this event" on the return.
*
* A UNIQUE-constraint rejection is normal, not a failure: the store already holds
* this id, or a newer version of a replaceable (kind 0/3/10000-19999). The outbox
* model routinely delivers the same event from several of a user's write relays, so
* a crawl produces these by the hundred-thousand — so only genuine persistence
* failures (I/O, full disk, corruption) are logged. Persistence is best-effort: an
* insert error is swallowed, not propagated, so it can't break a live subscription.
*
* This is the single verify-then-store sink every event-arrival path should funnel
* through, so the store stays the authoritative cache of what has been seen.
*/
suspend fun IEventStore.verifyAndInsert(event: Event): Boolean {
if (!event.verify()) {
Log.w("EventStore") { "dropped event ${event.id.take(8)} kind=${event.kind} — bad signature" }
return false
}
try {
insert(event)
} catch (t: Throwable) {
if (t.message?.contains("UNIQUE constraint", ignoreCase = true) != true) {
Log.w("EventStore") { "store insert failed for ${event.id.take(8)}: ${t.message}" }
}
}
return true
}
@@ -0,0 +1,68 @@
/*
* 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.utils.concurrent
/**
* A thread-safe hash map whose compound operations — [getOrPut] and [merge] —
* apply their update **atomically**, not merely one-lock-per-primitive-op. This
* is the contract a concurrent producer/consumer pipeline needs: two coroutines
* racing `getOrPut` on the same key must agree on a single value, and racing
* `merge` must not lose an increment.
*
* commonMain has no `java.util.concurrent.ConcurrentHashMap`, so this is
* expect/actual, matching the split already used by [com.vitorpamplona.quartz.utils.cache.ConcurrentHashCache]:
* - JVM / Android → `ConcurrentHashMap` (lock-free, true atomic `computeIfAbsent` / `merge`).
* - Native (Apple + Linux) → copy-on-write over an atomic reference, with a
* CAS retry loop giving the same atomicity. Correct but O(n)-per-write; the
* native targets never run the heavy crawl this backs, they only compile it.
*
* Only the operations the crawl actually uses are exposed — no full [MutableMap]
* surface — so the native copy-on-write actual stays small and obviously correct.
*/
expect class ConcurrentMap<K : Any, V : Any>() {
operator fun get(key: K): V?
operator fun set(
key: K,
value: V,
)
/** Atomically return the value for [key], computing and inserting [defaultValue] once if absent. */
fun getOrPut(
key: K,
defaultValue: () -> V,
): V
/**
* Atomically insert [value] if [key] is absent, else replace the existing
* value with `remap(existing, value)`. Returns the value now stored.
*/
fun merge(
key: K,
value: V,
remap: (old: V, new: V) -> V,
): V
fun size(): Int
/** A point-in-time copy of the entries — safe to iterate without holding a lock. */
fun snapshot(): Map<K, V>
}
@@ -0,0 +1,43 @@
/*
* 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.utils.concurrent
/**
* A thread-safe hash set for the crawl's cross-coroutine membership tracking
* (dead relays struck by drain workers while the router reads them, relay hints
* written by the ingest consumer while the producer reads them).
*
* commonMain has no `java.util.concurrent.ConcurrentHashMap.newKeySet()`, so this
* is expect/actual with the same JVM-vs-native split as [ConcurrentMap]:
* - JVM / Android → `ConcurrentHashMap.newKeySet()`.
* - Native → copy-on-write over an atomic reference (compile-only, never the hot path).
*/
expect class ConcurrentSet<E : Any>() {
/** Add [element]; returns true if it was not already present. */
fun add(element: E): Boolean
operator fun contains(element: E): Boolean
fun size(): Int
/** A point-in-time copy — safe to iterate or diff against without a lock. */
fun snapshot(): Set<E>
}
@@ -0,0 +1,246 @@
/*
* 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.HexKey
import kotlin.math.abs
import kotlin.math.exp
import kotlin.math.ln
import kotlin.math.max
import kotlin.random.Random
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertTrue
class GrapeRankTest {
private val obs = "observer"
private fun graphOf(edges: List<Triple<HexKey, HexKey, TrustRelation>>): TrustGraph {
val b = TrustGraphBuilder()
for ((source, target, relation) in edges) {
when (relation) {
TrustRelation.FOLLOW -> b.addFollows(source, listOf(target))
TrustRelation.MUTE -> b.addMutes(source, listOf(target))
TrustRelation.REPORT -> b.addReports(source, listOf(target))
}
}
return b.build()
}
private fun graphOf(vararg edges: Triple<HexKey, HexKey, TrustRelation>) = graphOf(edges.toList())
/** Score for a pubkey (0.0 if absent from the graph). */
private fun DoubleArray.of(
graph: TrustGraph,
pubkey: HexKey,
): Double {
val id = graph.idOf(pubkey)
return if (id < 0) 0.0 else this[id]
}
@Test
fun observerIsPinnedAtFullSelfTrust() {
val graph = graphOf(Triple(obs, "a", TrustRelation.FOLLOW))
val scores = GrapeRank().compute(graph, obs)
assertEquals(1.0, scores.of(graph, obs), 1e-12)
}
@Test
fun directFollowMatchesHandComputedValue() {
val graph = graphOf(Triple(obs, "a", TrustRelation.FOLLOW))
val scores = GrapeRank().compute(graph, obs)
// weight = 0.5 * 1.0 * 0.85 = 0.425 ; score = conf(0.425) = 0.2551612...
assertEquals(0.25516127, scores.of(graph, "a"), 1e-6)
}
@Test
fun trustDecaysSteeplyAcrossHops() {
val graph =
graphOf(
Triple(obs, "a", TrustRelation.FOLLOW),
Triple("a", "b", TrustRelation.FOLLOW),
)
val scores = GrapeRank().compute(graph, obs)
val a = scores.of(graph, "a")
val b = scores.of(graph, "b")
assertEquals(0.004499, b, 1e-5)
assertTrue(b < a / 10.0, "two-hop trust should be far below one-hop trust")
}
@Test
fun aMuteFromAnEndorsedUserLowersTheScore() {
val followOnlyGraph = graphOf(Triple(obs, "b", TrustRelation.FOLLOW))
val followOnly = GrapeRank().compute(followOnlyGraph, obs).of(followOnlyGraph, "b")
val muteGraph =
graphOf(
Triple(obs, "a", TrustRelation.FOLLOW),
Triple(obs, "b", TrustRelation.FOLLOW),
Triple("a", "b", TrustRelation.MUTE),
)
val withMute = GrapeRank().compute(muteGraph, obs).of(muteGraph, "b")
assertTrue(withMute < followOnly, "a mute from a trusted user should pull b below the follow-only baseline")
}
@Test
fun purelyReportedUserFloorsAtZero() {
val graph =
graphOf(
Triple(obs, "a", TrustRelation.FOLLOW),
Triple("a", "d", TrustRelation.REPORT),
)
val scores = GrapeRank().compute(graph, obs)
assertEquals(0.0, scores.of(graph, "d"), 1e-9)
}
@Test
fun unreachableUsersAreNotScored() {
val graph =
graphOf(
Triple(obs, "a", TrustRelation.FOLLOW),
Triple("x", "y", TrustRelation.FOLLOW),
)
val scores = GrapeRank().compute(graph, obs)
assertTrue(scores.of(graph, "a") > 0.0)
assertEquals(0.0, scores.of(graph, "y"), 1e-12, "a user with no path from the observer stays 0")
}
@Test
fun cyclesConverge() {
val graph =
graphOf(
Triple(obs, "a", TrustRelation.FOLLOW),
Triple("a", "b", TrustRelation.FOLLOW),
Triple("b", "a", TrustRelation.FOLLOW),
)
val scores = GrapeRank().compute(graph, obs)
assertTrue(scores.of(graph, "a") > 0.0)
assertTrue(scores.of(graph, "b") > 0.0)
}
@Test
fun deduplicatesRepeatedReportEdges() {
// Two report edges a->d collapse to one; the score matches a single report.
val once = graphOf(Triple(obs, "a", TrustRelation.FOLLOW), Triple("a", "d", TrustRelation.REPORT))
val twice =
graphOf(
Triple(obs, "a", TrustRelation.FOLLOW),
Triple("a", "d", TrustRelation.REPORT),
Triple("a", "d", TrustRelation.REPORT),
)
assertEquals(2, twice.edgeCount(), "duplicate report edge should be dropped")
assertEquals(
GrapeRank().compute(once, obs).of(once, "d"),
GrapeRank().compute(twice, obs).of(twice, "d"),
1e-12,
)
}
/**
* Adversarial cross-check: the worklist propagation must reach the same fixed
* point as a naive full-sweep (the reference `v1FullSweep`) on random graphs.
*/
@Test
fun worklistMatchesFullSweepOnRandomGraphs() {
val params = GrapeRankParams(convergence = 1e-10)
val engine = GrapeRank(params)
repeat(50) { seed ->
val rng = Random(seed)
val n = 3 + rng.nextInt(12)
val nodes = (0 until n).map { "u$it" }
val edges = ArrayList<Triple<HexKey, HexKey, TrustRelation>>()
for (src in nodes) {
for (dst in nodes) {
if (src == dst) continue
if (rng.nextDouble() < 0.25) {
val relation =
when (rng.nextInt(5)) {
0 -> TrustRelation.MUTE
1 -> TrustRelation.REPORT
else -> TrustRelation.FOLLOW
}
edges.add(Triple(src, dst, relation))
}
}
}
val observer = nodes.first()
val graph = graphOf(edges)
val scores = engine.compute(graph, observer)
val reference = fullSweep(edges, nodes, observer, params)
for (node in nodes) {
if (node == observer) continue // observer self-trust is not part of a ranking
val a = scores.of(graph, node)
val b = reference[node] ?: 0.0
assertEquals(b, a, 1e-5, "seed=$seed node=$node worklist=$a fullSweep=$b")
}
}
}
// Reference: blind full sweep over every user until nothing changes.
private fun fullSweep(
edges: List<Triple<HexKey, HexKey, TrustRelation>>,
nodes: List<HexKey>,
observer: HexKey,
params: GrapeRankParams,
): Map<HexKey, Double> {
// Dedup identical edges (mirrors the builder: report edges dedup; follow/mute
// sets are unique per source anyway).
val incoming = HashMap<HexKey, MutableSet<Pair<HexKey, TrustRelation>>>()
for ((s, t, r) in edges) {
if (s == t) continue
incoming.getOrPut(t) { LinkedHashSet() }.add(s to r)
}
fun confidence(
r: TrustRelation,
source: HexKey,
) = when (r) {
TrustRelation.FOLLOW -> if (source == observer) params.directFollowConfidence else params.indirectFollowConfidence
TrustRelation.MUTE -> params.muteConfidence
TrustRelation.REPORT -> params.reportConfidence
}
fun weightToConfidence(w: Double) = 1.0 - exp(-w * -ln(params.rigor))
val scores = HashMap<HexKey, Double>()
scores[observer] = 1.0
do {
var changed = false
for (target in nodes) {
if (target == observer) continue
var sumW = 0.0
var sumWR = 0.0
for ((source, r) in incoming[target] ?: emptySet()) {
val s = scores[source] ?: continue
val w = confidence(r, source) * s * params.attenuation
sumW += w
sumWR += w * r.rating
}
val newScore = if (abs(sumW) < 0.00001) 0.0 else max(weightToConfidence(sumW) * sumWR / sumW, 0.0)
val old = scores.put(target, newScore) ?: 0.0
changed = changed || abs(newScore - old) > params.convergence
}
} while (changed)
return scores
}
}
@@ -0,0 +1,107 @@
/*
* 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.HexKey
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertTrue
class TrustGraphBuilderTest {
private val alice = "alice"
private val bob = "bob"
private val carol = "carol"
private val dave = "dave"
/** Decode a node's incoming edges back to (source, relation) pairs from the CSR. */
private fun TrustGraph.incomingOf(pubkey: HexKey): Set<Pair<HexKey, TrustRelation>> {
val t = idOf(pubkey)
if (t < 0) return emptySet()
val out = HashSet<Pair<HexKey, TrustRelation>>()
var i = inOffsets[t]
val end = inOffsets[t + 1]
while (i < end) {
val packed = inPacked[i]
val source = pubkeyOf(packed and TrustGraph.SOURCE_MASK)
val relation = TrustRelation.entries.first { it.code == (packed ushr TrustGraph.SOURCE_BITS) }
out.add(source to relation)
i++
}
return out
}
@Test
fun buildsFollowMuteAndReportEdges() {
val b = TrustGraphBuilder()
b.addFollows(alice, listOf(bob, carol))
b.addMutes(bob, listOf(dave))
b.addReports(carol, listOf(dave))
val graph = b.build()
assertEquals(setOf(alice to TrustRelation.FOLLOW), graph.incomingOf(bob))
assertEquals(setOf(alice to TrustRelation.FOLLOW), graph.incomingOf(carol))
assertEquals(
setOf(bob to TrustRelation.MUTE, carol to TrustRelation.REPORT),
graph.incomingOf(dave),
)
}
@Test
fun dropsSelfEdges() {
val b = TrustGraphBuilder()
b.addFollows(alice, listOf(alice, bob))
val graph = b.build()
assertTrue(graph.incomingOf(alice).isEmpty(), "a self-follow must not become an edge")
assertEquals(setOf(alice to TrustRelation.FOLLOW), graph.incomingOf(bob))
}
@Test
fun dedupesRepeatedReports() {
val b = TrustGraphBuilder()
b.addReports(alice, listOf(dave))
b.addReports(alice, listOf(dave))
val graph = b.build()
assertEquals(1, graph.edgeCount())
assertEquals(setOf(alice to TrustRelation.REPORT), graph.incomingOf(dave))
}
@Test
fun keepsFollowAndMuteFromSameSourceAsDistinctEdges() {
val b = TrustGraphBuilder()
b.addFollows(alice, listOf(bob))
b.addMutes(alice, listOf(bob))
val graph = b.build()
assertEquals(
setOf(alice to TrustRelation.FOLLOW, alice to TrustRelation.MUTE),
graph.incomingOf(bob),
)
}
@Test
fun internsEachPubkeyOnce() {
val b = TrustGraphBuilder()
b.addFollows(alice, listOf(bob, carol))
b.addFollows(bob, listOf(carol))
val graph = b.build()
assertEquals(3, graph.nodeCount, "alice, bob, carol interned once each")
assertEquals(3, graph.edgeCount())
}
}
@@ -72,13 +72,15 @@ class BasicRelayClientTest {
}
@Test
fun onFailureWithMessageKeepsExistingFormat() {
fun onFailureWithMessageAppendsExceptionClassName() {
val (socket, listener) = connectAndCapture()
socket.onFailure(Exception("Connection reset"), null, null)
// The exception type is appended so listeners can classify the failure by
// its stable class name rather than by localized message text.
assertEquals(
listOf("WebSocket Failure: Connection reset"),
listOf("WebSocket Failure: Connection reset (Exception)"),
listener.cannotConnectMessages,
)
}
@@ -0,0 +1,108 @@
/*
* 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.utils.concurrent
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertFalse
import kotlin.test.assertNull
import kotlin.test.assertTrue
class ConcurrentCollectionsTest {
@Test
fun mapGetSet() {
val m = ConcurrentMap<String, Int>()
assertNull(m["a"])
m["a"] = 1
assertEquals(1, m["a"])
m["a"] = 2
assertEquals(2, m["a"])
assertEquals(1, m.size())
}
@Test
fun mapGetOrPutComputesOnce() {
val m = ConcurrentMap<String, Int>()
var calls = 0
assertEquals(
7,
m.getOrPut("k") {
calls++
7
},
)
// Present now: the default must NOT be recomputed.
assertEquals(
7,
m.getOrPut("k") {
calls++
99
},
)
assertEquals(1, calls)
assertEquals(7, m["k"])
}
@Test
fun mapMergeInsertsThenCombines() {
val m = ConcurrentMap<String, Int>()
// Absent -> inserts the value verbatim, remap not applied.
assertEquals(1, m.merge("k", 1) { a, b -> a + b })
// Present -> remap(existing, value).
assertEquals(4, m.merge("k", 3) { a, b -> a + b })
assertEquals(4, m["k"])
}
@Test
fun mapSnapshotIsDetached() {
val m = ConcurrentMap<String, Int>()
m["a"] = 1
m["b"] = 2
val snap = m.snapshot()
assertEquals(mapOf("a" to 1, "b" to 2), snap)
// Mutating the map after the snapshot must not change the snapshot.
m["c"] = 3
assertEquals(2, snap.size)
assertEquals(3, m.size())
}
@Test
fun setAddContainsSize() {
val s = ConcurrentSet<String>()
assertFalse("x" in s)
assertTrue(s.add("x"))
// Re-adding is a no-op and reports it.
assertFalse(s.add("x"))
assertTrue("x" in s)
assertTrue(s.add("y"))
assertEquals(2, s.size())
}
@Test
fun setSnapshotIsDetached() {
val s = ConcurrentSet<String>()
s.add("a")
val snap = s.snapshot()
s.add("b")
assertEquals(setOf("a"), snap)
assertEquals(2, s.size())
}
}
@@ -0,0 +1,55 @@
/*
* 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.utils.concurrent
import java.util.concurrent.ConcurrentHashMap
actual class ConcurrentMap<K : Any, V : Any> {
private val map = ConcurrentHashMap<K, V>()
actual operator fun get(key: K): V? = map[key]
actual operator fun set(
key: K,
value: V,
) {
map[key] = value
}
actual fun getOrPut(
key: K,
defaultValue: () -> V,
): V =
// Fast-path the present-key hit (the common case in the crawl's hot
// relay-hint accumulation) so it never allocates the mapping-function
// closure; only an absent key pays for the atomic computeIfAbsent.
map[key] ?: map.computeIfAbsent(key) { defaultValue() }
actual fun merge(
key: K,
value: V,
remap: (old: V, new: V) -> V,
): V = map.merge(key, value) { old, new -> remap(old, new) }!!
actual fun size(): Int = map.size
actual fun snapshot(): Map<K, V> = HashMap(map)
}
@@ -0,0 +1,35 @@
/*
* 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.utils.concurrent
import java.util.concurrent.ConcurrentHashMap
actual class ConcurrentSet<E : Any> {
private val set: MutableSet<E> = ConcurrentHashMap.newKeySet()
actual fun add(element: E): Boolean = set.add(element)
actual operator fun contains(element: E): Boolean = set.contains(element)
actual fun size(): Int = set.size
actual fun snapshot(): Set<E> = HashSet(set)
}
@@ -0,0 +1,80 @@
/*
* 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.utils.concurrent
import kotlin.concurrent.atomics.AtomicReference
import kotlin.concurrent.atomics.ExperimentalAtomicApi
// Copy-on-write, mirroring ConcurrentHashCache.linux: correct and simple. The
// native targets never run the crawl this backs (it is JVM/Android-only work);
// they only compile it, so the O(n)-per-write cost is irrelevant. A CAS retry
// loop gives getOrPut/merge the same atomicity the JVM actual gets for free.
@OptIn(ExperimentalAtomicApi::class)
actual class ConcurrentMap<K : Any, V : Any> {
private val ref = AtomicReference(HashMap<K, V>())
actual operator fun get(key: K): V? = ref.load()[key]
actual operator fun set(
key: K,
value: V,
) {
while (true) {
val cur = ref.load()
val copy = HashMap(cur)
copy[key] = value
if (ref.compareAndSet(cur, copy)) return
}
}
actual fun getOrPut(
key: K,
defaultValue: () -> V,
): V {
while (true) {
val cur = ref.load()
cur[key]?.let { return it }
val value = defaultValue()
val copy = HashMap(cur)
copy[key] = value
if (ref.compareAndSet(cur, copy)) return value
}
}
actual fun merge(
key: K,
value: V,
remap: (old: V, new: V) -> V,
): V {
while (true) {
val cur = ref.load()
val old = cur[key]
val merged = if (old == null) value else remap(old, value)
val copy = HashMap(cur)
copy[key] = merged
if (ref.compareAndSet(cur, copy)) return merged
}
}
actual fun size(): Int = ref.load().size
actual fun snapshot(): Map<K, V> = HashMap(ref.load())
}
@@ -0,0 +1,46 @@
/*
* 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.utils.concurrent
import kotlin.concurrent.atomics.AtomicReference
import kotlin.concurrent.atomics.ExperimentalAtomicApi
// Copy-on-write native actual — see ConcurrentMap.native for the rationale.
@OptIn(ExperimentalAtomicApi::class)
actual class ConcurrentSet<E : Any> {
private val ref = AtomicReference(HashSet<E>())
actual fun add(element: E): Boolean {
while (true) {
val cur = ref.load()
if (element in cur) return false
val copy = HashSet(cur)
copy.add(element)
if (ref.compareAndSet(cur, copy)) return true
}
}
actual operator fun contains(element: E): Boolean = element in ref.load()
actual fun size(): Int = ref.load().size
actual fun snapshot(): Set<E> = HashSet(ref.load())
}