From 5d58054aa75dbb25662eef2ccc9808b4ced07ccb Mon Sep 17 00:00:00 2001
From: Niels Mokkenstorm <niels@mokkenstorm.dev>
Date: Mon, 20 Jul 2026 21:09:52 +0200
Subject: [PATCH] fix(server): page catalog backfills to exhaustion and fix
 login parity

---
 server/src/at_record_server.gleam             | 108 +-------
 server/src/at_record_server/browse.gleam      |   7 +-
 .../src/at_record_server/handlers/oauth.gleam |  27 +-
 .../src/at_record_server/user_backfill.gleam  | 213 ++++++++++++++++
 server/test/user_backfill_test.gleam          | 230 ++++++++++++++++++
 5 files changed, 466 insertions(+), 119 deletions(-)
 create mode 100644 server/src/at_record_server/user_backfill.gleam
 create mode 100644 server/test/user_backfill_test.gleam

diff --git a/server/src/at_record_server.gleam b/server/src/at_record_server.gleam
index c04328b..d254429 100644
--- a/server/src/at_record_server.gleam
+++ b/server/src/at_record_server.gleam
@@ -1,31 +1,22 @@
-import at_record/gen/catalog/edit as catalog_edit
-import at_record/gen/client as generated_client
-import at_record/gen/shelf/entry as shelf_entry
 import at_record_server/atproto_client
-import at_record_server/browse
 import at_record_server/catalog_index
 import at_record_server/config_env
 import at_record_server/jetstream_consumer
-import at_record_server/known_users.{type KnownUser, type Store as KnownUsers}
+import at_record_server/known_users.{type Store as KnownUsers}
 import at_record_server/oauth/config
 import at_record_server/oauth/keys
 import at_record_server/oauth/store
 import at_record_server/router
+import at_record_server/user_backfill
 import at_record_server/wiring
 import atproto/xrpc.{type Client}
-import gleam/dynamic/decode
 import gleam/erlang/process
 import gleam/int
 import gleam/list
-import gleam/option.{None, Some}
 import mist
 import wisp
 import wisp/wisp_mist
 
