From bb2a83c1fe45a60bb4fcef597e6a454f2b47661f Mon Sep 17 00:00:00 2001 From: nrobi144 Date: Tue, 7 Jul 2026 13:31:41 +0300 Subject: [PATCH] feat(desktop,cli): route WoT kind-3 fetch through OutboxDispatcher (NIP-65) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 3 of the outbox refactor (PR #3483, per Vitor's directive). The WoT service's kind-3 seeding on Desktop and the `amy wot sync` verb now go through OutboxDispatcher — index relays discover each author's kind-10002 write relays, then per-outbox-relay REQs fetch kind-3. Changes: Desktop: - DesktopRelaySubscriptionsCoordinator gains an inner OutboxCacheGateway that bridges DesktopLocalCache (cachedAdvertisedRelayList / consume) to OutboxDispatcher. - New suspend loadKind3ViaOutbox(pubkeys) method returns the dispatcher's Result for observability. - Main.kt WoT-seed effect now: 1. gates on wotService.isDisabled to preserve MAX_FOLLOWS guardrail (fix 2 from Phase 1) 2. calls loadKind3ViaOutbox instead of the direct loadKind3Batched on index relays 3. keeps the 2s markReady safety net for cold-start UX - clear() now also clears outboxDispatcher's dedup markers. amy: - WotCommand.sync rewritten to construct an OutboxDispatcher, buffer events in the gateway, and persist to ctx.store after fetch returns (store.insert is suspending; can't call from non-suspend gateway callbacks). - --json output additively gains kind10002_received, outbox_covered_authors, fallback_authors, persisted keys. - --timeout N still supported; now maps to overallTimeoutMs. Not in this commit (deferred to a follow-up on same PR if reviewers want it): - Routing stranger-avatar kind-0 fetch through the outbox path (MetadataPreloader wiring is more invasive; keeps this diff focused on the primary WoT concern). Plan: commons/plans/2026-07-06-fix-wot-outbox-model-and-review-fixes-plan.md --- .../amethyst/cli/commands/WotCommand.kt | 97 ++++++++++++++----- .../vitorpamplona/amethyst/desktop/Main.kt | 26 ++++- .../DesktopRelaySubscriptionsCoordinator.kt | 58 +++++++++++ 3 files changed, 153 insertions(+), 28 deletions(-) diff --git a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/WotCommand.kt b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/WotCommand.kt index e7bb24b7a8..c81ce32c4b 100644 --- a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/WotCommand.kt +++ b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/WotCommand.kt @@ -24,14 +24,18 @@ 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.filters.Filter -import com.vitorpamplona.quartz.nip02FollowList.ContactListEvent +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 ` — Web-of-Trust score queries. @@ -113,41 +117,88 @@ object WotCommand { rest: Array, ): Int { val args = Args(rest) - val timeoutMs = args.flag("timeout")?.toLongOrNull()?.times(1000) ?: 5_000L + // 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()?.toList() + 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 } - // Index relays — shared with the Desktop app via - // `java.util.prefs`. Falls back to - // `PreferencesIndexRelays.DEFAULT_INDEX_RELAYS` when the - // user hasn't configured anything, so this is never empty - // in practice. val relays = ctx.indexRelays() if (relays.isEmpty()) return Output.error("no_relays", "no index relays configured") - // Chunk authors into ≤100 per Filter for relays with per-filter caps. - val filters = - follows.chunked(100).map { chunk -> - Filter( - kinds = listOf(ContactListEvent.KIND), - authors = chunk, - limit = chunk.size, + 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()) + 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` 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 received = ctx.drain(relays.associateWith { filters }, timeoutMs) - val kind3Events = received.mapNotNull { it.second as? ContactListEvent } - // Persist to store so future `get` / `list` see them. - kind3Events.forEach { runCatching { ctx.store.insert(it) } } - Output.emit(mapOf("received" to kind3Events.size, "followers" to follows.size)) - return 0 + + 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() + } } } diff --git a/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/Main.kt b/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/Main.kt index e1fa2596e5..e8ad5d343d 100644 --- a/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/Main.kt +++ b/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/Main.kt @@ -1576,16 +1576,32 @@ fun MainContent( iAccount.wotService.applyKind3(evt.pubKey, evt.verifiedFollowKeySet()) } } - // React to changes in the active user's follow set. + // React to changes in the active user's follow set. Under the + // outbox model (PR #3483 review directive from Vitor) kind-3 + // fetch goes to each author's declared write relays instead of a + // static index-relay broadcast — the OutboxDispatcher does the + // NIP-65 discovery, transposes with RelayListRecommendationProcessor + // and issues per-outbox-relay REQs. Falls back to index relays + // for authors that never returned a 10002. launch { localCache.followedUsers.collect { follows -> iAccount.wotService.onFollowSetChange(follows, account.pubKeyHex) - if (follows.isNotEmpty()) { - subscriptionsCoordinator.loadKind3Batched(follows) { + when { + iAccount.wotService.isDisabled.value -> { + // Guardrail — mega-follow accounts skip WoT + // entirely so we don't dispatch a batch that + // would be discarded anyway. iAccount.wotService.markReadyOnce() } - } else { - iAccount.wotService.markReadyOnce() + follows.isEmpty() -> { + iAccount.wotService.markReadyOnce() + } + else -> { + launch { + subscriptionsCoordinator.loadKind3ViaOutbox(follows) + iAccount.wotService.markReadyOnce() + } + } } } } diff --git a/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/subscriptions/DesktopRelaySubscriptionsCoordinator.kt b/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/subscriptions/DesktopRelaySubscriptionsCoordinator.kt index a159730f5e..33afa1921d 100644 --- a/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/subscriptions/DesktopRelaySubscriptionsCoordinator.kt +++ b/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/subscriptions/DesktopRelaySubscriptionsCoordinator.kt @@ -25,6 +25,8 @@ import com.vitorpamplona.amethyst.commons.relayClient.assemblers.FeedMetadataCoo import com.vitorpamplona.amethyst.commons.relayClient.preload.MetadataPreloader import com.vitorpamplona.amethyst.commons.relayClient.preload.MetadataRateLimiter import com.vitorpamplona.amethyst.commons.service.BasicBundledInsert +import com.vitorpamplona.amethyst.commons.wot.OutboxCacheGateway +import com.vitorpamplona.amethyst.commons.wot.OutboxDispatcher import com.vitorpamplona.amethyst.desktop.cache.DesktopLocalCache import com.vitorpamplona.amethyst.desktop.model.DesktopDmRelayState import com.vitorpamplona.quartz.nip01Core.core.Event @@ -33,6 +35,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import com.vitorpamplona.quartz.nip65RelayList.AdvertisedRelayListEvent import com.vitorpamplona.quartz.utils.Log import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineScope @@ -98,6 +101,51 @@ class DesktopRelaySubscriptionsCoordinator( }, ) + /** + * Bridges [OutboxDispatcher] to [DesktopLocalCache]. Every event the + * dispatcher receives goes through [DesktopLocalCache.consume] so it + * lands in the same code path as events arriving from feed + * subscriptions — kind-10002 caches into `addressableNotes`; kind-0 + * updates the user metadata; kind-3 fans out through + * `_contactListEvents` for the WoT service. + */ + private val outboxGateway = + object : OutboxCacheGateway { + override fun cachedOutbox(pubkey: HexKey): AdvertisedRelayListEvent? = localCache.cachedAdvertisedRelayList(pubkey) + + override fun onOutboxDiscovered( + event: AdvertisedRelayListEvent, + relay: NormalizedRelayUrl, + ) { + localCache.consume(event, relay) + } + + override fun onDiscoveredEvent( + event: Event, + relay: NormalizedRelayUrl, + ) { + localCache.consume(event, relay) + } + } + + /** + * NIP-65 outbox model for kind-0 and kind-3 fetching. Per PR #3483 + * review directive from Vitor: index relays discover each author's + * write-relay list, then kind-0/kind-3 REQs go to that author's + * declared write relays. See [OutboxDispatcher] for the pipeline. + * + * Kept as a val (not lazy) because [clear] must reset its dedup + * markers on account switch. The dispatcher itself is stateless + * across accounts as long as `clear()` is called. + */ + val outboxDispatcher = + OutboxDispatcher( + client = client, + scope = scope, + indexRelays = { indexRelays }, + gateway = outboxGateway, + ) + // Event bundler: batches consumed notes before emitting to SharedFlow // 250ms for desktop (Android uses 1000ms to save battery) private val eventBundler = @@ -387,10 +435,20 @@ class DesktopRelaySubscriptionsCoordinator( unsubscribeFromDms() feedMetadata.clear() + outboxDispatcher.clear() rateLimiter.reset() cleanupJob?.cancel() } + /** + * Fetch kind-3 (follow lists) for [pubkeys] via each author's outbox + * relay per NIP-65 (see [OutboxDispatcher]). Suspends until every + * phase EOSEs or times out. Callers typically launch this on a + * scope-owned coroutine and mark the WoT service ready in the + * continuation. Returns per-phase counters for observability. + */ + suspend fun loadKind3ViaOutbox(pubkeys: Set): OutboxDispatcher.Result = outboxDispatcher.fetchKind3Only(pubkeys) + // ----- Memory Cleanup ----- private val memoryBean = ManagementFactory.getMemoryMXBean()