mirror of
https://github.com/vitorpamplona/amethyst.git
synced 2026-08-11 16:57:39 +00:00
Merge remote-tracking branch 'origin/main' into claude/graperank-sync-crawl-1n05im
This commit is contained in:
Claude
@@ -448,6 +448,23 @@ class Context(
|
||||
/** Union of all three buckets. */
|
||||
suspend fun anyRelays(): Set<NormalizedRelayUrl> = outboxRelays() + inboxRelays() + keyPackageRelays()
|
||||
|
||||
/**
|
||||
* Index relays — the shared, app-global set used to fetch profile
|
||||
* metadata (kind 0) and follow lists (kind 3). Mirrors the Desktop
|
||||
* app's `LocalRelayCategories.indexRelays` by reading from the same
|
||||
* `java.util.prefs` node
|
||||
* (`com/vitorpamplona/amethyst/relays/index`). Falls back to the
|
||||
* shipping defaults when the user hasn't configured anything.
|
||||
*
|
||||
* This is what `amy wot sync` uses; `outboxRelays()` /
|
||||
* `inboxRelays()` remain for callers that want relay lists derived
|
||||
* from NIP-65 identity semantics.
|
||||
*/
|
||||
fun indexRelays(): Set<NormalizedRelayUrl> =
|
||||
com.vitorpamplona.amethyst.commons.relays.index
|
||||
.PreferencesIndexRelays()
|
||||
.effective()
|
||||
|
||||
/**
|
||||
* Seed relays for "look up someone we know nothing about" queries —
|
||||
* fetching another user's kind:10002 / 10050 / 10051 / 30443 before we
|
||||
|
||||
@@ -68,6 +68,7 @@ import com.vitorpamplona.amethyst.cli.commands.SubscribeCommand
|
||||
import com.vitorpamplona.amethyst.cli.commands.SyncCommand
|
||||
import com.vitorpamplona.amethyst.cli.commands.UseCommand
|
||||
import com.vitorpamplona.amethyst.cli.commands.VerifyCommand
|
||||
import com.vitorpamplona.amethyst.cli.commands.WotCommand
|
||||
import com.vitorpamplona.amethyst.cli.commands.ZapCommand
|
||||
import com.vitorpamplona.amethyst.cli.commands.cashu.CashuCommands
|
||||
import com.vitorpamplona.amethyst.cli.commands.cashu.CashuMintCommands
|
||||
@@ -287,6 +288,7 @@ private suspend fun dispatch(argv: Array<String>): Int {
|
||||
"podcast" -> PodcastCommands.dispatch(dataDir, tail)
|
||||
"podcast20" -> Podcast20Commands.dispatch(dataDir, tail)
|
||||
"bunker" -> BunkerCommand.run(dataDir, tail)
|
||||
"wot" -> WotCommand.dispatch(dataDir, tail)
|
||||
else -> {
|
||||
System.err.println("unknown subcommand: $head")
|
||||
printUsage()
|
||||
@@ -607,6 +609,15 @@ private fun printUsage() {
|
||||
| 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 update [--down] [--up] refresh every locally-known author's WoT record kinds
|
||||
| [--no-sync-deletions] [--timeout SECS] (0/3/10002/1984) from their own outbox: reads all
|
||||
| [--relay-concurrency N] [--author-chunk N] kind:10002 in the store, groups authors by write
|
||||
| [--min-authors N] [--report-limit N] relay, and runs one NIP-77 negentropy reconcile per
|
||||
| relay scoped to its authors. Bidirectional by default;
|
||||
| the deletion settle downloads the relay's kind:5 when
|
||||
| an uploaded record was rejected (author retracted it).
|
||||
| Falls back to a full paged download when a relay
|
||||
| can't reconcile via negentropy.
|
||||
| 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;
|
||||
|
||||
@@ -86,6 +86,12 @@ object Output {
|
||||
return 1
|
||||
}
|
||||
|
||||
/**
|
||||
* Shared `bad_args` failure for any command that takes a relay-URL
|
||||
* argument, so every command names the offending input the same way.
|
||||
*/
|
||||
fun invalidRelayUrl(raw: String): Int = error("bad_args", "invalid relay url: $raw")
|
||||
|
||||
private fun renderText(value: Any?): String {
|
||||
val color = Ansi.forStream(isStderr = false)
|
||||
val out = StringBuilder()
|
||||
|
||||
@@ -55,7 +55,7 @@ object AdminCommand {
|
||||
val args = Args(rest)
|
||||
val relayArg = args.positionalOrNull(0) ?: return Output.error("bad_args", "usage: admin RELAY METHOD [args]")
|
||||
val method = args.positionalOrNull(1) ?: return Output.error("bad_args", "missing method; e.g. supported-methods")
|
||||
val relay = RelayUrlNormalizer.normalizeOrNull(relayArg) ?: return Output.error("bad_args", "invalid relay url: $relayArg")
|
||||
val relay = RelayUrlNormalizer.normalizeOrNull(relayArg) ?: return Output.invalidRelayUrl(relayArg)
|
||||
val p2 = args.positionalOrNull(2)
|
||||
val reason = args.flag("reason")
|
||||
|
||||
|
||||
@@ -30,6 +30,7 @@ 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.GrapeRankUpdater
|
||||
import com.vitorpamplona.quartz.experimental.graperank.TrustGraphBuilder
|
||||
import com.vitorpamplona.quartz.nip01Core.core.Event
|
||||
import com.vitorpamplona.quartz.nip01Core.core.HexKey
|
||||
@@ -189,6 +190,7 @@ object GrapeRankCommand {
|
||||
"providers" -> providers(dataDir, tail.drop(1).toTypedArray())
|
||||
"operator" -> operator(dataDir, tail.drop(1).toTypedArray())
|
||||
"sync" -> sync(dataDir, tail.drop(1).toTypedArray())
|
||||
"update" -> update(dataDir, tail.drop(1).toTypedArray())
|
||||
"score" -> run(dataDir, tail.drop(1).toTypedArray(), forceOffline = true)
|
||||
else -> run(dataDir, tail)
|
||||
}
|
||||
@@ -514,6 +516,112 @@ object GrapeRankCommand {
|
||||
return 0
|
||||
}
|
||||
|
||||
/**
|
||||
* `amy graperank update [flags]` — refresh every locally-known author's WoT
|
||||
* record kinds (0 / 3 / 10002 / 1984) straight from their own outbox, so the
|
||||
* next `graperank score` runs on current data without a full follow-graph crawl.
|
||||
*
|
||||
* Thin wrapper over quartz's [GrapeRankUpdater]: it reads every kind:10002 in the
|
||||
* store, inverts them into a `write-relay -> authors` map (the outbox model), and
|
||||
* runs one NIP-77 negentropy reconcile per write relay scoped to its authors —
|
||||
* bidirectional, settling deletions over the residual (its applyDown direction
|
||||
* downloads the relay's kind:5 when an uploaded record was rejected), and falling
|
||||
* back to a full paged download when a relay can't reconcile. This command only
|
||||
* parses flags and renders the [GrapeRankUpdater.Result] as text/JSON.
|
||||
*
|
||||
* Flags: `--timeout SECS` (per-group idle watchdog, default 30),
|
||||
* `--relay-concurrency N` (relays reconciled at once, default 4),
|
||||
* `--author-chunk N` (authors per reconcile filter, default 500),
|
||||
* `--min-authors N` (skip relays hosting fewer than N of our authors, default 1),
|
||||
* `--report-limit N` (per-relay rows in the JSON, default 50),
|
||||
* `--down` / `--up` / `--no-sync-deletions`.
|
||||
*/
|
||||
private suspend fun update(
|
||||
dataDir: DataDir,
|
||||
rest: Array<String>,
|
||||
): Int {
|
||||
val args = Args(rest)
|
||||
val reportLimit = args.intFlag("report-limit", 50).coerceAtLeast(0)
|
||||
// Default is bidirectional; a single --down/--up narrows to that direction.
|
||||
val downFlag = args.bool("down")
|
||||
val upFlag = args.bool("up")
|
||||
|
||||
Context.openOrAnonymous(dataDir).use { ctx ->
|
||||
ctx.prepare()
|
||||
|
||||
val updater =
|
||||
GrapeRankUpdater(
|
||||
client = ctx.client,
|
||||
store = ctx.store,
|
||||
config =
|
||||
GrapeRankUpdater.Config(
|
||||
down = downFlag || !upFlag,
|
||||
up = upFlag || !downFlag,
|
||||
syncDeletions = !args.bool("no-sync-deletions"),
|
||||
relayConcurrency = args.intFlag("relay-concurrency", 4),
|
||||
authorChunk = args.intFlag("author-chunk", 500),
|
||||
minAuthors = args.intFlag("min-authors", 1),
|
||||
idleTimeoutMs = args.longFlag("timeout", 30L) * 1000,
|
||||
),
|
||||
log = { System.err.println(it) },
|
||||
)
|
||||
|
||||
val result = updater.update()
|
||||
|
||||
if (result.relays == 0) {
|
||||
Output.emit(
|
||||
linkedMapOf<String, Any?>(
|
||||
"relay_lists_in_store" to result.relayListsInStore,
|
||||
"authors_with_outbox" to result.authorsWithOutbox,
|
||||
"relays" to 0,
|
||||
"note" to "no kind:10002 write relays in the local store — run `graperank sync` first",
|
||||
),
|
||||
)
|
||||
return 0
|
||||
}
|
||||
|
||||
// Busiest relays first, capped so a many-thousand-relay run still emits a
|
||||
// bounded JSON object; totals below always cover every relay.
|
||||
val report =
|
||||
result.perRelay
|
||||
.sortedByDescending { it.downloaded + it.uploaded }
|
||||
.take(reportLimit)
|
||||
.map {
|
||||
linkedMapOf<String, Any?>(
|
||||
"relay" to it.relay.url,
|
||||
"authors" to it.authors,
|
||||
"need" to it.need,
|
||||
"have" to it.have,
|
||||
"downloaded" to it.downloaded,
|
||||
"uploaded" to it.uploaded,
|
||||
"deletions_sent_up" to it.deletionsSentUp,
|
||||
"deletions_applied_down" to it.deletionsAppliedDown,
|
||||
"paged_fallback" to it.pagedFallback,
|
||||
"error" to it.error,
|
||||
)
|
||||
}
|
||||
|
||||
Output.emit(
|
||||
linkedMapOf<String, Any?>(
|
||||
"kinds" to GrapeRankUpdater.DEFAULT_KINDS,
|
||||
"relay_lists_in_store" to result.relayListsInStore,
|
||||
"authors_with_outbox" to result.authorsWithOutbox,
|
||||
"relays" to result.relays,
|
||||
"relays_ok" to result.relaysOk,
|
||||
"relays_failed" to result.relaysFailed,
|
||||
"relays_paged_fallback" to result.relaysPagedFallback,
|
||||
"downloaded" to result.downloaded,
|
||||
"uploaded" to result.uploaded,
|
||||
"deletions_sent_up" to result.deletionsSentUp,
|
||||
"deletions_applied_down" to result.deletionsAppliedDown,
|
||||
"report_limit" to reportLimit,
|
||||
"per_relay" to report,
|
||||
),
|
||||
)
|
||||
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
|
||||
@@ -651,7 +759,7 @@ object GrapeRankCommand {
|
||||
"provider" to provider,
|
||||
"relay" to relay.url,
|
||||
"changed" to false,
|
||||
"based_on" to latest?.id,
|
||||
"based_on" to latest.id,
|
||||
),
|
||||
)
|
||||
return 0
|
||||
@@ -827,7 +935,7 @@ object GrapeRankCommand {
|
||||
latest?.serviceProviders()?.any {
|
||||
it.service == service && it.pubkey == providerPubkey && it.relayUrl == relay
|
||||
} ?: false
|
||||
if (alreadyListed) return latest?.id
|
||||
if (alreadyListed) return latest.id
|
||||
|
||||
val tag = ServiceProviderTag(service, providerPubkey, relay)
|
||||
val event =
|
||||
|
||||
@@ -237,7 +237,7 @@ object RelayCommands {
|
||||
val raw = args.positional(0, "relay-url")
|
||||
val normalized =
|
||||
raw.normalizeRelayUrlOrNull()
|
||||
?: return Output.error("bad_args", "invalid relay url: $raw")
|
||||
?: return Output.invalidRelayUrl(raw)
|
||||
val httpUrl = normalized.toHttp()
|
||||
|
||||
val request =
|
||||
@@ -286,21 +286,21 @@ object RelayCommands {
|
||||
val self = ctx.identity.pubKeyHex
|
||||
when (verb) {
|
||||
"add" -> {
|
||||
val url = parseUrl(args.positional(0, "url")) ?: return Output.error("bad_args", "invalid relay url")
|
||||
val url = urlArg(args) ?: return Output.invalidRelayUrl(args.positional(0, "url"))
|
||||
val existing = flat.read(ctx, self)
|
||||
val added = existing.none { it.url == url.url }
|
||||
if (added) ctx.verifyAndStore(flat.build(ctx, existing + url))
|
||||
Output.emit(mapOf("noun" to flat.noun, "kind" to flat.kind, "url" to url.url, "added" to added))
|
||||
}
|
||||
"remove", "rm" -> {
|
||||
val url = parseUrl(args.positional(0, "url")) ?: return Output.error("bad_args", "invalid relay url")
|
||||
val url = urlArg(args) ?: return Output.invalidRelayUrl(args.positional(0, "url"))
|
||||
val existing = flat.read(ctx, self)
|
||||
val removed = existing.any { it.url == url.url }
|
||||
if (removed) ctx.verifyAndStore(flat.build(ctx, existing.filterNot { it.url == url.url }))
|
||||
Output.emit(mapOf("noun" to flat.noun, "kind" to flat.kind, "url" to url.url, "removed" to removed))
|
||||
}
|
||||
"set" -> {
|
||||
val relays = parseUrls(args.positional) ?: return Output.error("bad_args", "invalid relay url")
|
||||
val relays = parseUrls(args.positional) ?: return badUrlIn(args.positional)
|
||||
if (relays.isEmpty()) return Output.error("bad_args", "set needs at least one URL; use `relay ${flat.noun} clear` to empty it")
|
||||
val signed = flat.build(ctx, relays)
|
||||
ctx.verifyAndStore(signed)
|
||||
@@ -335,7 +335,7 @@ object RelayCommands {
|
||||
when (verb) {
|
||||
"add", "remove", "rm" -> {
|
||||
val present = verb == "add"
|
||||
val url = parseUrl(args.positional(0, "url")) ?: return Output.error("bad_args", "invalid relay url")
|
||||
val url = urlArg(args) ?: return Output.invalidRelayUrl(args.positional(0, "url"))
|
||||
val changed = mutateNip65(ctx, self) { applyFacet(it, url, facet, present) }
|
||||
Output.emit(
|
||||
mapOf(
|
||||
@@ -352,7 +352,7 @@ object RelayCommands {
|
||||
if (verb == "clear") {
|
||||
emptyList()
|
||||
} else {
|
||||
val parsed = parseUrls(args.positional) ?: return Output.error("bad_args", "invalid relay url")
|
||||
val parsed = parseUrls(args.positional) ?: return badUrlIn(args.positional)
|
||||
if (parsed.isEmpty()) return Output.error("bad_args", "set needs at least one URL; use `relay ${facet.noun} clear` to empty it")
|
||||
parsed
|
||||
}
|
||||
@@ -388,7 +388,7 @@ object RelayCommands {
|
||||
)
|
||||
}
|
||||
"remove", "rm" -> {
|
||||
val url = parseUrl(args.positional(0, "url")) ?: return Output.error("bad_args", "invalid relay url")
|
||||
val url = urlArg(args) ?: return Output.invalidRelayUrl(args.positional(0, "url"))
|
||||
val removed = mutateNip65(ctx, self) { infos -> infos.filterNot { it.relayUrl.url == url.url } }
|
||||
Output.emit(mapOf("noun" to "nip65", "kind" to AdvertisedRelayListEvent.KIND, "url" to url.url, "removed" to removed))
|
||||
}
|
||||
@@ -416,7 +416,7 @@ object RelayCommands {
|
||||
args: Args,
|
||||
add: Boolean,
|
||||
): Int {
|
||||
val url = parseUrl(args.positional(0, "url")) ?: return Output.error("bad_args", "invalid relay url")
|
||||
val url = urlArg(args) ?: return Output.invalidRelayUrl(args.positional(0, "url"))
|
||||
Context.open(dataDir).use { ctx ->
|
||||
val self = ctx.identity.pubKeyHex
|
||||
val changed = linkedMapOf<String, Boolean>()
|
||||
@@ -518,6 +518,9 @@ object RelayCommands {
|
||||
|
||||
private fun parseUrl(raw: String): NormalizedRelayUrl? = raw.normalizeRelayUrlOrNull()
|
||||
|
||||
/** The single relay-URL argument every add/remove verb takes, or null if it doesn't parse. */
|
||||
private fun urlArg(args: Args): NormalizedRelayUrl? = parseUrl(args.positional(0, "url"))
|
||||
|
||||
/** Normalize + dedupe (order-preserving) a list of raw URLs, or null on any bad one. */
|
||||
private fun parseUrls(raws: List<String>): List<NormalizedRelayUrl>? {
|
||||
val out = mutableListOf<NormalizedRelayUrl>()
|
||||
@@ -525,6 +528,9 @@ object RelayCommands {
|
||||
return out.distinctBy { it.url }
|
||||
}
|
||||
|
||||
/** Error exit naming the first URL in [raws] that made [parseUrls] fail. */
|
||||
private fun badUrlIn(raws: List<String>): Int = Output.invalidRelayUrl(raws.first { parseUrl(it) == null })
|
||||
|
||||
private suspend fun readNip65(
|
||||
ctx: Context,
|
||||
self: HexKey,
|
||||
|
||||
@@ -26,8 +26,10 @@ import com.vitorpamplona.amethyst.cli.DataDir
|
||||
import com.vitorpamplona.amethyst.cli.Output
|
||||
import com.vitorpamplona.quartz.nip01Core.core.Event
|
||||
import com.vitorpamplona.quartz.nip01Core.core.HexKey
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.DeletionSettleResult
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.NegentropySyncException
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.negentropyReconcile
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.negentropySettleDeletions
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer
|
||||
import com.vitorpamplona.quartz.nip01Core.store.IdAndTime
|
||||
@@ -54,15 +56,29 @@ import java.util.concurrent.atomic.AtomicInteger
|
||||
* Pass both for a full bidirectional sync. The filter flags are the same as
|
||||
* `fetch`/`subscribe`; an empty filter reconciles the whole store.
|
||||
*
|
||||
* Both directions are pipelined with the reconcile: need-id batches feed
|
||||
* [DOWNLOAD_WORKERS] concurrent by-id REQ drains and have-ids feed a single
|
||||
* uploader, so downloads and uploads overlap the remaining reconcile rounds
|
||||
* instead of waiting for the full diff. Every downloaded event funnels
|
||||
* through `Context.drain`'s verify-and-store path, unchanged.
|
||||
* Deletion propagation (on by default; disable with `--no-sync-deletions`) is a
|
||||
* **second pass over the residual**, not per-event work in the content pass — so it
|
||||
* costs the same whether the database is tiny or huge. After the content settle, a
|
||||
* re-reconcile's leftover diff is (barring races) exactly the events a deletion kept
|
||||
* from converging:
|
||||
*
|
||||
* Thin assembly only: the windowing, streaming, and back-pressure live in
|
||||
* quartz (`negentropyReconcile`); this file only routes ids to
|
||||
* `Context.drain` / `Context.publish`.
|
||||
* - a residual **need** (relay has it, we still lack it after `--down` tried to
|
||||
* download) = we deleted it → publish OUR covering deletion up so the relay drops it;
|
||||
* - a residual **have** (we have it, relay still lacks it after `--up` tried to upload)
|
||||
* = the relay deleted it → pull the relay's covering kind-5 down and apply it locally.
|
||||
*
|
||||
* Coverage is any way a deletion reaches an event ([deletionsCovering]): a NIP-09 kind-5
|
||||
* by id (`e`) or address (`a`, cutoff-checked), or a NIP-62 vanish targeting this relay
|
||||
* (up direction only — a pulled vanish is not auto-applied, its blast radius being the
|
||||
* whole account). The residual is small (only real deletion mismatches), so only it is
|
||||
* fetched — never the whole need set. The loop repeats until a round resolves nothing.
|
||||
* So `amy sync` (default `--down`) makes the relay honor your deletions; `--up` makes
|
||||
* your store honor the relay's; `--up --down` converges both ways.
|
||||
*
|
||||
* Content is pipelined with the reconcile: need-id batches feed [DOWNLOAD_WORKERS]
|
||||
* concurrent by-id REQ drains and have-ids feed a single uploader. Thin assembly only:
|
||||
* the windowing, streaming, and back-pressure live in quartz (`negentropyReconcile`);
|
||||
* this file only routes ids to `Context.drain` / `Context.publish`.
|
||||
*/
|
||||
object SyncCommand {
|
||||
private const val ID_CHUNK = 500
|
||||
@@ -78,6 +94,15 @@ object SyncCommand {
|
||||
/** Overlapped `created_at`-window reconciles after an over-cap split. */
|
||||
private const val RECONCILE_CONCURRENCY = 2
|
||||
|
||||
/**
|
||||
* Cap on deletion-settle rounds. Each round resolves the residual it can and
|
||||
* re-reconciles; a healthy sync converges in 1–2 (round N sends/applies, round
|
||||
* N+1 confirms empty). The cap only bounds pathological non-convergence (e.g. a
|
||||
* relay that refuses a deletion), which the "resolved nothing → stop" check
|
||||
* normally catches first.
|
||||
*/
|
||||
private const val MAX_DELETION_ROUNDS = 4
|
||||
|
||||
suspend fun run(
|
||||
dataDir: DataDir,
|
||||
rest: Array<String>,
|
||||
@@ -88,11 +113,12 @@ object SyncCommand {
|
||||
?: return Output.error("bad_args", "sync requires --relay URL")
|
||||
val relay =
|
||||
RelayUrlNormalizer.normalizeOrNull(relayUrl)
|
||||
?: return Output.error("bad_args", "invalid relay url: $relayUrl")
|
||||
?: return Output.invalidRelayUrl(relayUrl)
|
||||
val timeoutMs = (args.flag("timeout")?.toLongOrNull() ?: 30L) * 1000
|
||||
// Default direction is download; --up adds upload.
|
||||
val up = args.bool("up")
|
||||
val down = args.bool("down") || !up
|
||||
val syncDeletions = !args.bool("no-sync-deletions")
|
||||
val filter = RawEventSupport.buildFilter(args)
|
||||
|
||||
Context.openOrAnonymous(dataDir).use { ctx ->
|
||||
@@ -104,23 +130,22 @@ object SyncCommand {
|
||||
val downloaded = AtomicInteger(0)
|
||||
val uploaded = AtomicInteger(0)
|
||||
|
||||
// ── Pass 1: content settle — download needs, upload haves. No deletion
|
||||
// logic, so a plain sync costs exactly what it always did.
|
||||
val result =
|
||||
try {
|
||||
coroutineScope {
|
||||
// needIds = relay has, we lack; haveIds = we have, relay lacks.
|
||||
// Bounded so a slow download back-pressures the reconcile
|
||||
// rounds instead of piling ids up in memory.
|
||||
val needBatches = Channel<List<HexKey>>(DOWNLOAD_WORKERS * 2)
|
||||
// Unbounded is fine here: have-ids reference events we already
|
||||
// hold locally, so memory is bounded by the local set.
|
||||
val haveBatches = Channel<List<HexKey>>(Channel.UNLIMITED)
|
||||
|
||||
val downloaders =
|
||||
List(DOWNLOAD_WORKERS) {
|
||||
launch {
|
||||
for (batch in needBatches) {
|
||||
val got = ctx.drain(mapOf(relay to listOf(Filter(ids = batch))), timeoutMs)
|
||||
downloaded.addAndGet(got.size)
|
||||
// drain verifies + stores; anything we deleted is
|
||||
// rejected by our own tombstone and stays a "need".
|
||||
downloaded.addAndGet(ctx.drain(mapOf(relay to listOf(Filter(ids = batch))), timeoutMs).size)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -129,8 +154,7 @@ object SyncCommand {
|
||||
for (batch in haveBatches) {
|
||||
for (id in batch) {
|
||||
val ev = localById[id] ?: continue
|
||||
val ack = ctx.publish(ev, setOf(relay))
|
||||
if (ack.values.any { it }) uploaded.incrementAndGet()
|
||||
if (ctx.publish(ev, setOf(relay)).values.any { it }) uploaded.incrementAndGet()
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -160,6 +184,28 @@ object SyncCommand {
|
||||
return Output.error("sync_error", e.message ?: "negentropy sync failed")
|
||||
}
|
||||
|
||||
// ── Pass 2+: deletion settle. The reusable quartz accessory re-reconciles
|
||||
// and resolves only the residual — send our deletions up for what we deleted
|
||||
// (bounded by --down), apply the relay's kind-5 down for what it deleted
|
||||
// (bounded by --up) — looping until stable. Cheap regardless of database size
|
||||
// (see negentropySettleDeletions), and best-effort so it can't fail the sync.
|
||||
val deletions =
|
||||
if (syncDeletions) {
|
||||
ctx.client.negentropySettleDeletions(
|
||||
relay = relay,
|
||||
filter = filter,
|
||||
store = ctx.store,
|
||||
sendUp = down,
|
||||
applyDown = up,
|
||||
batchSize = ID_CHUNK,
|
||||
idleTimeoutMs = timeoutMs,
|
||||
maxRounds = MAX_DELETION_ROUNDS,
|
||||
reconcileConcurrency = RECONCILE_CONCURRENCY,
|
||||
)
|
||||
} else {
|
||||
DeletionSettleResult(0, 0, 0)
|
||||
}
|
||||
|
||||
Output.emit(
|
||||
mapOf(
|
||||
"relay" to relay.url,
|
||||
@@ -169,6 +215,9 @@ object SyncCommand {
|
||||
"have" to result.haveCount,
|
||||
"downloaded" to downloaded.get(),
|
||||
"uploaded" to uploaded.get(),
|
||||
"deletions_sent_up" to deletions.sentUp,
|
||||
"deletions_applied_down" to deletions.appliedDown,
|
||||
"deletion_rounds" to deletions.rounds,
|
||||
),
|
||||
)
|
||||
return 0
|
||||
|
||||
@@ -0,0 +1,225 @@
|
||||
/*
|
||||
* 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.wot.OutboxCacheGateway
|
||||
import com.vitorpamplona.amethyst.commons.wot.OutboxDispatcher
|
||||
import com.vitorpamplona.amethyst.commons.wot.WoTService
|
||||
import com.vitorpamplona.quartz.nip01Core.core.Event
|
||||
import com.vitorpamplona.quartz.nip01Core.core.HexKey
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
|
||||
import com.vitorpamplona.quartz.nip65RelayList.AdvertisedRelayListEvent
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.SupervisorJob
|
||||
import kotlinx.coroutines.cancel
|
||||
import java.util.Collections
|
||||
|
||||
/**
|
||||
* `amy wot <get|list|sync>` — Web-of-Trust score queries.
|
||||
*
|
||||
* The score for a pubkey X is the count of accounts in the active user's
|
||||
* kind-3 follow set who also follow X. `get` and `list` are read-only —
|
||||
* they hydrate the score map from whatever kind-3 events already live in
|
||||
* the local event store. `sync` pulls fresh kind-3 events from the
|
||||
* configured relay pool so the next `get` / `list` is up to date.
|
||||
*/
|
||||
object WotCommand {
|
||||
suspend fun dispatch(
|
||||
dataDir: DataDir,
|
||||
rest: Array<String>,
|
||||
): Int {
|
||||
val head = rest.firstOrNull() ?: return usage()
|
||||
val tail = rest.drop(1).toTypedArray()
|
||||
return when (head) {
|
||||
"get" -> get(dataDir, tail)
|
||||
"list" -> list(dataDir, tail)
|
||||
"sync" -> sync(dataDir, tail)
|
||||
else -> usage()
|
||||
}
|
||||
}
|
||||
|
||||
private fun usage(): Int = Output.error("bad_args", "wot <get|list|sync>")
|
||||
|
||||
private suspend fun get(
|
||||
dataDir: DataDir,
|
||||
rest: Array<String>,
|
||||
): Int {
|
||||
if (rest.isEmpty()) return Output.error("bad_args", "wot get <pubkey|npub>")
|
||||
val userArg = rest[0]
|
||||
Context.open(dataDir).use { ctx ->
|
||||
ctx.prepare()
|
||||
val target = ctx.requireUserHex(userArg)
|
||||
val (svc, scope) = buildHydratedService(ctx)
|
||||
try {
|
||||
val score = svc.scoresSnapshot()[target] ?: 0
|
||||
Output.emit(mapOf("pubkey" to target, "score" to score))
|
||||
return 0
|
||||
} finally {
|
||||
scope.cancel()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private suspend fun list(
|
||||
dataDir: DataDir,
|
||||
rest: Array<String>,
|
||||
): Int {
|
||||
val args = Args(rest)
|
||||
val threshold = args.flag("threshold")?.toIntOrNull() ?: 1
|
||||
val limit = args.flag("limit")?.toIntOrNull() ?: 50
|
||||
Context.open(dataDir).use { ctx ->
|
||||
ctx.prepare()
|
||||
val (svc, scope) = buildHydratedService(ctx)
|
||||
try {
|
||||
val entries =
|
||||
svc
|
||||
.scoresSnapshot()
|
||||
.entries
|
||||
.asSequence()
|
||||
.filter { it.value >= threshold }
|
||||
.sortedByDescending { it.value }
|
||||
.take(limit)
|
||||
.map { mapOf("pubkey" to it.key, "score" to it.value) }
|
||||
.toList()
|
||||
Output.emit(mapOf("count" to entries.size, "entries" to entries))
|
||||
return 0
|
||||
} finally {
|
||||
scope.cancel()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private suspend fun sync(
|
||||
dataDir: DataDir,
|
||||
rest: Array<String>,
|
||||
): Int {
|
||||
val args = Args(rest)
|
||||
// Overall timeout; per-relay budget is set by OutboxDispatcher's
|
||||
// default (4s). `--timeout N` overrides the overall cap.
|
||||
val overallTimeoutMs = args.flag("timeout")?.toLongOrNull()?.times(1000) ?: 8_000L
|
||||
Context.open(dataDir).use { ctx ->
|
||||
ctx.prepare()
|
||||
val self = ctx.identity.pubKeyHex
|
||||
val myKind3 = ctx.contactsOf(self)
|
||||
val follows =
|
||||
myKind3?.verifiedFollowKeySet()?.toSet()
|
||||
?: return Output.error("no_follows", "no kind-3 in local store; run `amy follow` first")
|
||||
if (follows.isEmpty()) {
|
||||
Output.emit(mapOf("synced" to 0, "detail" to "empty follow set"))
|
||||
return 0
|
||||
}
|
||||
val relays = ctx.indexRelays()
|
||||
if (relays.isEmpty()) return Output.error("no_relays", "no index relays configured")
|
||||
|
||||
val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
|
||||
try {
|
||||
// Buffer discovered events; persist synchronously after
|
||||
// the fetch. `store.insert` is suspending so we can't call
|
||||
// it from the non-suspending gateway callbacks. This also
|
||||
// keeps `insert` errors surfaceable in a single log line
|
||||
// rather than swallowed into a race.
|
||||
val buffered = Collections.synchronizedList(mutableListOf<Event>())
|
||||
val gateway =
|
||||
object : OutboxCacheGateway {
|
||||
override fun cachedOutbox(pubkey: HexKey): AdvertisedRelayListEvent? =
|
||||
// Amy's store lookup is suspending; can't do
|
||||
// it here. The dispatcher then falls through
|
||||
// to Phase 1 discovery for every author, which
|
||||
// matches the old `amy wot sync` behaviour of
|
||||
// always re-asking. A future optimisation
|
||||
// could pre-populate a `Map<HexKey,
|
||||
// AdvertisedRelayListEvent>` before dispatch.
|
||||
null
|
||||
|
||||
override fun onOutboxDiscovered(
|
||||
event: AdvertisedRelayListEvent,
|
||||
relay: NormalizedRelayUrl,
|
||||
) {
|
||||
buffered.add(event)
|
||||
}
|
||||
|
||||
override fun onDiscoveredEvent(
|
||||
event: Event,
|
||||
relay: NormalizedRelayUrl,
|
||||
) {
|
||||
buffered.add(event)
|
||||
}
|
||||
}
|
||||
|
||||
val dispatcher =
|
||||
OutboxDispatcher(
|
||||
client = ctx.client,
|
||||
scope = scope,
|
||||
indexRelays = { relays },
|
||||
gateway = gateway,
|
||||
overallTimeoutMs = overallTimeoutMs,
|
||||
)
|
||||
|
||||
val result = dispatcher.fetchKind3Only(follows)
|
||||
|
||||
// Persist to store so future `get` / `list` see them.
|
||||
val eventsToPersist = synchronized(buffered) { buffered.toList() }
|
||||
eventsToPersist.forEach { runCatching { ctx.store.insert(it) } }
|
||||
|
||||
Output.emit(
|
||||
mapOf(
|
||||
"followers" to follows.size,
|
||||
"authors_requested" to result.authorsRequested,
|
||||
"kind10002_received" to result.kind10002Received,
|
||||
"kind3_received" to result.kind3Received,
|
||||
"outbox_covered_authors" to result.outboxCoveredAuthors,
|
||||
"fallback_authors" to result.fallbackAuthors,
|
||||
"persisted" to eventsToPersist.size,
|
||||
),
|
||||
)
|
||||
return 0
|
||||
} finally {
|
||||
scope.cancel()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Build a [WoTService], populate it from the local event store, then
|
||||
* return the (service, backing scope). Caller must cancel the scope
|
||||
* when done.
|
||||
*/
|
||||
private suspend fun buildHydratedService(ctx: Context): Pair<WoTService, CoroutineScope> {
|
||||
val self = ctx.identity.pubKeyHex
|
||||
val scope = CoroutineScope(SupervisorJob() + Dispatchers.Unconfined)
|
||||
val svc = WoTService(scope, writerDispatcher = Dispatchers.Unconfined)
|
||||
val myKind3 = ctx.contactsOf(self)
|
||||
val follows: Set<HexKey> = myKind3?.verifiedFollowKeySet() ?: emptySet()
|
||||
svc.onFollowSetChange(follows, self)
|
||||
// Pull each follower's kind-3 from the store and feed into the service.
|
||||
follows.forEach { follower ->
|
||||
val followerKind3 = ctx.contactsOf(follower) ?: return@forEach
|
||||
svc.applyKind3(follower, followerKind3.verifiedFollowKeySet())
|
||||
}
|
||||
svc.markReadyOnce()
|
||||
return svc to scope
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user