-// The per-user cap on the backfill fan-out; matches `browse`'s own cap so a
-// full backfill is never a heavier hit than a single browse load per user.
-const per_user_limit = 50
-
 pub fn main() -> Nil {
   wisp.configure_logger()
   let port = config_env.port()
@@ -83,105 +74,20 @@ pub fn main() -> Nil {
   process.sleep_forever()
 }
 
-/// One-shot seed for `catalog_index`: the same per-user `listRecords`
-/// fan-out `/browse` runs live on every request, paid once at boot instead,
-/// plus the adoption/edit edges each known user's shelf.entry and
-/// catalog.edit records carry. `jetstream_consumer` keeps the index live
-/// from here.
+/// One-shot seed for `catalog_index`: the same per-user, paged-to-exhaustion
+/// backfill the first-login path runs (`user_backfill.backfill_user`), paid
+/// once at boot for every already-known user instead of waiting on their
+/// next login. `jetstream_consumer` keeps the index live from here.
 fn backfill_catalog_index(
   client: Client,
   known_users: KnownUsers,
   index: catalog_index.Store,
 ) -> Nil {
   let users = known_users.list()
-  users
-  |> list.each(fn(user) {
-    browse.fetch_user_releases(client, user)
-    |> list.each(index.releases.upsert)
-    backfill_adoptions(client, user, index)
-    backfill_edits(client, user, index)
-  })
+  users |> list.each(user_backfill.backfill_user(client, _, index))
   wisp.log_info(
     "catalog_index: backfilled from "
     <> int.to_string(list.length(users))
     <> " known users",
   )
 }
-
-fn backfill_adoptions(
-  client: Client,
-  user: KnownUser,
-  index: catalog_index.Store,
-) -> Nil {
-  list_records(client, user, shelf_entry.collection)
-  |> list.each(fn(rec) {
-    let #(uri, value) = rec
-    case decode.run(value, shelf_entry.shelf_entry_decoder()) {
-      Ok(entry) ->
-        case entry.release {
-          Some(ref) ->
-            index.adoptions.upsert(catalog_index.Adoption(
-              entry_uri: uri,
-              did: user.did,
-              release_uri: ref.uri,
-              status: entry.action,
-              created_at: entry.created_at,
-              source: catalog_index.source_label(
-                option.then(entry.source, fn(s) { s.origin }),
-              ),
-            ))
-          None -> Nil
-        }
-      Error(_) -> Nil
-    }
-  })
-}
-
-fn backfill_edits(
-  client: Client,
-  user: KnownUser,
-  index: catalog_index.Store,
-) -> Nil {
-  list_records(client, user, catalog_edit.collection)
-  |> list.each(fn(rec) {
-    let #(uri, value) = rec
-    case decode.run(value, catalog_edit.catalog_edit_decoder()) {
-      Ok(edit) ->
-        case edit.subject {
-          Some(ref) ->
-            index.edits.upsert(catalog_index.Edit(
-              edit_uri: uri,
-              did: user.did,
-              subject_uri: ref.uri,
-              entity: edit.entity,
-              created_at: edit.created_at,
-            ))
-          None -> Nil
-        }
-      Error(_) -> Nil
-    }
-  })
-}
-
-/// A single page of one user's records in one collection, undecoded: shared
-/// by the adoption and edit backfills so there is exactly one place that
-/// builds the listRecords query. Any failure yields an empty list; this is
-/// a best-effort seed, not a requirement for boot to succeed.
-fn list_records(
-  client: Client,
-  user: KnownUser,
-  collection: String,
-) -> List(#(String, decode.Dynamic)) {
-  let params =
-    generated_client.RepoListRecordsParams(
-      collection:,
-      cursor: None,
-      limit: Some(per_user_limit),
-      repo: user.did,
-      reverse: None,
-    )
-  case generated_client.repo_list_records(client, user.pds, params, None) {
-    Ok(output) -> output.records |> list.map(fn(r) { #(r.uri, r.value) })
-    Error(_) -> []
-  }
-}
diff --git a/server/src/at_record_server/browse.gleam b/server/src/at_record_server/browse.gleam
index 331675d..b7da2fb 100644
--- a/server/src/at_record_server/browse.gleam
+++ b/server/src/at_record_server/browse.gleam
@@ -300,7 +300,12 @@ fn to_network_match(
   )
 }
 
-fn to_browse_row(
+/// A decoded `catalog.release` plus its publisher, shaped as the `BrowseRow`
+/// every catalog-index writer needs. Exported for `user_backfill`, which
+/// pages releases to exhaustion rather than through `fetch_user_releases`'s
+/// single-page fan-out: this is the one place that shape is built, so the
+/// backfill reuses it instead of duplicating it.
+pub fn to_browse_row(
   user: KnownUser,
   uri: String,
   cid: String,
diff --git a/server/src/at_record_server/handlers/oauth.gleam b/server/src/at_record_server/handlers/oauth.gleam
index 2f19e32..4281d25 100644
--- a/server/src/at_record_server/handlers/oauth.gleam
+++ b/server/src/at_record_server/handlers/oauth.gleam
@@ -2,7 +2,6 @@
 //// logout. The hard parts (PAR/PKCE/DPoP/token exchange) live under oauth/;
 //// these are just the HTTP-facing glue.
 
-import at_record_server/browse
 import at_record_server/context.{type Context, error_json}
 import at_record_server/known_users
 import at_record_server/oauth/flow
@@ -10,7 +9,7 @@ import at_record_server/oauth/session_store
 import at_record_server/oauth/sessions
 import at_record_server/oauth/store
 import at_record_server/oauth/tokens
-import gleam/int
+import at_record_server/user_backfill
 import gleam/json
 import gleam/list
 import gleam/option.{None, Some}
@@ -180,13 +179,14 @@ fn exchange_and_start_session(
   }
 }
 
-/// Seeds this one user's existing releases into `catalog_index` on login: the
-/// one-time boot backfill (`backfill_catalog_index` in `at_record_server`)
-/// only ever covers users already known at boot, so a first-time login is the
-/// only chance to backfill a user's pre-existing repo contents (Jetstream only
-/// surfaces commits from the point it started tailing, not history). Reuses
-/// `browse.fetch_user_releases`, which is already best-effort (empty list on
-/// any fetch/decode failure), so this can never fail the login.
+/// Seeds this one user's existing releases, adoptions, and edits into
+/// `catalog_index` on login via `user_backfill.backfill_user`: the one-time
+/// boot backfill (`backfill_catalog_index` in `at_record_server`) only ever
+/// covers users already known at boot, so a first-time login is the only
+/// chance to backfill a user's pre-existing repo contents (Jetstream only
+/// surfaces commits from the point it started tailing, not history).
+/// `backfill_user` is already best-effort (paging failures degrade to a
+/// warning log), so this can never fail the login.
 fn backfill_user_catalog(ctx: Context, session: sessions.OauthSession) -> Nil {
   let user =
     known_users.KnownUser(
@@ -194,12 +194,5 @@ fn backfill_user_catalog(ctx: Context, session: sessions.OauthSession) -> Nil {
       handle: session.handle,
       pds: session.pds,
     )
-  let rows = browse.fetch_user_releases(ctx.atproto.client, user)
-  rows |> list.each(ctx.catalog_index.releases.upsert)
-  wisp.log_info(
-    "catalog_index: backfilled "
-    <> int.to_string(list.length(rows))
-    <> " releases for "
-    <> session.did,
-  )
+  user_backfill.backfill_user(ctx.atproto.client, user, ctx.catalog_index)
 }
diff --git a/server/src/at_record_server/user_backfill.gleam b/server/src/at_record_server/user_backfill.gleam
new file mode 100644
index 0000000..b9276a3
--- /dev/null
+++ b/server/src/at_record_server/user_backfill.gleam
@@ -0,0 +1,213 @@
+//// Shared per-user backfill: pages one user's records in a collection to
+//// exhaustion (or to `max_pages`), decoding each row with a caller-supplied
+//// decoder, mirroring `shelf_owner.fetch_pages`'s cursor-following. Used by
+//// both the boot backfill (`at_record_server`, every known user) and the
+//// first-login backfill (`handlers/oauth`, one new user) so the two seed
+//// catalog.release, shelf.entry, and catalog.edit records identically --
+//// closing the gap where login only ever seeded releases, and the one
+//// where either backfill silently truncated at a single page.
+
+import at_record/gen/catalog/edit as catalog_edit
+import at_record/gen/catalog/release as catalog_release
+import at_record/gen/client as generated_client
+import at_record/gen/repo/list_records.{type RecordEntry}
+import at_record/gen/shelf/entry as shelf_entry
+import at_record_server/browse
+import at_record_server/catalog_index
+import at_record_server/known_users.{type KnownUser}
+import atproto/xrpc.{type Client}
+import gleam/dynamic/decode
+import gleam/int
+import gleam/list
+import gleam/option.{type Option, None, Some}
+import gleam/result
+import wisp
+
+/// The `com.atproto.repo.listRecords` page size, matching `shelf_owner`'s.
+const page_size = 100
+
+/// Hard cap on pages paged per user per collection (`max_pages * page_size`
+/// = 5000 records). Backfill is fire-and-forget, so hitting it degrades to
+/// a warning log rather than an error: whatever paged before the cap is
+/// still seeded.
+const max_pages = 50
+
+/// Page-to-exhaustion, best-effort seed of one user's `catalog.release`,
+/// `shelf.entry`, and `catalog.edit` records into `index`. The single place
+/// that decides what "backfilled" means for a user, so the boot and
+/// first-login call sites can never again drift out of parity.
+pub fn backfill_user(
+  client: Client,
+  user: KnownUser,
+  index: catalog_index.Store,
+) -> Nil {
+  let releases = backfill_releases(client, user, index)
+  let adoptions = backfill_adoptions(client, user, index)
+  let edits = backfill_edits(client, user, index)
+  wisp.log_info(
+    "user_backfill: seeded "
+    <> int.to_string(releases)
+    <> " release(s), "
+    <> int.to_string(adoptions)
+    <> " adoption(s), "
+    <> int.to_string(edits)
+    <> " edit(s) for "
+    <> user.did,
+  )
+}
+
+fn backfill_releases(
+  client: Client,
+  user: KnownUser,
+  index: catalog_index.Store,
+) -> Int {
+  let rows =
+    fetch_all(client, user, catalog_release.collection, fn(record) {
+      decode.run(record.value, catalog_release.catalog_release_decoder())
+      |> result.map(fn(value) {
+        browse.to_browse_row(user, record.uri, record.cid, value)
+      })
+      |> result.replace_error(Nil)
+    })
+  rows |> list.each(index.releases.upsert)
+  list.length(rows)
+}
+
+fn backfill_adoptions(
+  client: Client,
+  user: KnownUser,
+  index: catalog_index.Store,
+) -> Int {
+  let rows =
+    fetch_all(client, user, shelf_entry.collection, fn(record) {
+      use entry <- result.try(
+        decode.run(record.value, shelf_entry.shelf_entry_decoder())
+        |> result.replace_error(Nil),
+      )
+      use ref <- result.try(entry.release |> option.to_result(Nil))
+      Ok(catalog_index.Adoption(
+        entry_uri: record.uri,
+        did: user.did,
+        release_uri: ref.uri,
+        status: entry.action,
+        created_at: entry.created_at,
+        source: catalog_index.source_label(
+          option.then(entry.source, fn(s) { s.origin }),
+        ),
+      ))
+    })
+  rows |> list.each(index.adoptions.upsert)
+  list.length(rows)
+}
+
+fn backfill_edits(
+  client: Client,
+  user: KnownUser,
+  index: catalog_index.Store,
+) -> Int {
+  let rows =
+    fetch_all(client, user, catalog_edit.collection, fn(record) {
+      use edit <- result.try(
+        decode.run(record.value, catalog_edit.catalog_edit_decoder())
+        |> result.replace_error(Nil),
+      )
+      use ref <- result.try(edit.subject |> option.to_result(Nil))
+      Ok(catalog_index.Edit(
+        edit_uri: record.uri,
+        did: user.did,
+        subject_uri: ref.uri,
+        entity: edit.entity,
+        created_at: edit.created_at,
+      ))
+    })
+  rows |> list.each(index.edits.upsert)
+  list.length(rows)
+}
+
+/// One user's records in one collection, decoded with `decode_row`, paged
+/// to exhaustion or to `max_pages`. A page's transport failure or the cap
+/// itself keeps whatever paged so far and logs a warning: this backfill
+/// must never block boot or a login on a slow or misbehaving PDS.
+fn fetch_all(
+  client: Client,
+  user: KnownUser,
+  collection: String,
+  decode_row: fn(RecordEntry) -> Result(a, Nil),
+) -> List(a) {
+  fetch_pages(client, user, collection, decode_row, None, [], 0)
+}
+
+fn fetch_pages(
+  client: Client,
+  user: KnownUser,
+  collection: String,
+  decode_row: fn(RecordEntry) -> Result(a, Nil),
+  cursor: Option(String),
+  acc: List(a),
+  page: Int,
+) -> List(a) {
+  case page >= max_pages {
+    True -> {
+      wisp.log_warning(
+        "user_backfill: hit the "
+        <> int.to_string(max_pages)
+        <> "-page cap backfilling "
+        <> collection
+        <> " for "
+        <> user.did,
+      )
+      acc
+    }
+    False -> {
+      let params =
+        generated_client.RepoListRecordsParams(
+          collection:,
+          cursor:,
+          limit: Some(page_size),
+          repo: user.did,
+          reverse: None,
+        )
+      case generated_client.repo_list_records(client, user.pds, params, None) {
+        Error(err) -> {
+          wisp.log_warning(
+            "user_backfill: "
+            <> describe_error(err)
+            <> " backfilling "
+            <> collection
+            <> " for "
+            <> user.did,
+          )
+          acc
+        }
+        Ok(output) -> {
+          let rows = output.records |> list.filter_map(decode_row)
+          let all = list.append(acc, rows)
+          case output.cursor {
+            Some("") | None -> all
+            Some(next) ->
+              fetch_pages(
+                client,
+                user,
+                collection,
+                decode_row,
+                Some(next),
+                all,
+                page + 1,
+              )
+          }
+        }
+      }
+    }
+  }
+}
+
+fn describe_error(err: generated_client.RepoListRecordsError) -> String {
+  case err {
+    generated_client.RepoListRecordsTransport(e) ->
+      "transport failure: " <> xrpc.describe(e)
+    generated_client.RepoListRecordsUnexpected(status, message, _) ->
+      "unexpected pds status "
+      <> int.to_string(status)
+      <> { option.map(message, fn(m) { " " <> m }) |> option.unwrap("") }
+  }
+}
diff --git a/server/test/user_backfill_test.gleam b/server/test/user_backfill_test.gleam
new file mode 100644
index 0000000..9abbfa2
--- /dev/null
+++ b/server/test/user_backfill_test.gleam
@@ -0,0 +1,230 @@
+//// `user_backfill.backfill_user` tests against a host-branching stub xrpc
+//// client (see `actor_shelf_test.gleam`/`crate_overlap_test.gleam` for the
+//// pattern): cursor-chained multi-page release paging, the page-cap
+//// degrade-and-keep-partial behaviour, and the login-parity regression
+//// (releases, adoptions, and edits all seeded, not just releases).
+
+import at_record/gen/catalog/edit as catalog_edit
+import at_record/gen/catalog/release as catalog_release
+import at_record/gen/shelf/entry as shelf_entry
+import at_record_server/catalog_index
+import at_record_server/known_users.{KnownUser}
+import at_record_server/user_backfill
+import atproto/xrpc
+import gleam/bit_array
+import gleam/http/request.{type Request}
+import gleam/http/response
+import gleam/int
+import gleam/json
+import gleam/list
+import gleam/option.{None, Some}
+import gleam/result
+import gleam/string
+
+const pds_host = "pds.test"
+
+/// `[1, 2, .., count]`: `gleam/list` has no `range`, so build it the way
+/// `pagination_test.gleam` does.
+fn ints_up_to(count: Int) -> List(Int) {
+  list.repeat(Nil, count) |> list.index_map(fn(_, i) { i + 1 })
+}
+
+fn a_user() -> known_users.KnownUser {
+  KnownUser(did: "did:plc:me", handle: "me.test", pds: "https://" <> pds_host)
+}
+
+fn record(uri: String, cid: String, value: json.Json) -> json.Json {
+  json.object([
+    #("uri", json.string(uri)),
+    #("cid", json.string(cid)),
+    #("value", value),
+  ])
+}
+
+fn release_uri(rkey: String) -> String {
+  "at://did:plc:me/dev.mokkenstorm.crate.catalog.release/" <> rkey
+}
+
+fn release_record(rkey: String) -> json.Json {
+  record(
+    release_uri(rkey),
+    "bafy" <> rkey,
+    json.object([
+      #("title", json.string("Title " <> rkey)),
+      #("createdAt", json.string("2026-01-01T00:00:00Z")),
+    ]),
+  )
+}
+
+fn shelf_entry_record(rkey: String, release_rkey: String) -> json.Json {
+  record(
+    "at://did:plc:me/dev.mokkenstorm.crate.shelf.entry/" <> rkey,
+    "bafy" <> rkey,
+    json.object([
+      #("action", json.string("acquired")),
+      #("createdAt", json.string("2026-01-01T00:00:00Z")),
+      #(
+        "release",
+        json.object([
+          #("cid", json.string("bafyrel")),
+          #("uri", json.string(release_uri(release_rkey))),
+        ]),
+      ),
+    ]),
+  )
+}
+
+fn catalog_edit_record(rkey: String, subject_rkey: String) -> json.Json {
+  record(
+    "at://did:plc:me/dev.mokkenstorm.crate.catalog.edit/" <> rkey,
+    "bafy" <> rkey,
+    json.object([
+      #("createdAt", json.string("2026-01-01T00:00:00Z")),
+      #("entity", json.string("release")),
+      #("op", json.string("propose")),
+      #(
+        "subject",
+        json.object([
+          #("cid", json.string("bafyrel")),
+          #("uri", json.string(release_uri(subject_rkey))),
+        ]),
+      ),
+    ]),
+  )
+}
+
+fn page_body(
+  records: List(json.Json),
+  cursor: option.Option(String),
+) -> String {
+  json.object(
+    list.flatten([
+      [#("records", json.preprocessed_array(records))],
+      case cursor {
+        Some(c) -> [#("cursor", json.string(c))]
+        None -> []
+      },
+    ]),
+  )
+  |> json.to_string
+}
+
+fn empty_page_body() -> String {
+  page_body([], None)
+}
+
+fn collection_of(req: Request(BitArray)) -> String {
+  case req.query {
+    Some(q) ->
+      case string.contains(q, catalog_release.collection) {
+        True -> catalog_release.collection
+        False ->
+          case string.contains(q, shelf_entry.collection) {
+            True -> shelf_entry.collection
+            False -> catalog_edit.collection
+          }
+      }
+    None -> ""
+  }
+}
+
+fn cursor_of(req: Request(BitArray)) -> option.Option(String) {
+  use q <- option.then(req.query)
+  case string.split(q, "cursor=") {
+    [_, rest] -> rest |> string.split("&") |> list.first |> option.from_result
+    _ -> None
+  }
+}
+
+fn ok_body(
+  body: String,
+) -> Result(response.Response(BitArray), xrpc.TransportError) {
+  Ok(response.Response(200, [], bit_array.from_string(body)))
+}
+
+/// Three cursor-chained pages of `catalog.release` (40 + 40 + 15 = 95
+/// records, past the pre-fix 50-record single-page ceiling), and a single
+/// empty page for the other two collections.
+fn multi_page_release_client() -> xrpc.Client {
+  xrpc.Client(send: fn(req) {
+    case collection_of(req), cursor_of(req) {
+      col, None if col == catalog_release.collection ->
+        ok_body(page_body(
+          ints_up_to(40)
+            |> list.map(fn(n) { release_record("p1-" <> int.to_string(n)) }),
+          Some("p2"),
+        ))
+      col, Some("p2") if col == catalog_release.collection ->
+        ok_body(page_body(
+          ints_up_to(40)
+            |> list.map(fn(n) { release_record("p2-" <> int.to_string(n)) }),
+          Some("p3"),
+        ))
+      col, Some("p3") if col == catalog_release.collection ->
+        ok_body(page_body(
+          ints_up_to(15)
+            |> list.map(fn(n) { release_record("p3-" <> int.to_string(n)) }),
+          None,
+        ))
+      _, _ -> ok_body(empty_page_body())
+    }
+  })
+}
+
+pub fn backfill_user_pages_releases_to_exhaustion_test() {
+  let assert Ok(store) = catalog_index.start()
+  user_backfill.backfill_user(multi_page_release_client(), a_user(), store)
+  assert list.length(store.releases.list()) == 95
+}
+
+/// A client that never runs out of cursor: every response points at another
+/// page. Backfill must still terminate (at `max_pages`), landing exactly one
+/// distinct release per page paged before the cap, and no more.
+fn looping_release_client() -> xrpc.Client {
+  xrpc.Client(send: fn(req) {
+    case collection_of(req) {
+      col if col == catalog_release.collection -> {
+        let n = case cursor_of(req) {
+          Some(c) -> int.parse(c) |> result.unwrap(0)
+          None -> 0
+        }
+        ok_body(page_body(
+          [release_record("loop-" <> int.to_string(n))],
+          Some(int.to_string(n + 1)),
+        ))
+      }
+      _ -> ok_body(empty_page_body())
+    }
+  })
+}
+
+pub fn backfill_user_caps_pages_and_keeps_partial_results_test() {
+  let assert Ok(store) = catalog_index.start()
+  user_backfill.backfill_user(looping_release_client(), a_user(), store)
+  // 50 pages, one distinct release apiece, then the cap stops further paging.
+  assert list.length(store.releases.list()) == 50
+}
+
+/// One record per collection: the regression case for the login-backfill
+/// parity bug, where only releases ever landed in the index.
+fn one_of_each_client() -> xrpc.Client {
+  xrpc.Client(send: fn(req) {
+    case collection_of(req) {
+      col if col == catalog_release.collection ->
+        ok_body(page_body([release_record("r1")], None))
+      col if col == shelf_entry.collection ->
+        ok_body(page_body([shelf_entry_record("e1", "r1")], None))
+      col if col == catalog_edit.collection ->
+        ok_body(page_body([catalog_edit_record("d1", "r1")], None))
+      _ -> ok_body(empty_page_body())
+    }
+  })
+}
+
+pub fn backfill_user_seeds_adoptions_and_edits_not_just_releases_test() {
+  let assert Ok(store) = catalog_index.start()
+  user_backfill.backfill_user(one_of_each_client(), a_user(), store)
+  assert list.length(store.releases.list()) == 1
+  assert list.length(store.adoptions.list()) == 1
+  assert list.length(store.edits.for_subject(release_uri("r1"))) == 1
+}
-- 
2.51.2