refactor(graperank): make Step 0 an actual BFS (LocalGraphBFS)

Rename Sync2Bootstrap -> LocalGraphBFS / BootstrapState -> LocalGraph, and
replace the scan-then-walk (which loaded every kind:3 and kind:10002 in the
store into memory before using the few thousand reachable from the observer)
with a real level-synchronized BFS: it queries only the current frontier, one
batched author+kind lookup per level, so memory and query cost scale with the
reachable graph, not the whole store. kind:10002 watermarks + outbox are a
final batched pass over the discovered set.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0194n8XyaEhqJFDNiQBe6Bqa

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/experimental/graperank/sync2/LocalGraphBFS.kt

@@ -0,0 +1,179 @@
+/*
+ * 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.sync2
+
+import com.vitorpamplona.quartz.nip01Core.core.Event
+import com.vitorpamplona.quartz.nip01Core.core.HexKey
+import com.vitorpamplona.quartz.nip01Core.core.Kind
+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.nip02FollowList.ContactListEvent
+import com.vitorpamplona.quartz.nip65RelayList.AdvertisedRelayListEvent
+
+/**
+ * The newest `created_at` we already hold, per crawl-kind, for a single author.
+ *
+ * This is the incremental floor for sync2: when the pipeline re-fetches this
+ * user we ask their relays only for events *newer* than what we already stored
+ * ([sinceFor]), so a warm re-sync collects almost nothing but empty EOSEs
+ * except where something actually changed — that is what makes it both fast and
+ * up-to-date at the same time.
+ */
+class Watermarks(
+    private val newestByKind: Map<Kind, Long>,
+) {
+    /** Newest stored `created_at` for [kind], or 0 if we hold no such event. */
+    fun newest(kind: Kind): Long = newestByKind[kind] ?: 0L
+
+    /**
+     * The `since` floor for an incremental re-fetch of [kind]: `newest + 1`, so
+     * a relay returns only strictly-newer events. Null when we hold nothing for
+     * the kind (a cold fetch — no floor). Safe as `+1` because both crawl kinds
+     * (3 and 10002) are replaceable, so only a strictly-newer version matters.
+     */
+    fun sinceFor(kind: Kind): Long? = newestByKind[kind]?.let { it + 1 }
+
+    /** Kinds we hold at least one stored event for. */
+    fun kinds(): Set<Kind> = newestByKind.keys
+
+    override fun toString(): String = "Watermarks($newestByKind)"
+}
+
+/**
+ * The observer's follow graph as it exists in the local store right now — the
+ * output of [LocalGraphBFS]. Seeds the sync2 pipeline so the crawl starts
+ * incremental instead of from scratch.
+ *
+ * @param observer the pubkey the graph is anchored at (always present in [hopOf] at hop 0).
+ * @param hopOf follow-graph distance from the observer for every user the BFS
+ *   reached over the *stored* kind:3 contact lists. Its key set is the frontier
+ *   we already know about; the crawl expands it.
+ * @param watermarks per-author freshness floor (see [Watermarks]) for the users
+ *   the BFS touched — kind:3 for anyone whose contact list we hold, kind:10002 for
+ *   anyone whose relay list we hold.
+ * @param knownOutbox each discovered user's kind:10002 write relays as last stored,
+ *   so the router can route their content fetch without first resolving their
+ *   outbox, and the ingest stage can diff a fresh 10002 against it for added relays.
+ */
+class LocalGraph(
+    val observer: HexKey,
+    val hopOf: Map<HexKey, Int>,
+    val watermarks: Map<HexKey, Watermarks>,
+    val knownOutbox: Map<HexKey, Set<NormalizedRelayUrl>>,
+) {
+    /** Number of users reachable from the observer in the stored graph. */
+    val discoveredUsers: Int get() = hopOf.size
+
+    /** Users bucketed by follow-graph distance (hop -> count), ascending. */
+    fun hopHistogram(): Map<Int, Int> =
+        hopOf.values
+            .groupingBy { it }
+            .eachCount()
+            .toList()
+            .sortedBy { it.first }
+            .toMap()
+
+    /** Freshness floor for [pubkey], or null if we hold nothing for them yet. */
+    fun watermarkFor(pubkey: HexKey): Watermarks? = watermarks[pubkey]
+
+    /** Last-known kind:10002 write relays for [pubkey] (empty if unknown). */
+    fun outboxOf(pubkey: HexKey): Set<NormalizedRelayUrl> = knownOutbox[pubkey] ?: emptySet()
+}
+
+/**
+ * Step 0 of the sync2 pipeline: a breadth-first search of the observer's follow
+ * graph over the **local store only** (no network), so the crawl can start
+ * incremental.
+ *
+ * It is a real, level-synchronized BFS — it only ever queries the frontier it has
+ * reached, one indexed author+kind lookup per level (chunked to
+ * [AUTHORS_PER_QUERY]). It never scans the whole store: a node's kind:3 is pulled
+ * on demand as the frontier expands, so memory and query cost are bounded by the
+ * size of the reachable graph (a few thousand users), not the store's millions of
+ * contact lists.
+ *
+ * As it walks, it records each user's freshness [Watermarks] (kind:3 from the BFS
+ * pass, kind:10002 from a final pass over the discovered set) and their last-known
+ * outbox. Everything downstream seeds off the returned [LocalGraph].
+ */
+class LocalGraphBFS(
+    private val store: IEventStore,
+) {
+    suspend fun traverse(observer: HexKey): LocalGraph {
+        val hopOf = HashMap<HexKey, Int>()
+        val watermarks = HashMap<HexKey, HashMap<Kind, Long>>()
+
+        // Level-synchronized BFS: process the whole current frontier, collect the
+        // next frontier, advance the hop. The observer is hop 0 (present even on an
+        // empty store). Each level is one batched query over just its authors —
+        // never a full-store scan. kind:3 is replaceable, so the store returns at
+        // most one list per author and its created_at is the kind:3 watermark.
+        hopOf[observer] = 0
+        var frontier: List<HexKey> = listOf(observer)
+        var hop = 0
+        while (frontier.isNotEmpty()) {
+            val nextHop = hop + 1
+            val next = LinkedHashSet<HexKey>()
+            for (chunk in frontier.chunked(AUTHORS_PER_QUERY)) {
+                val lists = store.query<Event>(Filter(authors = chunk, kinds = listOf(ContactListEvent.KIND)))
+                for (e in lists) {
+                    if (e !is ContactListEvent) continue
+                    watermarks.getOrPut(e.pubKey) { HashMap() }[ContactListEvent.KIND] = e.createdAt
+                    for (follow in e.verifiedFollowKeySet()) {
+                        if (follow !in hopOf) {
+                            hopOf[follow] = nextHop
+                            next.add(follow)
+                        }
+                    }
+                }
+            }
+            frontier = next.toList()
+            hop = nextHop
+        }
+
+        // kind:10002 for the discovered set only: watermark + last-known outbox,
+        // batched the same way. Replaceable, so one relay list per author.
+        val knownOutbox = HashMap<HexKey, Set<NormalizedRelayUrl>>()
+        for (chunk in hopOf.keys.chunked(AUTHORS_PER_QUERY)) {
+            val lists = store.query<Event>(Filter(authors = chunk, kinds = listOf(AdvertisedRelayListEvent.KIND)))
+            for (e in lists) {
+                if (e !is AdvertisedRelayListEvent) continue
+                watermarks.getOrPut(e.pubKey) { HashMap() }[AdvertisedRelayListEvent.KIND] = e.createdAt
+                e.writeRelaysNorm()?.let { knownOutbox[e.pubKey] = it.toHashSet() }
+            }
+        }
+
+        return LocalGraph(observer, hopOf, watermarks.mapValues { Watermarks(it.value) }, knownOutbox)
+    }
+
+    companion object {
+        /** Authors per store query — one batched lookup covers a whole BFS level. */
+        const val AUTHORS_PER_QUERY = 500
+
+        /** The two kinds sync2 crawls to build the graph: follows and relay lists. */
+        val CRAWL_KINDS: List<Kind> =
+            listOf(
+                ContactListEvent.KIND,
+                AdvertisedRelayListEvent.KIND,
+            )
+    }
+}

quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/experimental/graperank/sync2/LocalGraphBFSTest.kt

@@ -0,0 +1,197 @@
+/*
+ * 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.sync2
+
+import com.vitorpamplona.quartz.nip01Core.core.Event
+import com.vitorpamplona.quartz.nip01Core.core.HexKey
+import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
+import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
+import com.vitorpamplona.quartz.nip01Core.store.FtsReindexProgress
+import com.vitorpamplona.quartz.nip01Core.store.IEventStore
+import com.vitorpamplona.quartz.nip02FollowList.ContactListEvent
+import com.vitorpamplona.quartz.nip65RelayList.AdvertisedRelayListEvent
+import kotlinx.coroutines.test.runTest
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertNull
+import kotlin.test.assertTrue
+
+class LocalGraphBFSTest {
+    // A dummy signature — the BFS never verifies, it only reads tags + created_at.
+    private val sig = "0".repeat(128)
+
+    /** A distinct valid 64-char hex pubkey/id for a small integer, e.g. pk(1). */
+    private fun pk(n: Int): HexKey = n.toString(16).padStart(64, '0')
+
+    private fun contactList(
+        author: Int,
+        follows: List<Int>,
+        createdAt: Long,
+    ): ContactListEvent {
+        val tags = follows.map { arrayOf("p", pk(it)) }.toTypedArray()
+        return ContactListEvent(pk(9000 + author), pk(author), createdAt, tags, "", sig)
+    }
+
+    private fun relayList(
+        author: Int,
+        writeRelays: List<String>,
+        createdAt: Long,
+    ): AdvertisedRelayListEvent {
+        val tags = writeRelays.map { arrayOf("r", it, "write") }.toTypedArray()
+        return AdvertisedRelayListEvent(pk(8000 + author), pk(author), createdAt, tags, "", sig)
+    }
+
+    private suspend fun bootstrap(
+        observer: Int,
+        events: List<Event>,
+    ): LocalGraph = LocalGraphBFS(ListStore(events)).traverse(pk(observer))
+
+    @Test
+    fun emptyStoreYieldsObserverAtHopZero() =
+        runTest {
+            val state = bootstrap(observer = 1, events = emptyList())
+            assertEquals(mapOf(pk(1) to 0), state.hopOf)
+            assertEquals(1, state.discoveredUsers)
+            assertNull(state.watermarkFor(pk(1)))
+            assertTrue(state.knownOutbox.isEmpty())
+        }
+
+    @Test
+    fun bfsStampsTransitiveHopDistances() =
+        runTest {
+            // 1 -> {2,3}, 2 -> {4}, 4 -> {5}; 3 has no stored list (leaf at hop 1).
+            val state =
+                bootstrap(
+                    observer = 1,
+                    events =
+                        listOf(
+                            contactList(1, listOf(2, 3), createdAt = 100),
+                            contactList(2, listOf(4), createdAt = 100),
+                            contactList(4, listOf(5), createdAt = 100),
+                        ),
+                )
+            assertEquals(0, state.hopOf[pk(1)])
+            assertEquals(1, state.hopOf[pk(2)])
+            assertEquals(1, state.hopOf[pk(3)])
+            assertEquals(2, state.hopOf[pk(4)])
+            assertEquals(3, state.hopOf[pk(5)])
+            assertEquals(mapOf(0 to 1, 1 to 2, 2 to 1, 3 to 1), state.hopHistogram())
+        }
+
+    @Test
+    fun bfsTakesShortestPathWhenAUserIsReachableTwoWays() =
+        runTest {
+            // 1 -> {2,3}, 2 -> {3}. User 3 is a direct follow (hop 1) AND 2's follow
+            // (hop 2); BFS must keep the shorter hop 1.
+            val state =
+                bootstrap(
+                    observer = 1,
+                    events =
+                        listOf(
+                            contactList(1, listOf(2, 3), createdAt = 100),
+                            contactList(2, listOf(3), createdAt = 100),
+                        ),
+                )
+            assertEquals(1, state.hopOf[pk(3)])
+        }
+
+    @Test
+    fun contactListSeedsWatermarkAndSinceFloor() =
+        runTest {
+            // The store holds one (replaceable) kind:3 per author; its created_at is
+            // the watermark, and the since floor is created_at + 1.
+            val state =
+                bootstrap(
+                    observer = 1,
+                    events = listOf(contactList(1, listOf(2), createdAt = 200)),
+                )
+            assertEquals(200, state.watermarkFor(pk(1))?.newest(ContactListEvent.KIND))
+            assertEquals(201, state.watermarkFor(pk(1))?.sinceFor(ContactListEvent.KIND))
+        }
+
+    @Test
+    fun relayListSeedsWatermarkAndKnownOutbox() =
+        runTest {
+            val state =
+                bootstrap(
+                    observer = 1,
+                    events =
+                        listOf(
+                            contactList(1, listOf(2), createdAt = 100),
+                            relayList(2, listOf("wss://alice.example"), createdAt = 150),
+                        ),
+                )
+            assertEquals(150, state.watermarkFor(pk(2))?.newest(AdvertisedRelayListEvent.KIND))
+            val outbox = state.outboxOf(pk(2)).map { it.url }
+            assertEquals(1, outbox.size)
+            assertTrue(outbox.first().contains("alice.example"), "got $outbox")
+        }
+
+    /**
+     * A read-only [IEventStore] over an in-memory list, matching via the real
+     * [Filter.match] so kind/author/since semantics are exercised for real. Only
+     * the query paths bootstrap uses are implemented; the write side errors.
+     */
+    private class ListStore(
+        private val events: List<Event>,
+    ) : IEventStore {
+        override val relay: NormalizedRelayUrl? = null
+
+        @Suppress("UNCHECKED_CAST")
+        override suspend fun <T : Event> query(filter: Filter): List<T> = events.filter(filter::match) as List<T>
+
+        @Suppress("UNCHECKED_CAST")
+        override suspend fun <T : Event> query(filters: List<Filter>): List<T> = events.filter { e -> filters.any { it.match(e) } } as List<T>
+
+        override suspend fun <T : Event> query(
+            filter: Filter,
+            onEach: (T) -> Unit,
+        ) = query<T>(filter).forEach(onEach)
+
+        override suspend fun <T : Event> query(
+            filters: List<Filter>,
+            onEach: (T) -> Unit,
+        ) = query<T>(filters).forEach(onEach)
+
+        override suspend fun count(filter: Filter): Int = query<Event>(filter).size
+
+        override suspend fun count(filters: List<Filter>): Int = query<Event>(filters).size
+
+        override suspend fun insert(event: Event) = error("read-only store")
+
+        override suspend fun transaction(body: IEventStore.ITransaction.() -> Unit) = error("read-only store")
+
+        override suspend fun delete(filter: Filter) = error("read-only store")
+
+        override suspend fun delete(filters: List<Filter>) = error("read-only store")
+
+        override suspend fun deleteExpiredEvents() = Unit
+
+        override suspend fun reindexFullTextSearch() = Unit
+
+        override suspend fun reindexFullTextSearch(
+            resumeFrom: String?,
+            batchSize: Int,
+        ): FtsReindexProgress = error("not supported")
+
+        override fun close() = Unit
+    }
+}