From bb74b16efc71ae2874d2c031fa59b9181bb60a03 Mon Sep 17 00:00:00 2001 From: Niels Mokkenstorm Date: Thu, 16 Jul 2026 22:27:38 +0200 Subject: [PATCH] fix(server): exclude gone adoption edges from counts, not from storage --- .../src/at_record_server/catalog_index.gleam | 19 +++++- .../catalog_index_postgres.gleam | 15 +++-- server/test/catalog_index_postgres_test.gleam | 33 ++++++++++ server/test/catalog_index_test.gleam | 64 +++++++++++++++++++ 4 files changed, 124 insertions(+), 7 deletions(-) diff --git a/server/src/at_record_server/catalog_index.gleam b/server/src/at_record_server/catalog_index.gleam index e6796c6..1421a86 100644 --- a/server/src/at_record_server/catalog_index.gleam +++ b/server/src/at_record_server/catalog_index.gleam @@ -29,6 +29,20 @@ pub type Adoption { ) } +/// `status` carries the raw shelf.entry `action` (see `jetstream_consumer` +/// and `at_record_server`'s backfill, both of which store `entry.action` +/// verbatim), not the folded `crate.Status` vocabulary. These are the two +/// actions `crate.status_of` maps to `Gone`; every other action (including +/// non-status-changing ones like "regraded") counts as adopted. An edge that +/// goes gone stays stored -- `adoption_count`/`adoption_counts` filter it +/// out, but a later re-upsert of the same `entry_uri` with an active status +/// (e.g. "acquired" again) counts again. +const gone_statuses = ["sold", "dropped"] + +fn is_active_status(status: String) -> Bool { + !list.contains(gone_statuses, status) +} + /// One catalog.edit event: a proposed change to some catalog entity, keyed by /// the edit record's own at-uri. pub type Edit { @@ -214,12 +228,15 @@ fn handle(state: State, msg: Msg) -> actor.Next(State, Msg) { fn count_for(state: State, release_uri: String) -> Int { dict.values(state.adoptions) - |> list.filter(fn(a) { a.release_uri == release_uri }) + |> list.filter(fn(a) { + a.release_uri == release_uri && is_active_status(a.status) + }) |> list.length } fn all_counts(state: State) -> Dict(String, Int) { dict.values(state.adoptions) + |> list.filter(fn(a) { is_active_status(a.status) }) |> list.fold(dict.new(), fn(acc, a) { dict.upsert(acc, a.release_uri, fn(existing) { case existing { diff --git a/server/src/at_record_server/catalog_index_postgres.gleam b/server/src/at_record_server/catalog_index_postgres.gleam index a3ca259..7fd7901 100644 --- a/server/src/at_record_server/catalog_index_postgres.gleam +++ b/server/src/at_record_server/catalog_index_postgres.gleam @@ -19,6 +19,11 @@ import wisp const cursor_id = "default" +/// Mirrors `catalog_index.gone_statuses`: shelf.entry actions that map to +/// `crate.Status`'s `Gone`. An edge with one of these stays stored (a status +/// can go back to active later) but is excluded from the adoption counts. +const active_status_filter = "status not in ('sold', 'dropped')" + pub fn table_store(conn: pog.Connection) -> Result(Store, String) { use _ <- result.try(migrate(conn)) Ok( @@ -266,9 +271,8 @@ fn adoption_count(conn: pog.Connection, release_uri: String) -> Int { decode.success(count) } case - pog.query( - "select count(*)::int as count from catalog_adoptions where release_uri = $1", - ) + pog.query("select count(*)::int as count from catalog_adoptions + where release_uri = $1 and " <> active_status_filter) |> pog.parameter(pog.text(release_uri)) |> pog.returning(row) |> pog.execute(conn) @@ -285,9 +289,8 @@ fn adoption_counts(conn: pog.Connection) -> Dict(String, Int) { decode.success(#(release_uri, count)) } case - pog.query( - "select release_uri, count(*)::int as count from catalog_adoptions group by release_uri", - ) + pog.query("select release_uri, count(*)::int as count from catalog_adoptions + where " <> active_status_filter <> " group by release_uri") |> pog.returning(row) |> pog.execute(conn) { diff --git a/server/test/catalog_index_postgres_test.gleam b/server/test/catalog_index_postgres_test.gleam index 207471e..f0dff09 100644 --- a/server/test/catalog_index_postgres_test.gleam +++ b/server/test/catalog_index_postgres_test.gleam @@ -90,6 +90,39 @@ pub fn adoption_and_edit_round_trip_test() { assert store.edits_for(release) == [] } +pub fn gone_edges_are_excluded_but_reacquiring_counts_again_test() { + use store <- with_store() + let release = "at://did:plc:pgtest/dev.mokkenstorm.crate.catalog.release/pg3" + let entry_uri = "at://did:plc:pgtest/dev.mokkenstorm.crate.shelf.entry/pg-e2" + store.upsert_adoption(Adoption( + entry_uri:, + did: "did:plc:pgtest", + release_uri: release, + status: "acquired", + created_at: "2024-01-01T00:00:00Z", + )) + assert store.adoption_count(release) == 1 + store.upsert_adoption(Adoption( + entry_uri:, + did: "did:plc:pgtest", + release_uri: release, + status: "sold", + created_at: "2024-02-01T00:00:00Z", + )) + // Gone, but still stored: the count excludes it without a delete. + assert store.adoption_count(release) == 0 + assert dict.get(store.adoption_counts(), release) == Error(Nil) + store.upsert_adoption(Adoption( + entry_uri:, + did: "did:plc:pgtest", + release_uri: release, + status: "acquired", + created_at: "2024-03-01T00:00:00Z", + )) + assert store.adoption_count(release) == 1 + store.delete_adoption(entry_uri) +} + pub fn cursor_round_trip_test() { use store <- with_store() store.save_cursor(1_700_000_000_123_456) diff --git a/server/test/catalog_index_test.gleam b/server/test/catalog_index_test.gleam index 1cc86cf..b2e37cc 100644 --- a/server/test/catalog_index_test.gleam +++ b/server/test/catalog_index_test.gleam @@ -84,6 +84,62 @@ pub fn adoption_upsert_by_entry_uri_overwrites_test() { status: "acquired", created_at: "2024-01-01T00:00:00Z", )) + store.upsert_adoption(Adoption( + entry_uri:, + did: "did:plc:a", + release_uri: release, + status: "wanted", + created_at: "2024-02-01T00:00:00Z", + )) + assert store.adoption_count(release) == 1 +} + +pub fn gone_edges_do_not_count_but_stay_stored_test() { + let assert Ok(store) = catalog_index.start() + let release = "at://did:plc:pub/dev.mokkenstorm.crate.catalog.release/r1" + let sold_entry = "at://did:plc:a/dev.mokkenstorm.crate.shelf.entry/e1" + let dropped_entry = "at://did:plc:b/dev.mokkenstorm.crate.shelf.entry/e2" + let owned_entry = "at://did:plc:c/dev.mokkenstorm.crate.shelf.entry/e3" + store.upsert_adoption(Adoption( + entry_uri: sold_entry, + did: "did:plc:a", + release_uri: release, + status: "sold", + created_at: "2024-01-01T00:00:00Z", + )) + store.upsert_adoption(Adoption( + entry_uri: dropped_entry, + did: "did:plc:b", + release_uri: release, + status: "dropped", + created_at: "2024-01-01T00:00:00Z", + )) + store.upsert_adoption(Adoption( + entry_uri: owned_entry, + did: "did:plc:c", + release_uri: release, + status: "acquired", + created_at: "2024-01-01T00:00:00Z", + )) + // Only the owned edge counts; the gone edges stay stored (no delete + // required to make them stop counting), so `adoption_counts` never even + // mentions a release whose only edges are all gone. + assert store.adoption_count(release) == 1 + assert store.adoption_counts() == dict.from_list([#(release, 1)]) +} + +pub fn gone_then_reacquired_entry_counts_again_test() { + let assert Ok(store) = catalog_index.start() + let entry_uri = "at://did:plc:a/dev.mokkenstorm.crate.shelf.entry/e1" + let release = "at://did:plc:pub/dev.mokkenstorm.crate.catalog.release/r1" + store.upsert_adoption(Adoption( + entry_uri:, + did: "did:plc:a", + release_uri: release, + status: "acquired", + created_at: "2024-01-01T00:00:00Z", + )) + assert store.adoption_count(release) == 1 store.upsert_adoption(Adoption( entry_uri:, did: "did:plc:a", @@ -91,6 +147,14 @@ pub fn adoption_upsert_by_entry_uri_overwrites_test() { status: "sold", created_at: "2024-02-01T00:00:00Z", )) + assert store.adoption_count(release) == 0 + store.upsert_adoption(Adoption( + entry_uri:, + did: "did:plc:a", + release_uri: release, + status: "acquired", + created_at: "2024-03-01T00:00:00Z", + )) assert store.adoption_count(release) == 1 } -- 2.51.2