diff --git a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/GrapeRankCommand.kt b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/GrapeRankCommand.kt index b5cde8a381..c41c71b999 100644 --- a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/GrapeRankCommand.kt +++ b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/GrapeRankCommand.kt @@ -30,28 +30,22 @@ import com.vitorpamplona.quartz.experimental.graperank.GrapeRank import com.vitorpamplona.quartz.experimental.graperank.GrapeRankDataCrawler import com.vitorpamplona.quartz.experimental.graperank.GrapeRankParams import com.vitorpamplona.quartz.experimental.graperank.GrapeRankPublisher +import com.vitorpamplona.quartz.experimental.graperank.GrapeRankUpdater import com.vitorpamplona.quartz.experimental.graperank.TrustGraphBuilder import com.vitorpamplona.quartz.nip01Core.core.Event import com.vitorpamplona.quartz.nip01Core.core.HexKey import com.vitorpamplona.quartz.nip01Core.core.toHexKey import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair -import com.vitorpamplona.quartz.nip01Core.metadata.MetadataEvent -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.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.nip01Core.store.IdAndTime 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.nip65RelayList.AdvertisedRelayListEvent import com.vitorpamplona.quartz.nip85TrustedAssertions.list.TrustProviderListEvent import com.vitorpamplona.quartz.nip85TrustedAssertions.list.serviceProviders import com.vitorpamplona.quartz.nip85TrustedAssertions.list.tags.ProviderTypes @@ -62,14 +56,7 @@ import com.vitorpamplona.quartz.nip85TrustedAssertions.users.tags.RankTag import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.async import kotlinx.coroutines.awaitAll -import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.coroutineScope -import kotlinx.coroutines.joinAll -import kotlinx.coroutines.launch -import kotlinx.coroutines.sync.Semaphore -import kotlinx.coroutines.sync.withPermit -import java.util.Collections -import java.util.concurrent.atomic.AtomicInteger import kotlin.math.roundToInt /** @@ -446,59 +433,18 @@ object GrapeRankCommand { return 0 } - // ── graperank update: outbox-model refresh of the WoT record kinds ────────── - - /** WoT record kinds a GrapeRank score is a function of, refreshed by `update`. */ - private val UPDATE_KINDS = - listOf( - MetadataEvent.KIND, // 0 — profiles - ContactListEvent.KIND, // 3 — follows - AdvertisedRelayListEvent.KIND, // 10002 — outbox relay lists - ReportEvent.KIND, // 1984 — reports - ) - - /** ids per reconcile chunk and per by-id fetch (mirrors [SyncCommand]). */ - private const val UPDATE_ID_CHUNK = 500 - - /** Concurrent by-id download REQs per relay (mirrors [SyncCommand]). */ - private const val UPDATE_DOWNLOAD_WORKERS = 4 - - /** Overlapped `created_at`-window reconciles after an over-cap split. */ - private const val UPDATE_RECONCILE_CONCURRENCY = 2 - - /** Cap on deletion-settle rounds; a healthy group converges in 1–2. */ - private const val UPDATE_MAX_DELETION_ROUNDS = 4 - /** * `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. * - * Unlike `sync` (which walks the reachable follow graph outward from an - * observer), this is a store-driven refresh: it reads every kind:10002 already - * in the local store, inverts them into a `write-relay -> authors` map (the - * outbox model — an author's events live on the relays they write to), then - * runs ONE NIP-77 negentropy reconcile per write relay scoped to exactly the - * authors who publish there. So each relay is asked only for the authors it - * actually hosts, and each author is reconciled only against their own relays. - * - * Bidirectional by default (`--down`/`--up` narrow it): - * - **down** downloads records the relay has and we lack (by-id REQ); - * - **up** uploads records we have and the relay lacks (EVENT). - * - * After the content pass each group runs [negentropySettleDeletions] over the - * reconcile residual (disable with `--no-sync-deletions`). Its **applyDown** - * direction is the "download the deletion when our upload was rejected" case: - * an event we pushed up that the relay keeps rejecting (it deleted it) surfaces - * as a residual *have*, so we pull the relay's covering kind:5 down and apply it - * locally — our store drops the event the author retracted. The **sendUp** - * direction publishes OUR covering deletions for records we deleted that the - * relay still serves. Cheap: it works off the small residual, not the whole set. - * - * If negentropy can't reconcile a relay (no NIP-77 support, an over-cap minimal - * window, a mid-sync disconnect, …) the group falls back to a full paged download - * ([Context.drainAllPages]) of the same authors+kinds, so those records are still - * refreshed — only the deletion settle (negentropy-only) is skipped there. + * 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), @@ -512,53 +458,38 @@ object GrapeRankCommand { rest: Array, ): Int { val args = Args(rest) - val timeoutMs = args.longFlag("timeout", 30L) * 1000 - val relayConcurrency = args.intFlag("relay-concurrency", 4).coerceAtLeast(1) - val authorChunk = args.intFlag("author-chunk", 500).coerceAtLeast(1) - val minAuthors = args.intFlag("min-authors", 1).coerceAtLeast(1) val reportLimit = args.intFlag("report-limit", 50).coerceAtLeast(0) - val syncDeletions = !args.bool("no-sync-deletions") // Default is bidirectional; a single --down/--up narrows to that direction. val downFlag = args.bool("down") val upFlag = args.bool("up") - val down = downFlag || !upFlag - val up = upFlag || !downFlag Context.openOrAnonymous(dataDir).use { ctx -> ctx.prepare() - // Every author's outbox relays, read from the kind:10002 already in the - // store. Replaceable, so the store holds the latest per author; guard with - // a createdAt max in case both an old and new copy linger. - val latestRelayList = HashMap() - for (event in ctx.store.query(Filter(kinds = listOf(AdvertisedRelayListEvent.KIND)))) { - if (event !is AdvertisedRelayListEvent) continue - val prev = latestRelayList[event.pubKey] - if (prev == null || event.createdAt > prev.createdAt) latestRelayList[event.pubKey] = event - } + val updater = + GrapeRankUpdater( + client = ctx.client, + store = ctx.store, + config = + GrapeRankUpdater.Config( + down = downFlag || !upFlag, + up = upFlag || !downFlag, + syncDeletions = !args.bool("no-sync-deletions"), + relayConcurrency = args.intFlag("relay-concurrency", 4), + authorChunk = args.intFlag("author-chunk", 500), + minAuthors = args.intFlag("min-authors", 1), + idleTimeoutMs = args.longFlag("timeout", 30L) * 1000, + ), + log = { System.err.println(it) }, + ) - // Invert to write-relay -> authors (the outbox model). An author with no - // write-marked relays contributes nothing (nowhere to reconcile them). - val relayToAuthors = HashMap>() - var authorsWithOutbox = 0 - for ((author, list) in latestRelayList) { - val writes = list.writeRelaysNorm() ?: continue - authorsWithOutbox++ - for (relay in writes) relayToAuthors.getOrPut(relay) { HashSet() }.add(author) - } + val result = updater.update() - // Drop relays hosting fewer than --min-authors of our authors; largest - // first so the heaviest groups start while permits are free. - val groups = - relayToAuthors.entries - .filter { it.value.size >= minAuthors } - .sortedByDescending { it.value.size } - - if (groups.isEmpty()) { + if (result.relays == 0) { Output.emit( linkedMapOf( - "relay_lists_in_store" to latestRelayList.size, - "authors_with_outbox" to authorsWithOutbox, + "relay_lists_in_store" to result.relayListsInStore, + "authors_with_outbox" to result.authorsWithOutbox, "relays" to 0, "note" to "no kind:10002 write relays in the local store — run `graperank sync` first", ), @@ -566,99 +497,40 @@ object GrapeRankCommand { return 0 } - val downloaded = AtomicInteger(0) - val uploaded = AtomicInteger(0) - val deletionsUp = AtomicInteger(0) - val deletionsDown = AtomicInteger(0) - val relaysOk = AtomicInteger(0) - val relaysFailed = AtomicInteger(0) - val relaysPaged = AtomicInteger(0) - val perRelay = Collections.synchronizedList(ArrayList>()) - - val gate = Semaphore(relayConcurrency) - coroutineScope { - groups.forEach { (relay, authors) -> - launch { - gate.withPermit { - val authorList = authors.toList() - var relayDownloaded = 0 - var relayUploaded = 0 - var relayDelUp = 0 - var relayDelDown = 0 - var relayNeed = 0 - var relayHave = 0 - var failed = false - var paged = false - var error: String? = null - - for (chunk in authorList.chunked(authorChunk)) { - val filter = Filter(kinds = UPDATE_KINDS, authors = chunk) - val res = syncGroup(ctx, relay, filter, down, up, syncDeletions, timeoutMs) - relayDownloaded += res.downloaded - relayUploaded += res.uploaded - relayDelUp += res.deletionsSentUp - relayDelDown += res.deletionsAppliedDown - relayNeed += res.need - relayHave += res.have - if (res.pagedFallback) paged = true - if (res.error != null) { - failed = true - error = res.error - } - } - - downloaded.addAndGet(relayDownloaded) - uploaded.addAndGet(relayUploaded) - deletionsUp.addAndGet(relayDelUp) - deletionsDown.addAndGet(relayDelDown) - if (failed) relaysFailed.incrementAndGet() else relaysOk.incrementAndGet() - if (paged) relaysPaged.incrementAndGet() - - System.err.println( - "[graperank update] ${relay.url}: ${authors.size} authors, " + - "down $relayDownloaded, up $relayUploaded, del↑ $relayDelUp, del↓ $relayDelDown" + - (if (paged) " (paged fallback)" else "") + - (if (error != null) " (error: $error)" else ""), - ) - perRelay.add( - linkedMapOf( - "relay" to relay.url, - "authors" to authors.size, - "need" to relayNeed, - "have" to relayHave, - "downloaded" to relayDownloaded, - "uploaded" to relayUploaded, - "deletions_sent_up" to relayDelUp, - "deletions_applied_down" to relayDelDown, - "paged_fallback" to paged, - "error" to error, - ), - ) - } - } - } - } - // Busiest relays first, capped so a many-thousand-relay run still emits a // bounded JSON object; totals below always cover every relay. val report = - perRelay - .sortedByDescending { (it["downloaded"] as Int) + (it["uploaded"] as Int) } + result.perRelay + .sortedByDescending { it.downloaded + it.uploaded } .take(reportLimit) + .map { + linkedMapOf( + "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( - "kinds" to UPDATE_KINDS, - "relay_lists_in_store" to latestRelayList.size, - "authors_with_outbox" to authorsWithOutbox, - "relays" to groups.size, - "relays_ok" to relaysOk.get(), - "relays_failed" to relaysFailed.get(), - "relays_paged_fallback" to relaysPaged.get(), - "downloaded" to downloaded.get(), - "uploaded" to uploaded.get(), - "deletions_sent_up" to deletionsUp.get(), - "deletions_applied_down" to deletionsDown.get(), + "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, ), @@ -667,146 +539,6 @@ object GrapeRankCommand { } } - /** Outcome of one relay/author-chunk reconcile in [update]. */ - private class GroupSyncResult( - val downloaded: Int, - val uploaded: Int, - val deletionsSentUp: Int, - val deletionsAppliedDown: Int, - val need: Int, - val have: Int, - /** True when negentropy couldn't reconcile and we paged the filter instead. */ - val pagedFallback: Boolean, - val error: String?, - ) - - /** - * Content pass + deletion settle for one relay scoped to [filter] (kinds + - * one author chunk). Mirrors [SyncCommand]'s two-pass structure exactly — the - * reconcile, windowing, and back-pressure all live in quartz's - * [negentropyReconcile] / [negentropySettleDeletions]; this only routes ids to - * [Context.drain] / [Context.publish]. Best-effort: a reconcile failure is - * captured in [GroupSyncResult.error], never thrown, so one bad relay can't - * abort the whole update. - */ - private suspend fun syncGroup( - ctx: Context, - relay: NormalizedRelayUrl, - filter: Filter, - down: Boolean, - up: Boolean, - syncDeletions: Boolean, - timeoutMs: Long, - ): GroupSyncResult { - val localEvents = ctx.store.query(filter) - val localById = localEvents.associateBy { it.id } - val localEntries = localEvents.map { IdAndTime(it.createdAt, it.id) } - - val downloaded = AtomicInteger(0) - val uploaded = AtomicInteger(0) - - val result = - try { - coroutineScope { - // needIds = relay has, we lack; haveIds = we have, relay lacks. - val needBatches = Channel>(UPDATE_DOWNLOAD_WORKERS * 2) - val haveBatches = Channel>(Channel.UNLIMITED) - - val downloaders = - List(UPDATE_DOWNLOAD_WORKERS) { - launch { - for (batch in needBatches) { - downloaded.addAndGet(ctx.drain(mapOf(relay to listOf(Filter(ids = batch))), timeoutMs).size) - } - } - } - val uploader = - launch { - for (batch in haveBatches) { - for (id in batch) { - val ev = localById[id] ?: continue - if (ctx.publish(ev, setOf(relay)).values.any { it }) uploaded.incrementAndGet() - } - } - } - - val reconcile = - try { - ctx.client.negentropyReconcile( - relay = relay, - filter = filter, - localEntries = localEntries, - batchSize = UPDATE_ID_CHUNK, - idleTimeoutMs = timeoutMs, - reconcileConcurrency = UPDATE_RECONCILE_CONCURRENCY, - onHaveIds = if (up) { batch -> haveBatches.send(batch) } else null, - onNeedIds = { batch -> if (down) needBatches.send(batch) }, - ) - } finally { - needBatches.close() - haveBatches.close() - } - - downloaders.joinAll() - uploader.join() - reconcile - } - } catch (e: NegentropySyncException) { - // Negentropy couldn't reconcile this relay (no NIP-77, an over-cap - // minimal window, a disconnect, …). Fall back to a full paged download - // of the SAME authors+kinds so `update` still refreshes the records — - // [Context.drainAllPages] walks each relay past its per-REQ cap and - // verifies+stores every event. Only the download direction has a paging - // analog; upload and the negentropy-only deletion settle are skipped. - var pageError: String? = null - if (down) { - try { - downloaded.addAndGet(ctx.drainAllPages(mapOf(relay to listOf(filter)), timeoutMs).size) - } catch (pe: Exception) { - pageError = "negentropy: ${e.message}; page fallback: ${pe::class.simpleName}: ${pe.message}" - } - } - return GroupSyncResult( - downloaded = downloaded.get(), - uploaded = uploaded.get(), - deletionsSentUp = 0, - deletionsAppliedDown = 0, - need = 0, - have = 0, - pagedFallback = true, - error = pageError, - ) - } - - val deletions = - if (syncDeletions) { - ctx.client.negentropySettleDeletions( - relay = relay, - filter = filter, - store = ctx.store, - sendUp = down, - applyDown = up, - batchSize = UPDATE_ID_CHUNK, - idleTimeoutMs = timeoutMs, - maxRounds = UPDATE_MAX_DELETION_ROUNDS, - reconcileConcurrency = UPDATE_RECONCILE_CONCURRENCY, - ) - } else { - DeletionSettleResult(0, 0, 0) - } - - return GroupSyncResult( - downloaded = downloaded.get(), - uploaded = uploaded.get(), - deletionsSentUp = deletions.sentUp, - deletionsAppliedDown = deletions.appliedDown, - need = result.needCount, - have = result.haveCount, - pagedFallback = false, - error = null, - ) - } - /** * Build + sign one kind:30382 [ContactCardEvent] per (target, rank), fanned * out across CPU cores (id-hash + Schnorr sign is CPU-bound). The signed diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/experimental/graperank/GrapeRankUpdater.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/experimental/graperank/GrapeRankUpdater.kt new file mode 100644 index 0000000000..62ded1e53c --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/experimental/graperank/GrapeRankUpdater.kt @@ -0,0 +1,425 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.quartz.experimental.graperank + +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.core.HexKey +import com.vitorpamplona.quartz.nip01Core.metadata.MetadataEvent +import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.NegentropySyncException +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.fetchAll +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.fetchAllPages +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.negentropyReconcile +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.negentropySettleDeletions +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.publishAndConfirm +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import com.vitorpamplona.quartz.nip01Core.store.IEventStore +import com.vitorpamplona.quartz.nip01Core.store.IdAndTime +import com.vitorpamplona.quartz.nip01Core.store.verifyAndInsert +import com.vitorpamplona.quartz.nip02FollowList.ContactListEvent +import com.vitorpamplona.quartz.nip56Reports.ReportEvent +import com.vitorpamplona.quartz.nip65RelayList.AdvertisedRelayListEvent +import kotlinx.coroutines.async +import kotlinx.coroutines.awaitAll +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.joinAll +import kotlinx.coroutines.launch +import kotlinx.coroutines.sync.Semaphore +import kotlinx.coroutines.sync.withPermit +import kotlin.concurrent.atomics.AtomicInt +import kotlin.concurrent.atomics.ExperimentalAtomicApi + +/** + * Store-driven, outbox-model refresh of the record kinds a [GrapeRank] score is a + * function of — the profiles (kind:0), follows (kind:3), outbox relay lists + * (kind:10002), and reports (kind:1984) of every author already known to the + * local [store]. + * + * Where [GrapeRankDataCrawler] discovers the graph by walking follows outward from + * an observer, this refreshes what is *already* known: it reads every kind:10002 in + * the store, inverts them into a `write-relay -> authors` map (the outbox model — an + * author's events live on the relays they write to), then runs one NIP-77 negentropy + * reconcile per write relay scoped to exactly the authors who publish there. So each + * relay is asked only for the authors it hosts, and each author is reconciled only + * against their own relays. Run it periodically to keep a scored network current + * without paying a full from-scratch crawl. + * + * Each per-relay group is bidirectional by default ([Config.down] / [Config.up]): + * - **down** downloads records the relay has and the store lacks (by-id fetch, + * verified + inserted into [store]); + * - **up** uploads records the store has and the relay lacks. + * + * After the content pass each group settles deletions over the reconcile residual + * ([negentropySettleDeletions], [Config.syncDeletions]). Its **applyDown** direction + * is the "download the deletion when our upload was rejected" case: an event pushed up + * that the relay keeps rejecting (it deleted it) surfaces as a residual *have*, so the + * relay's covering kind:5 is pulled down and applied locally and the store drops the + * retracted record. The **sendUp** direction publishes the store's covering deletions + * for records deleted locally that the relay still serves. Cheap: it works off the + * small residual, not the whole set. + * + * If negentropy can't reconcile a relay (no NIP-77 support, an over-cap minimal window, + * a mid-sync disconnect, …) and [Config.pageFallback] is on, the group falls back to a + * full paged download ([fetchAllPages]) of the same authors+kinds so those records are + * still refreshed — only the negentropy-only deletion settle is skipped there. + * + * Transport-agnostic within quartz: it takes an [INostrClient] and an [IEventStore]. + * Progress is emitted through [log]; a headless caller routes it to stderr, a UI ignores it. + */ +@OptIn(ExperimentalAtomicApi::class) +class GrapeRankUpdater( + private val client: INostrClient, + private val store: IEventStore, + private val config: Config = Config(), + private val log: (String) -> Unit = {}, +) { + /** + * @param kinds the record kinds refreshed per author (default: the WoT set + * 0 / 3 / 10002 / 1984). + * @param down download records the relay has that the store lacks. + * @param up upload records the store has that the relay lacks (also arms the + * deletion **applyDown** path — a rejected upload pulls the relay's kind:5 down). + * @param syncDeletions run the deletion settle over the reconcile residual. + * @param pageFallback page the filter when negentropy can't reconcile a relay. + * @param idChunk ids per reconcile chunk and per by-id fetch. + * @param downloadWorkers concurrent by-id download fetches per group. + * @param reconcileConcurrency overlapped `created_at`-window reconciles after an over-cap split. + * @param maxDeletionRounds hard cap on deletion-settle rounds (converges in 1–2). + * @param relayConcurrency write relays reconciled at once. + * @param authorChunk authors per reconcile filter (a relay with more is split into several). + * @param minAuthors skip relays hosting fewer than this many of the store's authors. + * @param idleTimeoutMs idle watchdog for reconciles / fetches / pages. + * @param publishTimeoutSecs OK-confirmation wait per uploaded event. + */ + class Config( + val kinds: List = DEFAULT_KINDS, + val down: Boolean = true, + val up: Boolean = true, + val syncDeletions: Boolean = true, + val pageFallback: Boolean = true, + val idChunk: Int = 500, + val downloadWorkers: Int = 4, + val reconcileConcurrency: Int = 2, + val maxDeletionRounds: Int = 4, + val relayConcurrency: Int = 4, + val authorChunk: Int = 500, + val minAuthors: Int = 1, + val idleTimeoutMs: Long = 30_000L, + val publishTimeoutSecs: Long = 15, + ) + + /** Per-write-relay outcome of an [update]. */ + class RelayResult( + val relay: NormalizedRelayUrl, + val authors: Int, + val need: Int, + val have: Int, + val downloaded: Int, + val uploaded: Int, + val deletionsSentUp: Int, + val deletionsAppliedDown: Int, + val pagedFallback: Boolean, + /** null on success; the negentropy (and any page-fallback) failure otherwise. */ + val error: String?, + ) + + /** Aggregate outcome of an [update], plus the per-relay breakdown. */ + class Result( + val relayListsInStore: Int, + val authorsWithOutbox: Int, + val relays: Int, + val relaysOk: Int, + val relaysFailed: Int, + val relaysPagedFallback: Int, + val downloaded: Int, + val uploaded: Int, + val deletionsSentUp: Int, + val deletionsAppliedDown: Int, + val perRelay: List, + ) + + /** + * Group the store's authors by their kind:10002 write relays (the outbox model). + * The latest kind:10002 per author wins; an author with no write-marked relays + * contributes nothing (there is nowhere to reconcile them). Public so callers can + * inspect the plan (relay count, largest groups) before running [update]. + */ + suspend fun writeRelayGroups(): Map> = groupByWriteRelay(loadLatestRelayLists()) + + /** Latest kind:10002 per author from the store (replaceable — newest createdAt wins). */ + private suspend fun loadLatestRelayLists(): Map { + val latest = HashMap() + for (event in store.query(Filter(kinds = listOf(AdvertisedRelayListEvent.KIND)))) { + if (event !is AdvertisedRelayListEvent) continue + val prev = latest[event.pubKey] + if (prev == null || event.createdAt > prev.createdAt) latest[event.pubKey] = event + } + return latest + } + + /** Invert the per-author relay lists into `write-relay -> authors`. */ + private fun groupByWriteRelay(latest: Map): Map> { + val relayToAuthors = HashMap>() + for ((author, list) in latest) { + val writes = list.writeRelaysNorm() ?: continue + for (relay in writes) relayToAuthors.getOrPut(relay) { HashSet() }.add(author) + } + return relayToAuthors + } + + /** + * Run the full outbox-model refresh: [writeRelayGroups] then one per-relay sync + * for each group hosting at least [Config.minAuthors] authors, up to + * [Config.relayConcurrency] relays at once (largest groups first). Best-effort — + * a relay that fails is recorded in its [RelayResult.error] and never aborts the run. + */ + suspend fun update(): Result { + val latest = loadLatestRelayLists() + val groups = groupByWriteRelay(latest) + val authorsWithOutbox = groups.values.flatMapTo(HashSet()) { it }.size + + val plan = + groups.entries + .filter { it.value.size >= config.minAuthors } + .sortedByDescending { it.value.size } + + val perRelay = + if (plan.isEmpty()) { + emptyList() + } else { + val gate = Semaphore(config.relayConcurrency.coerceAtLeast(1)) + coroutineScope { + plan + .map { (relay, authors) -> + async { gate.withPermit { syncRelay(relay, authors) } } + }.awaitAll() + } + } + + return Result( + relayListsInStore = latest.size, + authorsWithOutbox = authorsWithOutbox, + relays = plan.size, + relaysOk = perRelay.count { it.error == null }, + relaysFailed = perRelay.count { it.error != null }, + relaysPagedFallback = perRelay.count { it.pagedFallback }, + downloaded = perRelay.sumOf { it.downloaded }, + uploaded = perRelay.sumOf { it.uploaded }, + deletionsSentUp = perRelay.sumOf { it.deletionsSentUp }, + deletionsAppliedDown = perRelay.sumOf { it.deletionsAppliedDown }, + perRelay = perRelay, + ) + } + + /** Sync one write relay by folding each [Config.authorChunk]-sized author slice. */ + private suspend fun syncRelay( + relay: NormalizedRelayUrl, + authors: Set, + ): RelayResult { + var downloaded = 0 + var uploaded = 0 + var delUp = 0 + var delDown = 0 + var need = 0 + var have = 0 + var paged = false + var error: String? = null + + for (chunk in authors.toList().chunked(config.authorChunk.coerceAtLeast(1))) { + val filter = Filter(kinds = config.kinds, authors = chunk) + val res = syncGroup(relay, filter) + downloaded += res.downloaded + uploaded += res.uploaded + delUp += res.deletionsSentUp + delDown += res.deletionsAppliedDown + need += res.need + have += res.have + if (res.pagedFallback) paged = true + if (res.error != null) error = res.error + } + + log( + "[graperank update] ${relay.url}: ${authors.size} authors, " + + "down $downloaded, up $uploaded, del↑ $delUp, del↓ $delDown" + + (if (paged) " (paged fallback)" else "") + + (if (error != null) " (error: $error)" else ""), + ) + return RelayResult(relay, authors.size, need, have, downloaded, uploaded, delUp, delDown, paged, error) + } + + /** One relay + one author chunk. Mirrors the two-pass content+deletion sync. */ + private suspend fun syncGroup( + relay: NormalizedRelayUrl, + filter: Filter, + ): GroupResult { + val localEvents = store.query(filter) + val localById = localEvents.associateBy { it.id } + val localEntries = localEvents.map { IdAndTime(it.createdAt, it.id) } + + val downloaded = AtomicInt(0) + val uploaded = AtomicInt(0) + + val reconcileResult = + try { + coroutineScope { + // needIds = relay has, store lacks; haveIds = store has, relay lacks. + val needBatches = Channel>(config.downloadWorkers * 2) + val haveBatches = Channel>(Channel.UNLIMITED) + + val downloaders = + List(config.downloadWorkers.coerceAtLeast(1)) { + launch { + for (batch in needBatches) { + for (event in client.fetchAll(relay, Filter(ids = batch), config.idleTimeoutMs)) { + if (store.verifyAndInsert(event)) downloaded.addAndFetch(1) + } + } + } + } + val uploader = + launch { + for (batch in haveBatches) { + for (id in batch) { + val ev = localById[id] ?: continue + if (client.publishAndConfirm(ev, setOf(relay), config.publishTimeoutSecs)) uploaded.addAndFetch(1) + } + } + } + + val result = + try { + client.negentropyReconcile( + relay = relay, + filter = filter, + localEntries = localEntries, + batchSize = config.idChunk, + idleTimeoutMs = config.idleTimeoutMs, + reconcileConcurrency = config.reconcileConcurrency, + onHaveIds = if (config.up) { batch -> haveBatches.send(batch) } else null, + onNeedIds = { batch -> if (config.down) needBatches.send(batch) }, + ) + } finally { + needBatches.close() + haveBatches.close() + } + + downloaders.joinAll() + uploader.join() + result + } + } catch (e: NegentropySyncException) { + // Negentropy couldn't reconcile — page the same authors+kinds so the + // records still refresh. Deletion settle is negentropy-only, so skipped. + var pageError: String? = e.message ?: "negentropy sync failed" + if (config.pageFallback && config.down) { + pageError = + try { + downloaded.addAndFetch(pageDownload(relay, filter)) + null + } catch (pe: Exception) { + "negentropy: ${e.message}; page fallback: ${pe::class.simpleName}: ${pe.message}" + } + } + return GroupResult(downloaded.load(), uploaded.load(), 0, 0, 0, 0, pagedFallback = true, error = pageError) + } + + val deletions = + if (config.syncDeletions) { + client.negentropySettleDeletions( + relay = relay, + filter = filter, + store = store, + sendUp = config.down, + applyDown = config.up, + batchSize = config.idChunk, + idleTimeoutMs = config.idleTimeoutMs, + maxRounds = config.maxDeletionRounds, + reconcileConcurrency = config.reconcileConcurrency, + ) + } else { + null + } + + return GroupResult( + downloaded = downloaded.load(), + uploaded = uploaded.load(), + deletionsSentUp = deletions?.sentUp ?: 0, + deletionsAppliedDown = deletions?.appliedDown ?: 0, + need = reconcileResult.needCount, + have = reconcileResult.haveCount, + pagedFallback = false, + error = null, + ) + } + + /** + * Paged fallback: walk [relay] past its per-REQ cap for [filter], verifying and + * inserting each event into [store]. [fetchAllPages]'s `onEvent` can't suspend, so + * events funnel through a bounded channel to a single inserter. Returns how many + * were newly stored. + */ + private suspend fun pageDownload( + relay: NormalizedRelayUrl, + filter: Filter, + ): Int { + val stored = AtomicInt(0) + val events = Channel(Channel.UNLIMITED) + coroutineScope { + val inserter = + launch { + for (event in events) { + if (store.verifyAndInsert(event)) stored.addAndFetch(1) + } + } + try { + client.fetchAllPages(relay, listOf(filter), config.idleTimeoutMs) { event -> events.trySend(event) } + } finally { + events.close() + } + inserter.join() + } + return stored.load() + } + + private class GroupResult( + val downloaded: Int, + val uploaded: Int, + val deletionsSentUp: Int, + val deletionsAppliedDown: Int, + val need: Int, + val have: Int, + val pagedFallback: Boolean, + val error: String?, + ) + + companion object { + /** The record kinds a GrapeRank score is a function of. */ + val DEFAULT_KINDS = + listOf( + MetadataEvent.KIND, // 0 — profiles + ContactListEvent.KIND, // 3 — follows + AdvertisedRelayListEvent.KIND, // 10002 — outbox relay lists + ReportEvent.KIND, // 1984 — reports + ) + } +}