Merge remote-tracking branch 'origin/main' into claude/armada-nip29-integration-lwqard

# Conflicts:
#	cli/tests/.gitignore
This commit is contained in:
Claude
2026-07-09 21:48:04 +00:00
226 changed files with 21723 additions and 1072 deletions
@@ -143,6 +143,16 @@ data class Identity(
npub = pubHex.hexToByteArray().toNpub(),
)
/**
* Ephemeral, key-less identity for anonymous read-only runs (no
* account on disk). It mints a throwaway public key so the
* relay-list fallbacks (`outboxRelays()` etc.) resolve to the
* built-in defaults, and it carries no private key, so any attempt
* to sign/encrypt fails loudly — "you can read, you just can't
* auth". Used by [com.vitorpamplona.amethyst.cli.Context.openOrAnonymous].
*/
fun anonymous(): Identity = fromPublicKeyHex(KeyPair().pubKey.toHexKey())
/**
* Rebuild an in-memory identity after a load. Accepts the public
* parts that live on disk and a private key resolved from the
@@ -204,6 +214,17 @@ class DataDir(
val eventsDir: File,
val accountName: String,
val secrets: SecretStore,
/**
* Whether this points at a concrete account. `false` for the
* accountless directory [resolveOptional] hands back when `~/.amy/`
* has no unambiguous account — [root] then points at the shared
* sibling and only [eventsDir] (the cross-account event store) is
* meaningful. Read-only verbs run anonymously against it; signing
* verbs get [noAccountDetail] via `Context.open`.
*/
val hasAccount: Boolean = true,
/** Human-readable reason there is no account, for the signing-verb error. */
val noAccountDetail: String? = null,
) {
val identityFile = File(root, "identity.json")
val stateFile = File(root, "state.json")
@@ -213,14 +234,34 @@ 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)
SecureFileIO.secureMkdirs(groupsDir)
// Tighten perms on any data already on disk from an older, unhardened CLI.
SecureFileIO.tighten(identityFile)
SecureFileIO.tighten(stateFile)
SecureFileIO.tighten(marmotDir)
SecureFileIO.tighten(keyPackageBundleFile)
// The accountless dir only ever exposes the shared event store; don't
// seed per-account marmot dirs / tighten identity files under it.
if (hasAccount) {
SecureFileIO.secureMkdirs(groupsDir)
// Tighten perms on any data already on disk from an older, unhardened CLI.
SecureFileIO.tighten(identityFile)
SecureFileIO.tighten(stateFile)
SecureFileIO.tighten(marmotDir)
SecureFileIO.tighten(keyPackageBundleFile)
}
}
/**
@@ -375,6 +416,69 @@ class DataDir(
)
}
/**
* Like [resolve], but never throws when there is no account: read-only
* verbs can run without one. When `--account` is given it is honoured;
* otherwise the pin / sole-account are used if unambiguous. Failing
* that, returns an *accountless* [DataDir] (`hasAccount = false`) whose
* [root] is the shared sibling and whose [eventsDir] is still the
* cross-account event store — enough for anonymous relay queries and
* `store` maintenance. The reason no account was chosen is carried in
* [DataDir.noAccountDetail] so a signing verb can surface it.
*/
fun resolveOptional(
accountFlag: String?,
secrets: SecretStore,
): DataDir {
val rootBase = DEFAULT_ROOT
val sharedEvents = File(rootBase, "$SHARED_DIR_NAME/events-store").absoluteFile
if (accountFlag != null) {
val name = validateName(accountFlag)
return DataDir(File(rootBase, name).absoluteFile, sharedEvents, name, secrets)
}
val picked = pickAccountOptional(rootBase)
return if (picked.name != null) {
DataDir(File(rootBase, picked.name).absoluteFile, sharedEvents, picked.name, secrets)
} else {
DataDir(
root = File(rootBase, SHARED_DIR_NAME).absoluteFile,
eventsDir = sharedEvents,
accountName = SHARED_DIR_NAME,
secrets = secrets,
hasAccount = false,
noAccountDetail = picked.detail,
)
}
}
/** Result of [pickAccountOptional]: an account [name], or null plus a [detail] reason. */
private data class OptionalPick(
val name: String?,
val detail: String?,
)
/** Non-throwing sibling of [pickAccount]: null [name] with a [detail] when 0 / ambiguous. */
private fun pickAccountOptional(rootBase: File): OptionalPick {
val current = File(rootBase, CURRENT_MARKER_NAME)
if (current.isFile) {
val pinned = current.readText().trim()
if (pinned.isNotEmpty() && File(rootBase, pinned).isDirectory) {
return OptionalPick(pinned, null)
}
}
val accounts = listAccounts(rootBase)
return when (accounts.size) {
0 -> OptionalPick(null, "no account configured (create one with `amy --account <name> init`)")
1 -> OptionalPick(accounts.single(), null)
else ->
OptionalPick(
null,
"multiple accounts in ${rootBase.absolutePath} (${accounts.joinToString(", ")}); " +
"pick one with --account <name> or `amy use <name>`",
)
}
}
/**
* Auto-select an account when `--name` was not given. Honours
* `<root>/current` first (explicit pin from `amy use`), then
@@ -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
@@ -69,6 +71,7 @@ import com.vitorpamplona.quartz.nip60Cashu.wallet.CashuWalletEvent
import com.vitorpamplona.quartz.nip61Nutzaps.info.NutzapInfoEvent
import com.vitorpamplona.quartz.nip61Nutzaps.nutzap.NutzapEvent
import com.vitorpamplona.quartz.nip65RelayList.AdvertisedRelayListEvent
import com.vitorpamplona.quartz.nip66RelayMonitor.reachability.RelayReachabilityStore
import com.vitorpamplona.quartz.nip87Ecash.recommendation.MintRecommendationEvent
import com.vitorpamplona.quartz.utils.SeenIds
import kotlinx.coroutines.CompletableDeferred
@@ -78,7 +81,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 +103,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,
@@ -115,8 +121,40 @@ class Context(
val dataDir: DataDir,
val identity: Identity,
val state: RunState,
/**
* Anonymous read-only run: no account on disk, [identity] is an ephemeral
* key-less identity (see [Identity.anonymous]). Marmot state is not
* restored and run-state is not persisted — the run only reads relays and
* the shared event store. Signing verbs never take this path; they go
* through [Companion.open], which requires a real account.
*/
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(
@@ -146,6 +184,52 @@ 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(
// The starting per-relay concurrent-sub cap dominates whether the crawl
// floods a popular relay into timing out. Benchmarked: 16 is ~30% faster
// on a from-scratch GrapeRank crawl than the old 100 (which drowned
// damus/nos.lol in 100 concurrent giant REQs) at equal completeness, and
// is still generous for the single-user fetches other amy commands do.
startCap = 16,
).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
@@ -158,32 +242,40 @@ class Context(
.OkHttpNip05Fetcher { _ -> okhttp },
)
private val mlsStore = FileMlsGroupStateStore(dataDir.groupsDir)
private val keyPackageStore = FileKeyPackageBundleStore(dataDir.keyPackageBundleFile)
private val messageStore = FileMarmotMessageStore(dataDir.groupsDir)
// Lazy so an anonymous read (no account dir) never materialises the
// per-account marmot stores — constructing them would `mkdir` group dirs
// under the shared root. Real accounts build them on first marmot use.
private val mlsStore by lazy { FileMlsGroupStateStore(dataDir.groupsDir) }
private val keyPackageStore by lazy { FileKeyPackageBundleStore(dataDir.keyPackageBundleFile) }
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
/**
* Shared relay-reachability cache (NIP-66 kind:30166 records in [store]), signed by
* the machine's dedicated monitor key — derived from the operator master, NOT the
* account (see [OperatorKeys.monitorKey]). The crawler and the WoT updater read its
* dead set to skip proven-dead relays and write their findings back, so liveness
* knowledge is shared across procedures and runs instead of rediscovered each time.
* Lazy so a run that never touches relays doesn't materialize the operator master.
*/
val reachability: RelayReachabilityStore by lazy {
RelayReachabilityStore(
store = store,
signer = NostrSignerInternal(dataDir.operatorKeys().monitorKey()),
)
}
/** Fully-wired manager. Call [prepare] once before use to load persisted state. */
val marmot: MarmotManager = MarmotManager(signer, mlsStore, messageStore, keyPackageStore)
val marmot: MarmotManager by lazy { MarmotManager(signer, mlsStore, messageStore, keyPackageStore) }
// ------------------------------------------------------------------
// Cashu (NIP-60 / NIP-61) — shared wallet code from commons
@@ -302,7 +394,9 @@ class Context(
*/
suspend fun prepare() {
if (prepared) return
marmot.restoreAll()
// Anonymous runs have no account and therefore no marmot state to
// restore (and touching `marmot` would allocate the per-account stores).
if (!anonymous) marmot.restoreAll()
client.connect()
// A bunker account must open its NIP-46 response subscription and run
// the connect handshake before any signing/encryption call.
@@ -370,6 +464,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
@@ -411,15 +522,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 {
@@ -436,7 +558,7 @@ class Context(
relay: NormalizedRelayUrl,
forFilters: List<Filter>?,
) {
doneChannel.trySend(relay)
doneChannel.trySend(relay to "eose")
}
override fun onClosed(
@@ -444,7 +566,7 @@ class Context(
relay: NormalizedRelayUrl,
forFilters: List<Filter>?,
) {
doneChannel.trySend(relay)
doneChannel.trySend(relay to "closed:$message")
}
override fun onCannotConnect(
@@ -452,37 +574,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
@@ -582,26 +742,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]
@@ -852,7 +1003,8 @@ class Context(
}
override fun close() {
dataDir.saveRunState(state)
// Nothing to persist for an anonymous run (no account dir to write into).
if (!anonymous) dataDir.saveRunState(state)
(signer as? NostrSignerRemote)?.let {
try {
it.closeSubscription()
@@ -881,12 +1033,21 @@ class Context(
*/
private const val GIFT_WRAP_LOOKBACK_SECS: Long = 2L * 24 * 60 * 60
/** Build a Context but require an identity to already exist — most commands can't run without one. */
/**
* Build a Context but require an account with a usable identity —
* signing verbs can't run without one. Throws [IllegalArgumentException]
* (→ exit 2) when no account was resolvable, carrying the "which
* account?" hint from [DataDir.resolveOptional]; throws
* [IllegalStateException] when the account exists but has no identity.
*/
fun open(dataDir: DataDir): Context {
require(dataDir.hasAccount) {
dataDir.noAccountDetail ?: "no account selected; pass --account <name> or run `amy use <name>`"
}
val identity =
dataDir.loadIdentityOrNull()
?: run {
System.err.println("No identity found at ${dataDir.identityFile}. Run `amethyst-cli init` first.")
System.err.println("No identity found at ${dataDir.identityFile}. Run `amy --account ${dataDir.accountName} init` first.")
throw IllegalStateException("no identity")
}
return Context(
@@ -895,5 +1056,24 @@ class Context(
state = dataDir.loadRunState(),
)
}
/**
* Context for read-only verbs: use the resolved account when one is
* present, otherwise run anonymously (ephemeral key-less identity, no
* persisted state). Lets `fetch`/`subscribe`/`count`/`publish`/`outbox`/
* … query relays and the shared store with no account on disk — they
* read fine, they just can't sign.
*/
fun openOrAnonymous(dataDir: DataDir): Context =
if (dataDir.hasAccount && dataDir.identityExists()) {
open(dataDir)
} else {
Context(
dataDir = dataDir,
identity = Identity.anonymous(),
state = RunState(),
anonymous = true,
)
}
}
}
@@ -38,12 +38,14 @@ 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
import com.vitorpamplona.amethyst.cli.commands.KeyPackageCommands
import com.vitorpamplona.amethyst.cli.commands.KindCommand
import com.vitorpamplona.amethyst.cli.commands.LoginCommand
import com.vitorpamplona.amethyst.cli.commands.LogoffCommand
import com.vitorpamplona.amethyst.cli.commands.MarmotResetCommand
import com.vitorpamplona.amethyst.cli.commands.MessageCommands
import com.vitorpamplona.amethyst.cli.commands.NamecoinCommand
@@ -61,16 +63,20 @@ import com.vitorpamplona.amethyst.cli.commands.RelayCommands
import com.vitorpamplona.amethyst.cli.commands.RelayGroupCommands
import com.vitorpamplona.amethyst.cli.commands.SearchCommand
import com.vitorpamplona.amethyst.cli.commands.ServeCommand
import com.vitorpamplona.amethyst.cli.commands.StatusCommand
import com.vitorpamplona.amethyst.cli.commands.StoreCommands
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
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
@@ -104,6 +110,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" }) {
@@ -129,6 +141,16 @@ class AwaitTimeout(
message: String,
) : RuntimeException(message)
/**
* Verbs that create, select, or delete the account/identity on disk. They
* write to (or read) the per-account directory directly rather than through
* `Context.open`, so they need a concrete account and must resolve strictly —
* an accountless run has nowhere to put a new identity. Every other verb
* resolves via [DataDir.resolveOptional] and either runs anonymously (reads)
* or re-asserts the requirement inside `Context.open` (signing).
*/
private val STRICT_ACCOUNT_VERBS = setOf("init", "create", "login", "logoff", "whoami")
private suspend fun dispatch(argv: Array<String>): Int {
if (argv.isEmpty() || argv[0] == "--help" || argv[0] == "-h") {
printUsage()
@@ -150,6 +172,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
@@ -170,6 +193,14 @@ private suspend fun dispatch(argv: Array<String>): Int {
return UseCommand.run(tail)
}
// `status` is a cross-account, read-only overview of everything on
// disk under ~/.amy/. Like `use`, it must work regardless of how many
// accounts exist (zero, one, or many), so it dispatches before account
// resolution rather than through the single-account DataDir path.
if (head == "status") {
return StatusCommand.run(tail)
}
// Stateless local primitives (nak-style army-knife verbs). They operate
// purely on their arguments — no identity, no relays, no `~/.amy/` — so
// they dispatch before account resolution and work with zero state.
@@ -197,13 +228,33 @@ private suspend fun dispatch(argv: Array<String>): Int {
return CashuMintCommands.dispatch(tail.drop(1).toTypedArray())
}
// `offer info NOFFER` / `debit info NDEBIT` decode a CLINK pointer locally —
// no network, no account. The rest of `offer`/`debit` operates on the account.
if (head == "offer" && tail.firstOrNull() == "info") {
return OfferCommands.info(tail.drop(1).toTypedArray())
}
if (head == "debit" && tail.firstOrNull() == "info") {
return DebitCommands.info(tail.drop(1).toTypedArray())
}
val secrets = SecretStore.from(backendFlag = secretBackendFlag, passphraseFile = passphraseFileFlag)
val dataDir = DataDir.resolve(accountFlag = accountFlag, secrets = secrets)
// Identity-lifecycle verbs create / select / delete the account itself, so
// they need a concrete account and resolve strictly (helpful ambiguity
// errors). Everything else resolves optionally: read-only verbs then run
// anonymously when there is no account, while signing verbs re-assert the
// requirement through `Context.open`.
val dataDir =
if (head in STRICT_ACCOUNT_VERBS) {
DataDir.resolve(accountFlag = accountFlag, secrets = secrets)
} else {
DataDir.resolveOptional(accountFlag = accountFlag, secrets = secrets)
}
return when (head) {
"init" -> InitCommands.init(dataDir, Args(tail))
"create" -> CreateCommand.run(dataDir, tail)
"login" -> LoginCommand.run(dataDir, tail)
"logoff" -> LogoffCommand.run(dataDir, tail)
"whoami" -> InitCommands.whoami(dataDir)
"relay" -> RelayCommands.dispatch(dataDir, tail)
"marmot" -> marmotDispatch(dataDir, tail)
@@ -216,6 +267,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)
@@ -238,6 +290,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()
@@ -267,11 +320,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(
@@ -290,7 +345,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 {
@@ -315,20 +370,27 @@ 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:
| 1. --account X if given.
| 2. ~/.amy/current marker (set by `amy use X`).
| 3. Sole subdirectory of ~/.amy/ other than shared/.
| 4. Error — disambiguate with --account or `amy use`.
| 4. Read-only verbs (fetch, subscribe, count, publish, outbox,
| search, sync, store, profile/git/podcast reads, nsite/napplet
| fetch, decode/encode/… primitives, offer/debit info) run
| ANONYMOUSLY — they query relays and the shared store with no
| account, they just can't sign. Signing verbs error here:
| disambiguate with --account or `amy use`.
|
| Test harnesses isolate by overriding ${'$'}HOME for the amy
| subprocess (`HOME=/tmp/run.123 amy --account alice ...`).
@@ -336,6 +398,9 @@ private fun printUsage() {
| use NAME pin NAME as the active account
| use --clear remove the pin
| use print current pin + available accounts
| status read-only overview of every account, signer
| type, local Marmot/Cashu state, and the shared
| event store (no keychain prompt, no network)
|
|Output:
| Default: human-readable text on stdout.
@@ -382,6 +447,9 @@ private fun printUsage() {
| create [--name NAME] provision a full Amethyst-style account + publish bootstrap events
| login KEY [--password X] import (nsec|ncryptsec|mnemonic|npub|nprofile|hex|nip05|bunker://)
| whoami print current identity
| logoff [--yes] [--keep-events] log off: delete this account's key, per-account state,
| and its events in the shared store (--keep-events skips the
| cache purge). Requires --yes; without it, prints a dry run.
|
|Remote signing (NIP-46):
| bunker [--relay URL[,URL…]] run a remote signer for this (local-key) account; prints a
@@ -526,6 +594,42 @@ 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 dumps per-relay telemetry: outcome
| mix, yield, latency, and a LIVE/DEAD + limits
| classification table of every relay contacted).
| [--publish] [--min-rank N] OBSERVER: npub|nprofile|hex|name@domain (default:
| [--publish-limit N] [--publish-relay URL] active account). --offline scores from the local
| store only. --publish reconciles NIP-85 kind:30382
| cards signed by a per-observer service key: sends
| new/changed ranks >= --min-rank (default 2), skips
| unchanged, and retracts (kind:5) any card whose
| target left the graph or fell below the cutoff.
| 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;
| 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
@@ -617,11 +721,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,175 @@
/*
* 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++
}
}
/**
* The machine's dedicated NIP-66 relay-monitor identity, derived once from the
* operator master (independent of any amy account). Unlike [serviceKey] this is
* NOT per-observer — the machine publishes relay-reachability (kind:30166) under a
* single, stable monitor pubkey, so a re-probe *replaces* the prior 30166 for a
* relay instead of orphaning it. Re-derivable from the one master seed alone.
*/
fun monitorKey(): KeyPair {
val master = masterPriv()
var counter = 0
while (true) {
val material = master + "$MONITOR_LABEL$counter".encodeToByteArray()
val kp = runCatching { KeyPair(privKey = sha256(material)) }.getOrNull()
if (kp?.privKey != null) 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:"
private const val MONITOR_LABEL = "relay-monitor:"
}
}
@@ -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()
@@ -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,121 @@
/*
* 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 java.io.IOException
import java.nio.file.Files
import java.nio.file.Path
import java.util.concurrent.TimeUnit
import kotlin.io.path.exists
/**
* Read-only introspection of a file-backed Nostr event store on disk.
*
* Pure filesystem walk — no relay traffic, no writer lock, no [Context].
* Shared by `amy store stat` (full detail) and `amy status` (a compact
* roll-up alongside the account overview).
*/
data class StoreStats(
val events: Long,
/** Per-kind event counts derived from `idx/kind/<k>/`, sorted by kind string. */
val byKind: Map<String, Long>,
val diskBytes: Long,
/** Oldest / newest event file mtime, in unix seconds. Null on an empty store. */
val oldestAt: Long?,
val newestAt: Long?,
val root: Path,
) {
val distinctKinds: Int get() = byKind.size
companion object {
/** Compute stats for the store rooted at [storeRoot]. Missing dir → all-zero. */
fun of(storeRoot: Path): StoreStats {
if (!storeRoot.exists()) {
return StoreStats(0, emptyMap(), 0L, null, null, storeRoot.toAbsolutePath())
}
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
}
}
}
return StoreStats(
events = count,
byKind = byKind,
diskBytes = walkSize(storeRoot),
oldestAt = oldest,
newestAt = newest,
root = storeRoot.toAbsolutePath(),
)
}
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
}
}
}
@@ -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")
@@ -85,7 +85,8 @@ object BlossomCommands {
.map { it.trim() }
.filter { it.isNotEmpty() }
Context.open(dataDir).use { _ ->
// Read-only HEAD probe — no auth, so it runs anonymously without an account.
Context.openOrAnonymous(dataDir).use { _ ->
val http = OkHttpClient()
val results =
hashes.map { hash ->
@@ -188,7 +189,8 @@ object BlossomCommands {
val server = args.flag("server")
val url = if (server != null && !target.startsWith("http")) BlossomServerUrl.blob(server, target) else target
Context.open(dataDir).use { ctx ->
// Public download — no auth, so it runs anonymously without an account.
Context.openOrAnonymous(dataDir).use { ctx ->
val bytes =
BlossomClient().download(url)
?: return Output.error("not_found", "server returned no blob for $url")
@@ -47,7 +47,7 @@ object CountCommand {
val timeoutMs = (args.flag("timeout")?.toLongOrNull() ?: 15L) * 1000
val filter = RawEventSupport.buildFilter(args)
Context.open(dataDir).use { ctx ->
Context.openOrAnonymous(dataDir).use { ctx ->
ctx.prepare()
val relays = RawEventSupport.queryTargets(ctx, args)
if (relays.isEmpty()) return Output.error("no_relays", "no relays available; pass --relay or run `amy relay add`")
@@ -60,7 +60,7 @@ object DebitCommands {
)
/** Local decode of an `ndebit` pointer — no network, no account needed. */
private fun info(rest: Array<String>): Int {
internal fun info(rest: Array<String>): Int {
val args = Args(rest)
val debit =
ClinkPointerParser.parse(args.positional(0, "ndebit").trim()) as? NDebit
@@ -61,7 +61,10 @@ object FeedCommand {
val until = args.flag("until")?.toLongOrNull()
val timeoutSecs = args.longFlag("timeout", 8L)
Context.open(dataDir).use { ctx ->
// Read-only: runs anonymously when there is no account. `--author` /
// `--following` still work; the bare "self" feed just has no self to
// resolve without an account.
Context.openOrAnonymous(dataDir).use { ctx ->
ctx.prepare()
val (authors, mode) =
@@ -94,7 +94,7 @@ object FetchCommand {
val filter = RawEventSupport.buildFilter(args).copy(limit = effectiveLimit)
val paginate = args.bool("paginate") || args.bool("all")
Context.open(dataDir).use { ctx ->
Context.openOrAnonymous(dataDir).use { ctx ->
ctx.prepare()
val relays = RawEventSupport.queryTargets(ctx, args)
if (relays.isEmpty()) return Output.error("no_relays", "no relays available; pass --relay or run `amy relay add`")
@@ -148,7 +148,7 @@ object FetchCommand {
timeoutMs: Long,
): Int {
val code = codeArg.removePrefix("nostr:")
Context.open(dataDir).use { ctx ->
Context.openOrAnonymous(dataDir).use { ctx ->
ctx.prepare()
var filter: Filter
@@ -113,7 +113,9 @@ object GitCommands {
rest: Array<String>,
): Int {
val args = Args(rest)
Context.open(dataDir).use { ctx ->
// Read-only: runs anonymously when there is no account (defaults to
// the anonymous key, so pass a USER to list someone's repos).
Context.openOrAnonymous(dataDir).use { ctx ->
ctx.prepare()
val author = args.positionalOrNull(0)?.let { ctx.requireUserHex(it) } ?: ctx.identity.pubKeyHex
val relays = RawEventSupport.queryTargets(ctx, args)
@@ -143,7 +145,7 @@ object GitCommands {
return Output.error("bad_args", "not a git repository address (expected kind ${GitRepositoryEvent.KIND}, got ${addr.kind})")
}
Context.open(dataDir).use { ctx ->
Context.openOrAnonymous(dataDir).use { ctx ->
ctx.prepare()
val repo = fetchRepo(ctx, addr, args) ?: return Output.error("not_found", "no repository announcement found for $coord")
Output.emit(repoSummary(repo) + mapOf("event_id" to repo.id, "content" to repo.content))
@@ -0,0 +1,990 @@
/*
* 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.GrapeRankCrawler
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
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.CancellationException
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.asCoroutineDispatcher
import kotlinx.coroutines.async
import kotlinx.coroutines.awaitAll
import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.withContext
import java.net.InetSocketAddress
import java.net.Socket
import java.net.URI
import java.util.concurrent.Executors
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 crawl [OBSERVER]` — network only: crawl the reachable graph's
* kind 3/10000/1984/10002 into the local store (aliased as the former `sync`).
* Idempotent and cumulative, so run it a few times to make sure everything is
* loaded. Scores nothing.
* - `amy graperank score [OBSERVER]` — local only: build the graph from the store
* and score (same as bare `--offline`). Instant and param-tunable; repeat with
* different `--rigor`/`--attenuation`/cutoffs without re-crawling.
* - bare `amy graperank [OBSERVER]` — the convenience combo: crawl then score.
*
* Sub-verbs complete the NIP-85 provider experience — the discovery layer that
* lets clients find and consume those assertions:
* - `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()
// Network-wide aggregators that scrape and hold kind:3 for users whose own
// outbox lacks it. The crawler queries these for a straggler's CONTENT (kind:3),
// not just their kind:10002 relay list. Measured on observer 460c25e6, the distinct
// missing authors whose kind:3 each holds: kindpag.es 369, yabu 126, oxtr.dev 76,
// nos.lol 72, ditto 56, nostr1 29, momostr 11, mostr 3. So beyond the profile
// indexers (kindpag/purplepag/coracle/yabu/nostr1) and the ActivityPub bridges
// (ditto/momostr/mostr, which host bridged users' lists), two big general relays --
// nostr.oxtr.dev and nos.lol -- carry ~150 more that no indexer has.
private val CONTENT_AGGREGATOR_RELAYS: Set<NormalizedRelayUrl> =
DefaultIndexerRelayList +
listOf(
"wss://relay.ditto.pub",
"wss://relay.momostr.pink",
"wss://relay.mostr.pub",
"wss://nostr.oxtr.dev",
"wss://nos.lol",
).mapNotNull { RelayUrlNormalizer.normalizeOrNull(it) }.toSet()
private const val PROBE_TIMEOUT_MS = 2000
// The probe does BLOCKING DNS + TCP connect, and dead-domain DNS lookups can hang
// far past the connect timeout. On the shared Dispatchers.IO those hanging lookups
// starve the crawl's own IO — measured +462s on the finishing drain at hop-3. Run
// them on a dedicated, isolated daemon pool instead so the crawl's IO is untouched.
private val probeDispatcher =
Executors
.newFixedThreadPool(128) { r -> Thread(r, "relay-probe").apply { isDaemon = true } }
.asCoroutineDispatcher()
/**
* Cheap reachability pre-probe: a raw TCP connect (one round trip) with a tight
* timeout. Returns false only when the port won't even accept a socket — a dead
* dropper, refusal, or unroutable/onion/LAN host — which the crawler drops into
* deadHosts before the WS path pays its 7s connectTimeout. A busy-but-alive relay
* accepts the SYN instantly at the kernel level (its slowness is at the app layer),
* so it passes here and is left for the real WS attempt. Unparseable host → true,
* so an odd URL is never culled on a parse quirk — let the WS decide.
*/
private suspend fun tcpReachable(relay: NormalizedRelayUrl): Boolean =
withContext(probeDispatcher) {
val hostPort = relayHostPort(relay) ?: return@withContext true
try {
Socket().use { it.connect(InetSocketAddress(hostPort.first, hostPort.second), PROBE_TIMEOUT_MS) }
true
} catch (e: Exception) {
if (e is CancellationException) throw e
false
}
}
private fun relayHostPort(relay: NormalizedRelayUrl): Pair<String, Int>? =
try {
val uri = URI(relay.url)
val host = uri.host ?: return null
val port =
if (uri.port > 0) {
uri.port
} else if (relay.url.startsWith("wss://", ignoreCase = true)) {
443
} else {
80
}
host to port
} catch (e: Exception) {
null
}
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` is the pre-rename name kept as a back-compat alias; `crawl` is
// canonical (disambiguates from negentropy `amy sync` / `graperank update`).
"crawl", "sync" -> crawl(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)
}
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: GrapeRankCrawler.Stats? = null
if (!offline) {
val stats = newCrawler(ctx, args).crawl(observer, builder)
crawlStats = stats
contactListsFed = stats.contactListsFed
flushReachability(ctx, args, stats)
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() },
"contact_lists_by_hop" to crawlStats?.contactsFedByHop.orEmpty().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 crawl`.
* 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,
): GrapeRankCrawler {
val discoveryRelays =
ctx.bootstrapRelays() + Constants.eventFinderRelays + DefaultIndexerRelayList + EXTRA_DISCOVERY_RELAYS
val contentFallback = ctx.bootstrapRelays() + Constants.eventFinderRelays
// Aggregator kind:3 recovery for stragglers is on by default; --no-aggregators
// disables it for A/B comparison.
val aggregators = if (args.bool("no-aggregators")) emptySet() else CONTENT_AGGREGATOR_RELAYS
// Seed the crawl with relays a prior run/monitor proved dead within the cache's
// TTL, so the WS path never re-pays their connect timeouts (--no-reachability-cache
// to skip). The crawl's own final live/dead set is flushed back by the caller.
val knownDead =
if (args.bool("no-reachability-cache")) emptySet() else ctx.reachability.snapshot().dead
return GrapeRankCrawler(
client = ctx.client,
store = ctx.store,
limiter = ctx.relayLimiter,
config =
GrapeRankCrawler.Config(
relayListDiscoveryRelays = discoveryRelays,
knownDeadRelays = knownDead,
contentFallbackRelays = contentFallback,
contentAggregatorRelays = aggregators,
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),
timeoutEvictStrikes = args.intFlag("timeout-evict", 3),
// Cheap TCP reachability pre-probe (--no-probe to disable). No Tor
// transport here, so .onion relays are skipped on sight.
reachabilityProbe = if (args.bool("no-probe")) null else ::tcpReachable,
torEnabled = false,
// shedDeadDiscovery / shardRotations keep their benchmarked-best
// Config defaults.
),
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()}")
}
}
/**
* Flush the crawl's final live/dead relay verdicts into the shared reachability
* cache (NIP-66 kind:30166) so the next crawl and the WoT updater start warm and
* skip proven-dead relays. Best-effort and behind `--no-reachability-cache`: a
* cache write must never fail the crawl it is summarizing.
*/
private suspend fun flushReachability(
ctx: Context,
args: Args,
stats: GrapeRankCrawler.Stats,
) {
if (args.bool("no-reachability-cache")) return
runCatching {
ctx.reachability.record(reachable = stats.liveRelays, dead = stats.deadRelays)
System.err.println(
"[graperank] reachability cache: recorded ${stats.liveRelays.size} live, ${stats.deadRelays.size} dead",
)
}.onFailure { System.err.println("[graperank] reachability cache flush failed: ${it.message}") }
}
/**
* `amy graperank crawl [OBSERVER]` — network-only WoT data crawl (aliased as the
* former `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 crawl(
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)
flushReachability(ctx, args, stats)
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() },
"contact_lists_by_hop" to stats.contactsFedByHop.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
}
/**
* `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()
// Skip relays a crawl/monitor proved dead within the cache's TTL — a dead
// relay cannot serve its authors, so reconciling it only burns a timeout.
// Live author-advertised relays are always synced (--no-reachability-cache
// to reconcile every relay regardless).
val knownDead =
if (args.bool("no-reachability-cache")) emptySet() else ctx.reachability.snapshot().dead
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,
knownDead = knownDead,
),
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 crawl` 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
* 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
}
}
@@ -77,7 +77,7 @@ object KeyCommands {
Output.emit(mapOf("valid" to false))
return 0
}
val npub = hex!!.hexToByteArray().toNpub()
val npub = hex.hexToByteArray().toNpub()
Output.emit(mapOf("valid" to true, "pubkey" to hex, "npub" to npub))
return 0
}
@@ -0,0 +1,167 @@
/*
* 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.DataDir
import com.vitorpamplona.amethyst.cli.Output
import com.vitorpamplona.quartz.nip01Core.jackson.JacksonMapper
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.store.fs.FsEventStore
import java.io.File
/**
* `amy logoff [--yes] [--keep-events]` — log off an account and clear its
* local data.
*
* "Logging off" a CLI with no server session means removing everything the
* account left on this machine:
* - the identity file and any backend-held secret (keychain / ncryptsec /
* plaintext) — via [DataDir.deleteIdentity],
* - the rest of the per-account directory `~/.amy/<account>/` (run-state
* cursors, aliases, cashu counters, all MLS/Marmot state),
* - the active-account pin at `~/.amy/current`, if it points here,
* - and the account's events in the SHARED store at
* `~/.amy/shared/events-store/`.
*
* The event store is shared across every account on the machine, so this
* does NOT wipe it wholesale — it deletes only the events that involve this
* account: those it authored (`authors`) plus those addressed to it via a
* `#p` tag (inbound gift wraps, nutzaps, reactions, mentions…). Other
* accounts' cached events are untouched. Pass `--keep-events` to leave the
* shared cache alone and only remove the identity + per-account state.
*
* The account is selected the normal way (the `--account` flag, the
* `current` pin, or the sole account) — when more than one account exists
* and none is pinned, [DataDir.resolve] already errors out asking the caller
* to disambiguate, so logoff never guesses which account to destroy.
*
* Reads the public key straight from `identity.json` (never unlocking the
* private key), so it needs no passphrase and pops no keychain prompt.
*
* Requires `--yes` to execute, because it is destructive and cannot be
* undone — the private key is gone with the identity file. Without `--yes`
* the command reports what it would delete and exits with code 2.
*/
object LogoffCommand {
suspend fun run(
dataDir: DataDir,
tail: Array<String>,
): Int {
val confirmed = tail.any { it == "--yes" || it == "-y" }
val keepEvents = tail.any { it == "--keep-events" }
// Read the on-disk identity metadata only — no SecretStore round-trip,
// so we never prompt for a passphrase or trip a keychain dialog just
// to log off.
val idFile =
dataDir.loadIdentityFileOrNull()
?: return Output.error(
"no_account",
"no identity at ${dataDir.identityFile.absolutePath}; nothing to log off",
)
val pubkey = idFile.pubKeyHex
val marker = File(DataDir.DEFAULT_ROOT, DataDir.CURRENT_MARKER_NAME)
val isPinned = marker.isFile && marker.readText().trim() == dataDir.accountName
// Everything the account touched in the shared store: authored by it,
// or addressed to it via a #p tag (gift wraps, nutzaps, reactions…).
val involvedFilters =
listOf(
Filter(authors = listOf(pubkey)),
Filter(tags = mapOf("p" to listOf(pubkey))),
)
if (!confirmed) {
val eventCount = if (keepEvents) 0 else withStore(dataDir) { it.count(involvedFilters) }
Output.emit(
mapOf(
"dry_run" to true,
"account" to dataDir.accountName,
"npub" to idFile.npub,
"pubkey" to pubkey,
"account_dir" to dataDir.root.absolutePath,
"pinned" to isPinned,
"events_to_purge" to eventCount,
"keep_events" to keepEvents,
"detail" to "pass --yes to permanently delete this account's key, local state" +
(if (keepEvents) "" else ", and cached events"),
),
)
return 2
}
// 1. Purge the account's events from the shared store.
var purged = 0
if (!keepEvents) {
withStore(dataDir) { store ->
val before = store.count(involvedFilters)
store.delete(involvedFilters)
purged = (before - store.count(involvedFilters)).coerceAtLeast(0)
}
}
// 2. Remove the identity file and any backend-held secret.
dataDir.deleteIdentity()
// 3. Wipe the rest of the per-account directory (run-state, aliases,
// cashu counters, Marmot/MLS state). The shared events-store lives
// outside this directory, so it is not affected.
val dirFullyRemoved = dataDir.root.deleteRecursively()
// 4. Drop the active-account pin if it pointed at this account.
val clearedPin = isPinned && marker.delete()
Output.emit(
mapOf(
"logoff" to true,
"account" to dataDir.accountName,
"npub" to idFile.npub,
"events_purged" to purged,
"removed_dir" to dataDir.root.absolutePath,
"dir_fully_removed" to dirFullyRemoved,
"cleared_pin" to clearedPin,
),
)
return 0
}
/**
* Open the shared [FsEventStore] directly — logoff needs the store but no
* identity, signer, or relays, so it skips [com.vitorpamplona.amethyst.cli.Context.open]
* (which requires a bootstrapped identity). Mirrors `StoreCommands.withStore`.
*/
private inline fun <T> withStore(
dataDir: DataDir,
body: (FsEventStore) -> T,
): T {
val store =
FsEventStore(
root = dataDir.eventsDir.toPath(),
eventToJson = JacksonMapper::toJsonPretty,
)
try {
return body(store)
} finally {
store.close()
}
}
}
@@ -79,7 +79,7 @@ object NappletCommands {
val extraRelays = StaticSiteFetch.commaList(args.flag("relay"))
val timeoutSecs = args.longFlag("timeout", 8L)
Context.open(dataDir).use { ctx ->
Context.openOrAnonymous(dataDir).use { ctx ->
ctx.prepare()
val authorHex = ctx.requireUserHex(author)
val relays =
@@ -134,7 +134,7 @@ object NappletCommands {
val extraRelays = StaticSiteFetch.commaList(args.flag("relay"))
val timeoutSecs = args.longFlag("timeout", 8L)
Context.open(dataDir).use { ctx ->
Context.openOrAnonymous(dataDir).use { ctx ->
ctx.prepare()
val authorHex = ctx.requireUserHex(author)
val relays =
@@ -193,7 +193,7 @@ object NappletCommands {
val extraServers = StaticSiteFetch.commaList(args.flag("server"))
val extraRelays = StaticSiteFetch.commaList(args.flag("relay"))
Context.open(dataDir).use { ctx ->
Context.openOrAnonymous(dataDir).use { ctx ->
ctx.prepare()
val relays =
extraRelays
@@ -81,7 +81,7 @@ object NostrConnect {
}
}
if (secret == null) return null
return Offer(clientPubkey, relays, secret!!, name)
return Offer(clientPubkey, relays, secret, name)
}
private fun buildOffer(
@@ -79,7 +79,7 @@ object NsiteCommands {
val extraRelays = StaticSiteFetch.commaList(args.flag("relay"))
val timeoutSecs = args.longFlag("timeout", 8L)
Context.open(dataDir).use { ctx ->
Context.openOrAnonymous(dataDir).use { ctx ->
ctx.prepare()
val authorHex = ctx.requireUserHex(author)
val relays =
@@ -145,7 +145,7 @@ object NsiteCommands {
val extraRelays = StaticSiteFetch.commaList(args.flag("relay"))
val timeoutSecs = args.longFlag("timeout", 8L)
Context.open(dataDir).use { ctx ->
Context.openOrAnonymous(dataDir).use { ctx ->
ctx.prepare()
val authorHex = ctx.requireUserHex(author)
val relays =
@@ -203,7 +203,7 @@ object NsiteCommands {
val extraServers = StaticSiteFetch.commaList(args.flag("server"))
val extraRelays = StaticSiteFetch.commaList(args.flag("relay"))
Context.open(dataDir).use { ctx ->
Context.openOrAnonymous(dataDir).use { ctx ->
ctx.prepare()
val authorHex = ctx.requireUserHex(author)
@@ -109,7 +109,7 @@ object OfferCommands {
}
/** Local decode of a `noffer` pointer — no network, no account needed. */
private fun info(rest: Array<String>): Int {
internal fun info(rest: Array<String>): Int {
val args = Args(rest)
val offer =
ClinkPointerParser.parse(args.positional(0, "noffer").trim()) as? NOffer
@@ -44,7 +44,7 @@ object OutboxCommand {
val refresh = args.bool("refresh")
val timeoutMs = (args.flag("timeout")?.toLongOrNull() ?: 8L) * 1000
Context.open(dataDir).use { ctx ->
Context.openOrAnonymous(dataDir).use { ctx ->
ctx.prepare()
val pubkey = ctx.requireUserHex(user)
@@ -219,7 +219,9 @@ object Podcast20Commands {
): Int {
val args = Args(rest)
val limit = args.intFlag("limit", 50)
Context.open(dataDir).use { ctx ->
// Read-only: runs anonymously when there is no account (pass a USER to
// list someone else's episodes).
Context.openOrAnonymous(dataDir).use { ctx ->
ctx.prepare()
val author = args.positionalOrNull(0)?.let { ctx.requireUserHex(it) } ?: ctx.identity.pubKeyHex
val relays = RawEventSupport.queryTargets(ctx, args)
@@ -136,7 +136,9 @@ object PodcastCommands {
): Int {
val args = Args(rest)
val limit = args.intFlag("limit", 50)
Context.open(dataDir).use { ctx ->
// Read-only: runs anonymously when there is no account (pass a USER to
// list someone else's podcasts).
Context.openOrAnonymous(dataDir).use { ctx ->
ctx.prepare()
val author = args.positionalOrNull(0)?.let { ctx.requireUserHex(it) } ?: ctx.identity.pubKeyHex
val relays = RawEventSupport.queryTargets(ctx, args)
@@ -63,7 +63,9 @@ object ProfileCommands {
val args = Args(rest)
val refresh = args.bool("refresh")
val timeoutSecs = args.longFlag("timeout", 8L)
Context.open(dataDir).use { ctx ->
// Read-only: runs anonymously when there is no account (an explicit
// USER is then required, since there is no "own" profile to default to).
Context.openOrAnonymous(dataDir).use { ctx ->
ctx.prepare()
val pubKey =
args.positionalOrNull(0)?.let { ctx.requireUserHex(it) }
@@ -55,7 +55,7 @@ object PublishCommand {
return Output.error("invalid_event", "event id/signature does not verify — refusing to publish")
}
Context.open(dataDir).use { ctx ->
Context.openOrAnonymous(dataDir).use { ctx ->
ctx.prepare()
val targets = RawEventSupport.publishTargets(ctx, args)
if (targets.isEmpty()) {
@@ -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,
@@ -145,7 +145,7 @@ object SearchCommand {
timeoutMs: Long,
render: (List<Event>) -> List<Map<String, Any?>>,
): Int {
Context.open(dataDir).use { ctx ->
Context.openOrAnonymous(dataDir).use { ctx ->
ctx.prepare()
val relays =
SearchActions.resolveSearchRelays(
@@ -0,0 +1,167 @@
/*
* 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.fasterxml.jackson.module.kotlin.readValue
import com.vitorpamplona.amethyst.cli.DataDir
import com.vitorpamplona.amethyst.cli.Output
import com.vitorpamplona.amethyst.cli.RunState
import com.vitorpamplona.amethyst.cli.StoreStats
import com.vitorpamplona.amethyst.cli.secrets.IdentityFile
import com.vitorpamplona.amethyst.cli.secrets.IdentitySecret
import java.io.File
/**
* `amy status` — a single at-a-glance overview of everything amy is
* holding on disk under `~/.amy/`. Built for the returning user: "I
* haven't run this in months — what accounts do I have, which one is
* active, can they still sign, and how big is the local database?"
*
* Cross-account by design, so it dispatches *before* account resolution
* (like `use`) and never fails on "zero accounts" or "ambiguous account".
* It is strictly read-only and metadata-only: it parses the on-disk
* `identity.json` / `state.json` / `aliases.json` and walks the shared
* event store, but it never unlocks a private key (no keychain prompt,
* no NIP-49 passphrase) and never touches the network.
*
* Per account it reports the npub, how the key is stored (local keychain
* / ncryptsec / plaintext, a NIP-46 bunker, or read-only), whether it can
* sign, and the local footprint that account has accumulated: aliases,
* Marmot groups, a published KeyPackage bundle, a Cashu wallet, and the
* sync cursors that tell catch-up commands where they left off.
*/
object StatusCommand {
fun run(tail: Array<String>): Int {
// `status` takes no positional args; tolerate an accidental one
// rather than erroring — it's a read-only inspection command.
val rootBase = DataDir.DEFAULT_ROOT
val currentPin =
File(rootBase, DataDir.CURRENT_MARKER_NAME)
.takeIf { it.isFile }
?.readText()
?.trim()
?.ifEmpty { null }
val accountNames = DataDir.listAccounts(rootBase)
val accounts = accountNames.map { accountRow(File(rootBase, it), it, it == currentPin) }
// The event store is shared across every account.
val store = StoreStats.of(File(rootBase, "shared/events-store").toPath())
Output.emit(
mapOf(
"root" to rootBase.absolutePath,
"current" to currentPin,
"account_count" to accounts.size,
"accounts" to accounts,
"store" to
mapOf(
"events" to store.events,
"distinct_kinds" to store.distinctKinds,
"disk_bytes" to store.diskBytes,
"oldest_at" to store.oldestAt,
"newest_at" to store.newestAt,
"root" to store.root.toString(),
),
),
)
return 0
}
private fun accountRow(
accountRoot: File,
name: String,
isCurrent: Boolean,
): Map<String, Any?> {
val identity = readIdentity(File(accountRoot, "identity.json"))
val signer = classifySigner(identity)
val marmotGroups =
File(accountRoot, "marmot/groups")
.listFiles { f -> f.name.endsWith(".state") }
?.size ?: 0
val hasKeyPackage = File(accountRoot, "marmot/keypackages.bundle").isFile
val hasCashuWallet = File(accountRoot, "cashu.json").isFile
val aliasCount = readAliases(File(accountRoot, "aliases.json")).size
val runState = readRunState(File(accountRoot, "state.json"))
// LinkedHashMap so the text renderer prints fields in this order.
val row = LinkedHashMap<String, Any?>()
row["name"] = name
row["current"] = isCurrent
row["npub"] = identity?.npub
row["hex"] = identity?.pubKeyHex
row["signer"] = signer.kind
row["key_storage"] = signer.storage
row["can_sign"] = signer.canSign
if (signer.bunkerRelays != null) row["bunker_relays"] = signer.bunkerRelays
row["aliases"] = aliasCount
row["marmot_groups"] = marmotGroups
row["key_package_published"] = hasKeyPackage
row["cashu_wallet"] = hasCashuWallet
row["dm_cursor_at"] = runState.giftWrapSince
row["marmot_group_cursors"] = runState.groupSince.size
return row
}
/**
* How this account can sign, derived purely from the on-disk
* [IdentityFile] — never resolves the secret itself.
* - `local` — an on-device private key ([storage] says where).
* - `bunker` — a NIP-46 remote signer ([bunkerRelays] lists it).
* - `read-only` — imported from an npub/nprofile/NIP-05; cannot sign.
*/
private data class SignerInfo(
val kind: String,
val storage: String?,
val canSign: Boolean,
val bunkerRelays: List<String>?,
)
private fun classifySigner(identity: IdentityFile?): SignerInfo {
if (identity == null) return SignerInfo("unknown", null, false, null)
identity.bunker?.let { bunker ->
return SignerInfo("bunker", secretStorageLabel(identity.secret), true, bunker.relays)
}
val storage = secretStorageLabel(identity.secret)
return when {
identity.secret != null -> SignerInfo("local", storage, true, null)
// Pre-secret-store data-dirs kept the key inline; still signable.
identity.privKeyHex != null || identity.nsec != null -> SignerInfo("local", "legacy-plaintext", true, null)
else -> SignerInfo("read-only", null, false, null)
}
}
private fun secretStorageLabel(secret: IdentitySecret?): String? =
when (secret) {
is IdentitySecret.Keychain -> "keychain:${secret.backend}"
is IdentitySecret.Ncryptsec -> "ncryptsec"
is IdentitySecret.Plaintext -> "plaintext"
null -> null
}
private fun readIdentity(file: File): IdentityFile? = if (file.isFile) runCatching { Output.mapper.readValue<IdentityFile>(file.readText()) }.getOrNull() else null
private fun readAliases(file: File): Map<String, String> = if (file.isFile) runCatching { Output.mapper.readValue<Map<String, String>>(file.readText()) }.getOrElse { emptyMap() } else emptyMap()
private fun readRunState(file: File): RunState = if (file.isFile) runCatching { Output.mapper.readValue<RunState>(file.readText()) }.getOrElse { RunState() } else RunState()
}
@@ -22,9 +22,13 @@ package com.vitorpamplona.amethyst.cli.commands
import com.vitorpamplona.amethyst.cli.DataDir
import com.vitorpamplona.amethyst.cli.Output
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
@@ -33,19 +37,24 @@ 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.
@@ -68,7 +77,52 @@ object StoreCommands {
),
)
private fun stat(dataDir: DataDir): Int {
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(
"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(
@@ -140,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
@@ -184,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 {
@@ -53,7 +53,7 @@ object SubscribeCommand {
val timeoutMs = args.flag("timeout")?.toLongOrNull()?.let { it * 1000 }
val filter = RawEventSupport.buildFilter(args)
Context.open(dataDir).use { ctx ->
Context.openOrAnonymous(dataDir).use { ctx ->
ctx.prepare()
val relays = RawEventSupport.queryTargets(ctx, args)
if (relays.isEmpty()) return Output.error("no_relays", "no relays available; pass --relay or run `amy relay add`")
@@ -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 12 (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,14 +113,15 @@ 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.open(dataDir).use { ctx ->
Context.openOrAnonymous(dataDir).use { ctx ->
ctx.prepare()
val localEvents = ctx.store.query<Event>(filter)
val localById = localEvents.associateBy { it.id }
@@ -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
}
}