From a39430f88431e824f29d1a84f57dae97375bd19b Mon Sep 17 00:00:00 2001 From: daviddao Date: Wed, 23 Sep 2026 09:26:05 +0200 Subject: [PATCH 1/2] fix: record history honors record and account deletion Deleting a record now removes its stored versions (instead of adding a tombstone), and PurgeActorData removes an account's versions together with its records, so deleted, deactivated or taken-down accounts leave no queryable past versions. The purge runs on every delete delivery and even for collections no longer tracked, so history recorded earlier is never stranded. Migration 016 indexes record_version(did) for account purges. --- .agents/skills/hyperindex/SKILL.md | 2 +- .../hyperindex/references/schema-reference.md | 8 +- .../add-record-version-history.yaml | 3 +- AGENTS.md | 2 +- README.md | 2 +- cmd/hyperindex/main.go | 3 + .../database/migrations/migrations_test.go | 15 ++- .../016_add_record_version_did_index.down.sql | 1 + .../016_add_record_version_did_index.up.sql | 2 + .../016_add_record_version_did_index.down.sql | 1 + .../016_add_record_version_did_index.up.sql | 2 + .../database/repositories/record_versions.go | 29 +++-- .../repositories/record_versions_test.go | 43 +++++-- internal/database/repositories/records.go | 7 ++ .../graphql/recordhistory/recordhistory.go | 10 +- internal/tap/handler.go | 40 ++----- internal/tap/handler_history_test.go | 109 ++++++++++++------ 17 files changed, 176 insertions(+), 103 deletions(-) create mode 100644 internal/database/migrations/postgres/016_add_record_version_did_index.down.sql create mode 100644 internal/database/migrations/postgres/016_add_record_version_did_index.up.sql create mode 100644 internal/database/migrations/sqlite/016_add_record_version_did_index.down.sql create mode 100644 internal/database/migrations/sqlite/016_add_record_version_did_index.up.sql diff --git a/.agents/skills/hyperindex/SKILL.md b/.agents/skills/hyperindex/SKILL.md index 6535781..f043abc 100644 --- a/.agents/skills/hyperindex/SKILL.md +++ b/.agents/skills/hyperindex/SKILL.md @@ -259,7 +259,7 @@ Use typed collection queries when the caller needs collection-specific filters, ## Record version history -For collections the deployment tracks, `recordHistory(uri: "at://…")` returns the observed versions of one record, oldest first, 100 per page by default (`first` up to 500; pass the last `id` as `after` for more). `value` holds the record body, or null for a delete. Use it to show how a record changed, for example who renamed an observation: +For collections the deployment tracks, `recordHistory(uri: "at://…")` returns the observed versions of one record, oldest first, 100 per page by default (`first` up to 500; pass the last `id` as `after` for more). `value` holds the record body. Deleted records (and records of deleted accounts) have no history. Use it to show how a record changed, for example who renamed an observation: ```graphql { diff --git a/.agents/skills/hyperindex/references/schema-reference.md b/.agents/skills/hyperindex/references/schema-reference.md index fe1005e..5d9ab23 100644 --- a/.agents/skills/hyperindex/references/schema-reference.md +++ b/.agents/skills/hyperindex/references/schema-reference.md @@ -349,21 +349,21 @@ External label predicates bound by the containing filter field. ## Record version history -`recordHistory(uri: String!, first: Int = 100, after: String)` returns `[RecordVersion!]!`, oldest first. `first` must be 1 to 500; for the next page pass the last version's `id` as `after`. Versions are recorded for collections the deployment tracks (`RECORD_HISTORY_COLLECTIONS`; production tracks `app.gainforest.dwc.occurrence`); records that were never tracked return an empty list, and history recorded before a collection was removed stays queryable. A `baseline` version is what was indexed when history was switched on; earlier edits are not available. +`recordHistory(uri: String!, first: Int = 100, after: String)` returns `[RecordVersion!]!`, oldest first. `first` must be 1 to 500; for the next page pass the last version's `id` as `after`. Versions are recorded for collections the deployment tracks (`RECORD_HISTORY_COLLECTIONS`; production tracks `app.gainforest.dwc.occurrence`); records that were never tracked return an empty list, and history recorded before a collection was removed stays queryable. A `baseline` version is what was indexed when history was switched on; earlier edits are not available. Deleting a record, or deleting, deactivating or taking down its account, removes its history. ### `RecordVersion` | Field | Type | Description | | --- | --- | --- | -| `action` | `String!` | `baseline`, `create`, `update`, or `delete`. | -| `cid` | `String!` | CID of this version (for deletes, the version that was deleted). | +| `action` | `String!` | `baseline`, `create`, or `update`. | +| `cid` | `String!` | CID of this version. | | `collection` | `String!` | Collection NSID. | | `did` | `String!` | Repository DID. | | `id` | `String!` | Monotonic version id; later versions have larger ids. | | `live` | `Boolean` | True when seen on the live stream, false for a resync delivery, null for baseline rows. | | `observedAt` | `String!` | When the indexer observed this version (RFC 3339). | | `uri` | `String!` | Record AT-URI. | -| `value` | `JSON` | The record body at this version; null for deletes. | +| `value` | `JSON` | The record body at this version. | ## External label support diff --git a/.changes/unreleased/add-record-version-history.yaml b/.changes/unreleased/add-record-version-history.yaml index fe23a57..1fa0383 100644 --- a/.changes/unreleased/add-record-version-history.yaml +++ b/.changes/unreleased/add-record-version-history.yaml @@ -1,5 +1,4 @@ kind: added -body: Add opt-in record version history. Collections listed in RECORD_HISTORY_COLLECTIONS (exact NSIDs or prefix.* patterns, Tap mode) keep every observed version and delete in a new record_version table, seeded on startup with a baseline row per existing record, and the new root recordHistory(uri, first, after) GraphQL query pages through them oldest first. History is off by default; migration 015 adds the table. -time: 2026-09-23T03:10:00.000000+02:00 +body: Add opt-in record version history. Collections listed in RECORD_HISTORY_COLLECTIONS (exact NSIDs or prefix.* patterns, Tap mode) keep every observed version in a new record_version table (removed again when the record or its account is deleted), seeded on startup with a baseline row per existing record, and the new root recordHistory(uri, first, after) GraphQL query pages through them oldest first. History is off by default; migrations 015 and 016 add the table and its indexes. custom: Affects: operator diff --git a/AGENTS.md b/AGENTS.md index e5c7188..e9e8b31 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -91,7 +91,7 @@ Run verification based on what changed. - `SECRET_KEY_BASE` must be at least 64 characters. - `TAP_ENABLED=true` switches record ingestion to Tap mode. After Tap becomes healthy, Hyperindex runs one bounded background pass that fills missing `actor.handle` values from Tap's local `/info/:did` metadata without delaying API startup. - `LABELER_SUBSCRIBE_ENABLED=true` with `LABELER_SUBSCRIBE_URLS` starts optional external `com.atproto.label.subscribeLabels` ingestion. -- `RECORD_HISTORY_COLLECTIONS` (Tap mode) keeps every version of matching records in `record_version`, exposed by the root `recordHistory(uri)` query. Startup seeds a `baseline` row for existing records that have no history yet; the version is written before the current-state upsert so Tap redelivery keeps both consistent. +- `RECORD_HISTORY_COLLECTIONS` (Tap mode) keeps every version of matching records in `record_version`, exposed by the root `recordHistory(uri)` query. Startup seeds a `baseline` row for existing records that have no history yet; the version is written before the current-state upsert so Tap redelivery keeps both consistent. Record deletes and account purges delete history too (always, even for collections no longer tracked). - Migrations run automatically on startup. - Backend container images bundle published helper Lexicons from `lexicons/`, pinned by `lexicons.json`, and set `LEXICON_DIR=/app/lexicons`. Database Lexicons override bundled documents with the same NSID. - Be careful with `ALLOWED_ORIGINS`: current code allows all origins when unset, even if older prose suggests stricter defaults. diff --git a/README.md b/README.md index 3d3b6fd..ff26df0 100644 --- a/README.md +++ b/README.md @@ -122,7 +122,7 @@ TAP_SIGNAL_COLLECTION=app.bsky.feed.post docker compose -f docker-compose.tap.ym | `TAP_COLLECTION_FILTERS` | Comma-separated collection NSIDs for Tap sidecar record filtering; set independently from legacy `JETSTREAM_COLLECTIONS` | *(empty)* | | `RECORD_HISTORY_COLLECTIONS` | Comma-separated collection NSIDs or `prefix.*` patterns whose versions are kept in `record_version` and served, paged, by the `recordHistory` GraphQL query (Tap mode only) | *(empty: no history)* | -**Record version history.** Records in collections listed in `RECORD_HISTORY_COLLECTIONS` keep an append-only history: one row per distinct version (CID) the indexer observes, plus delete tombstones. On startup, every existing record in a newly listed collection gets a `baseline` row holding its current version, so history starts from what was indexed when it was switched on. Earlier versions are not recoverable because repositories only keep the latest one. Query it with `recordHistory(uri: "at://…", first: 100, after: "")`, one page of up to 500 versions at a time, oldest first; pass the last version's `id` as `after` for the next page. Versions are told apart by CID (or a hash of the body when Tap sends none), so repeats are stored once. Removing a collection from the variable stops recording; history already stored stays queryable. +**Record version history.** Records in collections listed in `RECORD_HISTORY_COLLECTIONS` keep an append-only history: one row per distinct version (CID) the indexer observes. Deleting a record removes its history, and a deleted, deactivated or taken-down account loses all of its history, just as its current records are purged. On startup, every existing record in a newly listed collection gets a `baseline` row holding its current version, so history starts from what was indexed when it was switched on. Earlier versions are not recoverable because repositories only keep the latest one. Query it with `recordHistory(uri: "at://…", first: 100, after: "")`, one page of up to 500 versions at a time, oldest first; pass the last version's `id` as `after` for the next page. Versions are told apart by CID (or a hash of the body when Tap sends none), so repeats are stored once. Removing a collection from the variable stops recording; history already stored stays queryable. Tap Docker deployments also require `ADMIN_API_KEY` in `.env` because Hyperindex requires admin authentication at startup. `TAP_COLLECTION_FILTERS` is read by the Tap sidecar only; legacy `JETSTREAM_COLLECTIONS` remains part of Jetstream mode and is not used as a Tap filtering fallback. diff --git a/cmd/hyperindex/main.go b/cmd/hyperindex/main.go index e6956b3..b2fdaea 100644 --- a/cmd/hyperindex/main.go +++ b/cmd/hyperindex/main.go @@ -954,6 +954,9 @@ func startTap( // Create handler that stores records and publishes to subscriptions. handler := tap.NewIndexHandler(svc.records, svc.actors, svc.activity, pubsub, svc.validator) + // Deletes purge stored history even when no collection is tracked, so the + // version store is always wired; the matcher decides what gets recorded. + handler.WithRecordHistory(svc.recordVersions, repositories.CollectionMatcher{}) if historyCollections := repositories.NewCollectionMatcher(cfg.RecordHistoryCollections); !historyCollections.Empty() { // Seed the current version of every opted-in record before live events // start, so each history begins with what was indexed at switch-on. diff --git a/internal/database/migrations/migrations_test.go b/internal/database/migrations/migrations_test.go index d6f3614..07ba40a 100644 --- a/internal/database/migrations/migrations_test.go +++ b/internal/database/migrations/migrations_test.go @@ -298,7 +298,7 @@ func TestMigrations_RunPostgresRenamesIndexingActivity(t *testing.T) { } assertPostgresIndexNotExists(ctx, t, exec, schemaName, "idx_record_json_gin") - for _, version := range []string{"015", "014", "013", "012", "011"} { + for _, version := range []string{"016", "015", "014", "013", "012", "011"} { if err := migrations.Rollback(ctx, exec); err != nil { t.Fatalf("Rollback(%s) returned error: %v", version, err) } @@ -491,12 +491,14 @@ func TestMigrations_Rollback(t *testing.T) { } } -// rollbackRecordVersionMigration undoes 015 so tests written against 014 as the -// newest migration keep exercising 014 and 013. +// rollbackRecordVersionMigration undoes 016 and 015 so tests written against +// 014 as the newest migration keep exercising 014 and 013. func rollbackRecordVersionMigration(ctx context.Context, t *testing.T, exec *sqlite.Executor) { t.Helper() - if err := migrations.Rollback(ctx, exec); err != nil { - t.Fatalf("Rollback(015) error = %v", err) + for _, version := range []string{"016", "015"} { + if err := migrations.Rollback(ctx, exec); err != nil { + t.Fatalf("Rollback(%s) error = %v", version, err) + } } } @@ -509,6 +511,9 @@ func TestMigrations_RecordVersionUpAndDown(t *testing.T) { if !sqliteIndexExists(t, exec, "idx_record_version_uri_key") || !migrationVersionExists(t, exec, "015") { t.Fatal("migration 015 did not create record_version") } + if !sqliteIndexExists(t, exec, "idx_record_version_did") || !migrationVersionExists(t, exec, "016") { + t.Fatal("migration 016 did not create the record_version did index") + } rollbackRecordVersionMigration(ctx, t, exec) if sqliteIndexExists(t, exec, "idx_record_version_uri_key") || migrationVersionExists(t, exec, "015") { t.Fatal("migration 015 schema/version remained after rollback") diff --git a/internal/database/migrations/postgres/016_add_record_version_did_index.down.sql b/internal/database/migrations/postgres/016_add_record_version_did_index.down.sql new file mode 100644 index 0000000..684e9ae --- /dev/null +++ b/internal/database/migrations/postgres/016_add_record_version_did_index.down.sql @@ -0,0 +1 @@ +DROP INDEX IF EXISTS idx_record_version_did; diff --git a/internal/database/migrations/postgres/016_add_record_version_did_index.up.sql b/internal/database/migrations/postgres/016_add_record_version_did_index.up.sql new file mode 100644 index 0000000..a6bb1d3 --- /dev/null +++ b/internal/database/migrations/postgres/016_add_record_version_did_index.up.sql @@ -0,0 +1,2 @@ +-- Account purges delete an account's record history by DID. +CREATE INDEX IF NOT EXISTS idx_record_version_did ON record_version(did); diff --git a/internal/database/migrations/sqlite/016_add_record_version_did_index.down.sql b/internal/database/migrations/sqlite/016_add_record_version_did_index.down.sql new file mode 100644 index 0000000..684e9ae --- /dev/null +++ b/internal/database/migrations/sqlite/016_add_record_version_did_index.down.sql @@ -0,0 +1 @@ +DROP INDEX IF EXISTS idx_record_version_did; diff --git a/internal/database/migrations/sqlite/016_add_record_version_did_index.up.sql b/internal/database/migrations/sqlite/016_add_record_version_did_index.up.sql new file mode 100644 index 0000000..a6bb1d3 --- /dev/null +++ b/internal/database/migrations/sqlite/016_add_record_version_did_index.up.sql @@ -0,0 +1,2 @@ +-- Account purges delete an account's record history by DID. +CREATE INDEX IF NOT EXISTS idx_record_version_did ON record_version(did); diff --git a/internal/database/repositories/record_versions.go b/internal/database/repositories/record_versions.go index e209f8d..3803d95 100644 --- a/internal/database/repositories/record_versions.go +++ b/internal/database/repositories/record_versions.go @@ -19,13 +19,13 @@ const ( RecordVersionBaseline = "baseline" RecordVersionCreate = "create" RecordVersionUpdate = "update" - RecordVersionDelete = "delete" // MaxRecordHistoryPageSize bounds one recordHistory response. MaxRecordHistoryPageSize = 500 ) -// RecordVersion is one observed version of a record, or a delete tombstone. +// RecordVersion is one observed version of a record. Deleting a record (or +// its account) removes its history, so there are no delete entries. type RecordVersion struct { ID int64 URI string @@ -33,7 +33,7 @@ type RecordVersion struct { DID string Collection string Action string - // JSON is the record body; nil for delete tombstones. + // JSON is the record body. JSON *string // Live reports whether Tap delivered the event from the live stream // (false for resync/backfill deliveries); nil for baseline rows. @@ -127,12 +127,9 @@ func NewRecordVersionsRepository(db database.Executor) *RecordVersionsRepository return &RecordVersionsRepository{db: db} } -// versionKey is the dedupe identity of a version: its CID; for Tap events -// without a CID, a hash of the body; for tombstones, the deleted CID. +// versionKey is the dedupe identity of a version: its CID, or for Tap events +// without a CID, a hash of the body. func versionKey(v RecordVersionWrite) string { - if v.Action == RecordVersionDelete { - return "delete:" + v.CID - } if v.CID != "" { return v.CID } @@ -145,14 +142,14 @@ func versionKey(v RecordVersionWrite) string { } // Append records one version. A version already stored for the URI (same -// version key) is left untouched, so redelivered and resynced events, -// including repeated deletes, are no-ops. +// version key) is left untouched, so redelivered and resynced events are +// no-ops. func (r *RecordVersionsRepository) Append(ctx context.Context, v RecordVersionWrite) error { if v.URI == "" || v.DID == "" || v.Collection == "" { return fmt.Errorf("record version requires uri, did and collection") } switch v.Action { - case RecordVersionCreate, RecordVersionUpdate, RecordVersionDelete: + case RecordVersionCreate, RecordVersionUpdate: default: return fmt.Errorf("unsupported record version action %q", v.Action) } @@ -185,6 +182,16 @@ func (r *RecordVersionsRepository) Append(ctx context.Context, v RecordVersionWr return nil } +// DeleteByURI removes a record's history. Deleting a record in its repo +// deletes its past versions from the index as well. +func (r *RecordVersionsRepository) DeleteByURI(ctx context.Context, uri string) error { + sqlStr := fmt.Sprintf("DELETE FROM record_version WHERE uri = %s", r.db.Placeholder(1)) + if _, err := r.db.Exec(ctx, sqlStr, []database.Value{database.Text(uri)}); err != nil { + return fmt.Errorf("delete record versions for %s: %w", uri, err) + } + return nil +} + // SeedBaseline adds a `baseline` version for every current record in a // matched collection that has no history yet, stamped with the record's // indexed_at. Idempotent: records that already have versions are skipped, so diff --git a/internal/database/repositories/record_versions_test.go b/internal/database/repositories/record_versions_test.go index dcfa3f9..cd9c5a4 100644 --- a/internal/database/repositories/record_versions_test.go +++ b/internal/database/repositories/record_versions_test.go @@ -60,17 +60,15 @@ func TestRecordVersions_AppendDedupesAndListsOldestFirst(t *testing.T) { write(repositories.RecordVersionCreate, "cid1", strPtr(`{"scientificName":"Hirundo rustica","identifiedBy":"ai:gemini-3.8-flash"}`)) write(repositories.RecordVersionCreate, "cid1", strPtr(`{"scientificName":"Hirundo rustica","identifiedBy":"ai:gemini-3.8-flash"}`)) // redelivery write(repositories.RecordVersionUpdate, "cid2", strPtr(`{"scientificName":"Hirundo tahitica","identifiedBy":"Maria"}`)) - write(repositories.RecordVersionDelete, "cid2", nil) - write(repositories.RecordVersionDelete, "cid2", nil) // redelivered delete versions, err := repo.ListByURI(ctx, uri, 0, 500) if err != nil { t.Fatalf("ListByURI: %v", err) } - if len(versions) != 3 { - t.Fatalf("got %d versions, want 3 (redelivery deduped): %+v", len(versions), versions) + if len(versions) != 2 { + t.Fatalf("got %d versions, want 2 (redelivery deduped): %+v", len(versions), versions) } - wantActions := []string{"create", "update", "delete"} + wantActions := []string{"create", "update"} for i, v := range versions { if v.Action != wantActions[i] { t.Errorf("version %d action = %q, want %q", i, v.Action, wantActions[i]) @@ -82,8 +80,11 @@ func TestRecordVersions_AppendDedupesAndListsOldestFirst(t *testing.T) { if versions[1].JSON == nil || versions[1].CID != "cid2" { t.Errorf("update version = %+v", versions[1]) } - if versions[2].JSON != nil { - t.Errorf("delete tombstone should have no body, got %q", *versions[2].JSON) + if err := repo.DeleteByURI(ctx, uri); err != nil { + t.Fatalf("DeleteByURI: %v", err) + } + if remaining, err := repo.ListByURI(ctx, uri, 0, 500); err != nil || len(remaining) != 0 { + t.Fatalf("history after DeleteByURI = %d versions (err %v), want none", len(remaining), err) } if versions[0].Live == nil || !*versions[0].Live { t.Errorf("live flag not kept: %+v", versions[0].Live) @@ -190,3 +191,31 @@ func TestRecordVersions_CIDLessVersionsStayDistinctAndPages(t *testing.T) { }) } } + +func TestRecordVersions_AccountPurgeRemovesHistory(t *testing.T) { + for name, db := range recordVersionTestDBs(t) { + t.Run(name, func(t *testing.T) { + ctx := context.Background() + suffix := strconv.FormatInt(time.Now().UnixNano(), 36) + gone, kept := "did:plc:gone"+suffix, "did:plc:kept"+suffix + for _, did := range []string{gone, kept} { + body := `{"scientificName":"Quercus robur"}` + if err := db.RecordVersions.Append(ctx, repositories.RecordVersionWrite{ + URI: "at://" + did + "/app.gainforest.dwc.occurrence/1", CID: "c1", DID: did, + Collection: "app.gainforest.dwc.occurrence", Action: repositories.RecordVersionCreate, JSON: &body, + }); err != nil { + t.Fatalf("Append: %v", err) + } + } + if err := db.Records.PurgeActorData(ctx, gone); err != nil { + t.Fatalf("PurgeActorData: %v", err) + } + if v, _ := db.RecordVersions.ListByURI(ctx, "at://"+gone+"/app.gainforest.dwc.occurrence/1", 0, 500); len(v) != 0 { + t.Fatalf("purged account still has %d versions", len(v)) + } + if v, _ := db.RecordVersions.ListByURI(ctx, "at://"+kept+"/app.gainforest.dwc.occurrence/1", 0, 500); len(v) != 1 { + t.Fatalf("other account lost its history: %d versions", len(v)) + } + }) + } +} diff --git a/internal/database/repositories/records.go b/internal/database/repositories/records.go index 7ec7b41..82012e1 100644 --- a/internal/database/repositories/records.go +++ b/internal/database/repositories/records.go @@ -2669,6 +2669,13 @@ func (r *RecordsRepository) PurgeActorData(ctx context.Context, did string) erro return fmt.Errorf("failed to delete records by did: %w", err) } + // Record version history belongs to the account too: a deleted, + // deactivated or taken-down account leaves no queryable past versions. + deleteVersionsSQL := fmt.Sprintf("DELETE FROM record_version WHERE did = %s", r.db.Placeholder(1)) + if _, err := tx.ExecContext(ctx, deleteVersionsSQL, did); err != nil { + return fmt.Errorf("failed to delete record versions by did: %w", err) + } + deleteActorSQL := fmt.Sprintf("DELETE FROM actor WHERE did = %s", r.db.Placeholder(1)) if _, err := tx.ExecContext(ctx, deleteActorSQL, did); err != nil { return fmt.Errorf("failed to delete actor by did: %w", err) diff --git a/internal/graphql/recordhistory/recordhistory.go b/internal/graphql/recordhistory/recordhistory.go index a584713..ef9a925 100644 --- a/internal/graphql/recordhistory/recordhistory.go +++ b/internal/graphql/recordhistory/recordhistory.go @@ -17,26 +17,26 @@ import ( // DefaultPageSize is the number of versions returned when `first` is omitted. const DefaultPageSize = 100 -// Type is one observed version of a record, or a delete tombstone. +// Type is one observed version of a record. var Type = graphql.NewObject(graphql.ObjectConfig{ Name: "RecordVersion", - Description: "One observed version of a record (oldest first), or a delete tombstone. Only kept for collections the indexer is configured to track.", + Description: "One observed version of a record (oldest first). Only kept for collections the indexer is configured to track; deleting the record or its account removes its history.", Fields: graphql.Fields{ "id": &graphql.Field{ Type: graphql.NewNonNull(graphql.String), Description: "Monotonic version id; later versions have larger ids.", }, "uri": &graphql.Field{Type: graphql.NewNonNull(graphql.String), Description: "Record AT-URI."}, - "cid": &graphql.Field{Type: graphql.NewNonNull(graphql.String), Description: "CID of this version (for deletes, the version that was deleted)."}, + "cid": &graphql.Field{Type: graphql.NewNonNull(graphql.String), Description: "CID of this version."}, "did": &graphql.Field{Type: graphql.NewNonNull(graphql.String), Description: "Repository DID."}, "collection": &graphql.Field{Type: graphql.NewNonNull(graphql.String), Description: "Collection NSID."}, "action": &graphql.Field{ Type: graphql.NewNonNull(graphql.String), - Description: "baseline (current when history was switched on), create, update, or delete.", + Description: "baseline (current when history was switched on), create, or update.", }, "value": &graphql.Field{ Type: types.JSONScalar, - Description: "The record body at this version; null for deletes.", + Description: "The record body at this version.", }, "live": &graphql.Field{ Type: graphql.Boolean, diff --git a/internal/tap/handler.go b/internal/tap/handler.go index 3dd3ff7..9769918 100644 --- a/internal/tap/handler.go +++ b/internal/tap/handler.go @@ -2,8 +2,6 @@ package tap import ( "context" - "database/sql" - "errors" "fmt" "log/slog" "strings" @@ -27,8 +25,10 @@ type IndexHandler struct { historyCollections repositories.CollectionMatcher } -// WithRecordHistory enables append-only version history for the collections -// the matcher selects. Every distinct version (CID) and delete is recorded. +// WithRecordHistory gives the handler the version store. New versions are +// recorded for the collections the matcher selects (an empty matcher records +// none); deletes always purge a record's stored history, even for +// collections no longer tracked. func (h *IndexHandler) WithRecordHistory(versions *repositories.RecordVersionsRepository, collections repositories.CollectionMatcher) *IndexHandler { h.versions = versions h.historyCollections = collections @@ -138,35 +138,17 @@ func (h *IndexHandler) HandleRecord(ctx context.Context, event *RecordEvent) err } case ActionDelete: - // Tombstone the version being deleted before removing it. If either - // write fails the event is retried: the tombstone insert is idempotent - // (keyed by the deleted CID) and the record is still there to delete. - if h.keepsHistory(event.Collection) { - current, err := h.records.GetByURI(ctx, uri) - switch { - case err == nil: - live := event.Live - if err := h.versions.Append(ctx, repositories.RecordVersionWrite{ - URI: uri, - CID: current.CID, - DID: event.DID, - Collection: event.Collection, - Action: repositories.RecordVersionDelete, - Live: &live, - }); err != nil { - return fmt.Errorf("failed to record delete in version history: %w", err) - } - case errors.Is(err, sql.ErrNoRows): - // Nothing indexed to delete (or already deleted and tombstoned). - default: - return fmt.Errorf("failed to read record before delete: %w", err) - } - } - deleted, err := h.records.DeleteReturning(ctx, uri) if err != nil { return fmt.Errorf("failed to delete record: %w", err) } + // A deleted record takes its history with it. Runs on every delivery, + // so a failed purge is retried when Tap redelivers the event. + if h.versions != nil { + if err := h.versions.DeleteByURI(ctx, uri); err != nil { + return fmt.Errorf("failed to delete record version history: %w", err) + } + } if h.pubsub != nil { previousCID := "" deleteDID := event.DID diff --git a/internal/tap/handler_history_test.go b/internal/tap/handler_history_test.go index a166fd2..0c56b68 100644 --- a/internal/tap/handler_history_test.go +++ b/internal/tap/handler_history_test.go @@ -9,67 +9,102 @@ import ( "github.com/GainForest/hyperindex/internal/tap" ) +const historyURI = "at://did:plc:alice/app.gainforest.dwc.occurrence/occ1" + +func occurrenceEvent(action tap.ActionType, cid, body string, live bool) *tap.RecordEvent { + event := &tap.RecordEvent{ + Live: live, DID: "did:plc:alice", Collection: "app.gainforest.dwc.occurrence", RKey: "occ1", + Action: action, CID: cid, + } + if body != "" { + event.Record = json.RawMessage(body) + } + return event +} + +func versionActions(t *testing.T, repo *repositories.RecordVersionsRepository, uri string) []string { + t.Helper() + versions, err := repo.ListByURI(context.Background(), uri, 0, 500) + if err != nil { + t.Fatalf("ListByURI: %v", err) + } + actions := make([]string, 0, len(versions)) + for _, v := range versions { + actions = append(actions, v.Action+":"+v.CID) + } + return actions +} + func TestIndexHandler_RecordHistory(t *testing.T) { handler, db, _ := setupHandler(t) handler.WithRecordHistory(db.RecordVersions, repositories.NewCollectionMatcher("app.gainforest.dwc.occurrence")) ctx := context.Background() - occurrence := func(action tap.ActionType, cid, body string, live bool) *tap.RecordEvent { - return &tap.RecordEvent{ - Live: live, DID: "did:plc:alice", Collection: "app.gainforest.dwc.occurrence", RKey: "occ1", - Action: action, CID: cid, Record: json.RawMessage(body), - } - } - events := []*tap.RecordEvent{ - occurrence(tap.ActionCreate, "cid1", `{"scientificName":"Hirundo rustica","identifiedBy":"ai:gemini-3.8-flash"}`, true), - occurrence(tap.ActionCreate, "cid1", `{"scientificName":"Hirundo rustica","identifiedBy":"ai:gemini-3.8-flash"}`, false), // resync of the same version - occurrence(tap.ActionUpdate, "cid2", `{"scientificName":"Hirundo tahitica","identifiedBy":"Maria"}`, true), - {Live: true, DID: "did:plc:alice", Collection: "app.gainforest.dwc.occurrence", RKey: "occ1", Action: tap.ActionDelete}, - {Live: true, DID: "did:plc:alice", Collection: "app.gainforest.dwc.occurrence", RKey: "occ1", Action: tap.ActionDelete}, // redelivered delete + for i, event := range []*tap.RecordEvent{ + occurrenceEvent(tap.ActionCreate, "cid1", `{"scientificName":"Hirundo rustica","identifiedBy":"ai:gemini-3.8-flash"}`, true), + occurrenceEvent(tap.ActionCreate, "cid1", `{"scientificName":"Hirundo rustica","identifiedBy":"ai:gemini-3.8-flash"}`, false), // resync of the same version + occurrenceEvent(tap.ActionUpdate, "cid2", `{"scientificName":"Hirundo tahitica","identifiedBy":"Maria"}`, true), // Collections without history are indexed as before and leave no versions. {Live: true, DID: "did:plc:alice", Collection: "app.bsky.feed.post", RKey: "p1", Action: tap.ActionCreate, CID: "cidp", Record: json.RawMessage(`{"text":"hi"}`)}, - } - for i, event := range events { + } { if err := handler.HandleRecord(ctx, event); err != nil { t.Fatalf("event %d: HandleRecord: %v", i, err) } } - versions, err := db.RecordVersions.ListByURI(ctx, "at://did:plc:alice/app.gainforest.dwc.occurrence/occ1", 0, 500) - if err != nil { - t.Fatalf("ListByURI: %v", err) - } - var actions []string - for _, v := range versions { - actions = append(actions, v.Action+":"+v.CID) + got := versionActions(t, db.RecordVersions, historyURI) + if len(got) != 2 || got[0] != "create:cid1" || got[1] != "update:cid2" { + t.Fatalf("versions = %v, want [create:cid1 update:cid2]", got) } - want := []string{"create:cid1", "update:cid2", "delete:cid2"} - if len(actions) != len(want) { - t.Fatalf("versions = %v, want %v", actions, want) + if posts := versionActions(t, db.RecordVersions, "at://did:plc:alice/app.bsky.feed.post/p1"); len(posts) != 0 { + t.Fatalf("untracked collection versions = %v, want none", posts) } - for i := range want { - if actions[i] != want[i] { - t.Fatalf("versions = %v, want %v", actions, want) + + // Deleting the record removes its history; a redelivered delete is fine. + for i := 0; i < 2; i++ { + if err := handler.HandleRecord(ctx, occurrenceEvent(tap.ActionDelete, "", "", true)); err != nil { + t.Fatalf("delete %d: HandleRecord: %v", i, err) } } + if got := versionActions(t, db.RecordVersions, historyURI); len(got) != 0 { + t.Fatalf("versions after delete = %v, want none", got) + } +} - posts, err := db.RecordVersions.ListByURI(ctx, "at://did:plc:alice/app.bsky.feed.post/p1", 0, 500) - if err != nil || len(posts) != 0 { - t.Fatalf("untracked collection versions = %v (err %v), want none", posts, err) +func TestIndexHandler_DeletePurgesHistoryAfterTrackingStops(t *testing.T) { + handler, db, _ := setupHandler(t) + ctx := context.Background() + // History was recorded while the collection was tracked... + body := `{"scientificName":"Quercus robur"}` + if err := db.RecordVersions.Append(ctx, repositories.RecordVersionWrite{ + URI: historyURI, CID: "cid1", DID: "did:plc:alice", Collection: "app.gainforest.dwc.occurrence", + Action: repositories.RecordVersionCreate, JSON: &body, + }); err != nil { + t.Fatalf("Append: %v", err) + } + // ...and tracking has since been switched off (empty matcher). + handler.WithRecordHistory(db.RecordVersions, repositories.CollectionMatcher{}) + if err := handler.HandleRecord(ctx, occurrenceEvent(tap.ActionCreate, "cid2", `{"scientificName":"Quercus petraea"}`, true)); err != nil { + t.Fatalf("create: %v", err) + } + if got := versionActions(t, db.RecordVersions, historyURI); len(got) != 1 { + t.Fatalf("untracked write recorded a version: %v", got) + } + if err := handler.HandleRecord(ctx, occurrenceEvent(tap.ActionDelete, "", "", true)); err != nil { + t.Fatalf("delete: %v", err) + } + if got := versionActions(t, db.RecordVersions, historyURI); len(got) != 0 { + t.Fatalf("versions after delete = %v, want none", got) } } func TestIndexHandler_NoHistoryByDefault(t *testing.T) { handler, db, _ := setupHandler(t) ctx := context.Background() - if err := handler.HandleRecord(ctx, &tap.RecordEvent{ - Live: true, DID: "did:plc:alice", Collection: "app.gainforest.dwc.occurrence", RKey: "occ2", - Action: tap.ActionCreate, CID: "cid1", Record: json.RawMessage(`{"scientificName":"Quercus robur"}`), - }); err != nil { + if err := handler.HandleRecord(ctx, occurrenceEvent(tap.ActionCreate, "cid1", `{"scientificName":"Quercus robur"}`, true)); err != nil { t.Fatalf("HandleRecord: %v", err) } - versions, err := db.RecordVersions.ListByURI(ctx, "at://did:plc:alice/app.gainforest.dwc.occurrence/occ2", 0, 500) - if err != nil || len(versions) != 0 { - t.Fatalf("versions without WithRecordHistory = %v (err %v), want none", versions, err) + if got := versionActions(t, db.RecordVersions, historyURI); len(got) != 0 { + t.Fatalf("versions without WithRecordHistory = %v, want none", got) } } From 66107bcdd095083a4a7b22ae13f634b5af550aba Mon Sep 17 00:00:00 2001 From: daviddao Date: Wed, 23 Sep 2026 09:45:58 +0200 Subject: [PATCH 2/2] fix: delete history in the same transaction as the record Addresses review on #137: - RecordsRepository.Delete, DeleteReturning and DeleteByDID now delete the record's record_version rows in the same transaction as the record, so no query ever sees history for a deleted record and every delete path (Tap, legacy Jetstream, backfill resets) inherits it. The handler-level purge and RecordVersionsRepository.DeleteByURI are gone. - Migration 016 (renamed record_version_honor_deletes) removes what 015-era builds stored: delete tombstones, versions before a tombstone, and history of records that no longer exist (including purged accounts), then adds the did index. --- cmd/hyperindex/main.go | 3 - .../database/migrations/migrations_test.go | 64 +++++++++++++++ .../016_add_record_version_did_index.down.sql | 1 - .../016_add_record_version_did_index.up.sql | 2 - .../016_record_version_honor_deletes.down.sql | 2 + .../016_record_version_honor_deletes.up.sql | 16 ++++ .../016_add_record_version_did_index.down.sql | 1 - .../016_add_record_version_did_index.up.sql | 2 - .../016_record_version_honor_deletes.down.sql | 2 + .../016_record_version_honor_deletes.up.sql | 16 ++++ .../database/repositories/record_versions.go | 12 +-- .../repositories/record_versions_test.go | 8 +- internal/database/repositories/records.go | 77 +++++++++++++++---- internal/tap/handler.go | 14 +--- internal/tap/handler_history_test.go | 3 +- 15 files changed, 171 insertions(+), 52 deletions(-) delete mode 100644 internal/database/migrations/postgres/016_add_record_version_did_index.down.sql delete mode 100644 internal/database/migrations/postgres/016_add_record_version_did_index.up.sql create mode 100644 internal/database/migrations/postgres/016_record_version_honor_deletes.down.sql create mode 100644 internal/database/migrations/postgres/016_record_version_honor_deletes.up.sql delete mode 100644 internal/database/migrations/sqlite/016_add_record_version_did_index.down.sql delete mode 100644 internal/database/migrations/sqlite/016_add_record_version_did_index.up.sql create mode 100644 internal/database/migrations/sqlite/016_record_version_honor_deletes.down.sql create mode 100644 internal/database/migrations/sqlite/016_record_version_honor_deletes.up.sql diff --git a/cmd/hyperindex/main.go b/cmd/hyperindex/main.go index b2fdaea..e6956b3 100644 --- a/cmd/hyperindex/main.go +++ b/cmd/hyperindex/main.go @@ -954,9 +954,6 @@ func startTap( // Create handler that stores records and publishes to subscriptions. handler := tap.NewIndexHandler(svc.records, svc.actors, svc.activity, pubsub, svc.validator) - // Deletes purge stored history even when no collection is tracked, so the - // version store is always wired; the matcher decides what gets recorded. - handler.WithRecordHistory(svc.recordVersions, repositories.CollectionMatcher{}) if historyCollections := repositories.NewCollectionMatcher(cfg.RecordHistoryCollections); !historyCollections.Empty() { // Seed the current version of every opted-in record before live events // start, so each history begins with what was indexed at switch-on. diff --git a/internal/database/migrations/migrations_test.go b/internal/database/migrations/migrations_test.go index 07ba40a..fde594c 100644 --- a/internal/database/migrations/migrations_test.go +++ b/internal/database/migrations/migrations_test.go @@ -903,3 +903,67 @@ func assertPostgresSequenceExists(ctx context.Context, t *testing.T, exec *postg t.Errorf("postgres sequence %q count = %d, want 1", sequenceName, count) } } + +func TestMigrations_016RemovesHistoryOfDeletedRecords(t *testing.T) { + exec := newTestExecutor(t) + ctx := context.Background() + if err := migrations.Run(ctx, exec); err != nil { + t.Fatalf("Run() error = %v", err) + } + if err := migrations.Rollback(ctx, exec); err != nil { + t.Fatalf("Rollback(016) error = %v", err) + } + + // State left by the 015-era handler: a live record with history, a + // deleted record with a tombstone, a recreated record whose earlier life + // ended in a tombstone, and history of a purged account. + db := exec.DB() + mustExec := func(query string, args ...any) { + t.Helper() + if _, err := db.ExecContext(ctx, query, args...); err != nil { + t.Fatalf("exec %q: %v", query, err) + } + } + mustExec(`INSERT INTO record (uri, cid, did, collection, json) VALUES + ('at://did:plc:a/c/live', 'l2', 'did:plc:a', 'c', '{}'), + ('at://did:plc:a/c/recreated', 'r3', 'did:plc:a', 'c', '{}')`) + insertVersion := func(uri, cid, action string) { + t.Helper() + mustExec(`INSERT INTO record_version (uri, cid, version_key, did, collection, action, json) VALUES (?, ?, ?, 'did:plc:a', 'c', ?, '{}')`, + uri, cid, action+":"+cid, action) + } + insertVersion("at://did:plc:a/c/live", "l1", "create") + insertVersion("at://did:plc:a/c/live", "l2", "update") + insertVersion("at://did:plc:a/c/deleted", "d1", "create") + insertVersion("at://did:plc:a/c/deleted", "d1", "delete") + insertVersion("at://did:plc:a/c/recreated", "r1", "create") + insertVersion("at://did:plc:a/c/recreated", "r1", "delete") + insertVersion("at://did:plc:a/c/recreated", "r3", "create") + insertVersion("at://did:plc:purged/c/x", "p1", "create") + + if err := migrations.Run(ctx, exec); err != nil { + t.Fatalf("Run() applying 016 error = %v", err) + } + + rows, err := db.QueryContext(ctx, `SELECT uri || ' ' || cid || ' ' || action FROM record_version ORDER BY id`) + if err != nil { + t.Fatalf("query record_version: %v", err) + } + defer func() { _ = rows.Close() }() + var got []string + for rows.Next() { + var row string + if err := rows.Scan(&row); err != nil { + t.Fatalf("scan: %v", err) + } + got = append(got, row) + } + want := []string{ + "at://did:plc:a/c/live l1 create", + "at://did:plc:a/c/live l2 update", + "at://did:plc:a/c/recreated r3 create", + } + if strings.Join(got, "\n") != strings.Join(want, "\n") { + t.Fatalf("record_version after 016 =\n%s\nwant\n%s", strings.Join(got, "\n"), strings.Join(want, "\n")) + } +} diff --git a/internal/database/migrations/postgres/016_add_record_version_did_index.down.sql b/internal/database/migrations/postgres/016_add_record_version_did_index.down.sql deleted file mode 100644 index 684e9ae..0000000 --- a/internal/database/migrations/postgres/016_add_record_version_did_index.down.sql +++ /dev/null @@ -1 +0,0 @@ -DROP INDEX IF EXISTS idx_record_version_did; diff --git a/internal/database/migrations/postgres/016_add_record_version_did_index.up.sql b/internal/database/migrations/postgres/016_add_record_version_did_index.up.sql deleted file mode 100644 index a6bb1d3..0000000 --- a/internal/database/migrations/postgres/016_add_record_version_did_index.up.sql +++ /dev/null @@ -1,2 +0,0 @@ --- Account purges delete an account's record history by DID. -CREATE INDEX IF NOT EXISTS idx_record_version_did ON record_version(did); diff --git a/internal/database/migrations/postgres/016_record_version_honor_deletes.down.sql b/internal/database/migrations/postgres/016_record_version_honor_deletes.down.sql new file mode 100644 index 0000000..effc063 --- /dev/null +++ b/internal/database/migrations/postgres/016_record_version_honor_deletes.down.sql @@ -0,0 +1,2 @@ +-- The history removed on upgrade is not restored. +DROP INDEX IF EXISTS idx_record_version_did; diff --git a/internal/database/migrations/postgres/016_record_version_honor_deletes.up.sql b/internal/database/migrations/postgres/016_record_version_honor_deletes.up.sql new file mode 100644 index 0000000..7f2dce7 --- /dev/null +++ b/internal/database/migrations/postgres/016_record_version_honor_deletes.up.sql @@ -0,0 +1,16 @@ +-- Deleting a record, or purging its account, now removes its version history. +-- Earlier builds kept delete tombstones and the versions before them, and +-- account purges left history behind; remove what they stored. +DELETE FROM record_version +WHERE EXISTS ( + SELECT 1 FROM record_version tombstone + WHERE tombstone.uri = record_version.uri + AND tombstone.action = 'delete' + AND tombstone.id > record_version.id +); +DELETE FROM record_version +WHERE action = 'delete' + OR NOT EXISTS (SELECT 1 FROM record r WHERE r.uri = record_version.uri); + +-- Account purges delete an account's record history by DID. +CREATE INDEX IF NOT EXISTS idx_record_version_did ON record_version(did); diff --git a/internal/database/migrations/sqlite/016_add_record_version_did_index.down.sql b/internal/database/migrations/sqlite/016_add_record_version_did_index.down.sql deleted file mode 100644 index 684e9ae..0000000 --- a/internal/database/migrations/sqlite/016_add_record_version_did_index.down.sql +++ /dev/null @@ -1 +0,0 @@ -DROP INDEX IF EXISTS idx_record_version_did; diff --git a/internal/database/migrations/sqlite/016_add_record_version_did_index.up.sql b/internal/database/migrations/sqlite/016_add_record_version_did_index.up.sql deleted file mode 100644 index a6bb1d3..0000000 --- a/internal/database/migrations/sqlite/016_add_record_version_did_index.up.sql +++ /dev/null @@ -1,2 +0,0 @@ --- Account purges delete an account's record history by DID. -CREATE INDEX IF NOT EXISTS idx_record_version_did ON record_version(did); diff --git a/internal/database/migrations/sqlite/016_record_version_honor_deletes.down.sql b/internal/database/migrations/sqlite/016_record_version_honor_deletes.down.sql new file mode 100644 index 0000000..effc063 --- /dev/null +++ b/internal/database/migrations/sqlite/016_record_version_honor_deletes.down.sql @@ -0,0 +1,2 @@ +-- The history removed on upgrade is not restored. +DROP INDEX IF EXISTS idx_record_version_did; diff --git a/internal/database/migrations/sqlite/016_record_version_honor_deletes.up.sql b/internal/database/migrations/sqlite/016_record_version_honor_deletes.up.sql new file mode 100644 index 0000000..7f2dce7 --- /dev/null +++ b/internal/database/migrations/sqlite/016_record_version_honor_deletes.up.sql @@ -0,0 +1,16 @@ +-- Deleting a record, or purging its account, now removes its version history. +-- Earlier builds kept delete tombstones and the versions before them, and +-- account purges left history behind; remove what they stored. +DELETE FROM record_version +WHERE EXISTS ( + SELECT 1 FROM record_version tombstone + WHERE tombstone.uri = record_version.uri + AND tombstone.action = 'delete' + AND tombstone.id > record_version.id +); +DELETE FROM record_version +WHERE action = 'delete' + OR NOT EXISTS (SELECT 1 FROM record r WHERE r.uri = record_version.uri); + +-- Account purges delete an account's record history by DID. +CREATE INDEX IF NOT EXISTS idx_record_version_did ON record_version(did); diff --git a/internal/database/repositories/record_versions.go b/internal/database/repositories/record_versions.go index 3803d95..d8597cf 100644 --- a/internal/database/repositories/record_versions.go +++ b/internal/database/repositories/record_versions.go @@ -25,7 +25,7 @@ const ( ) // RecordVersion is one observed version of a record. Deleting a record (or -// its account) removes its history, so there are no delete entries. +// purging its account) removes its history, see RecordsRepository.Delete. type RecordVersion struct { ID int64 URI string @@ -182,16 +182,6 @@ func (r *RecordVersionsRepository) Append(ctx context.Context, v RecordVersionWr return nil } -// DeleteByURI removes a record's history. Deleting a record in its repo -// deletes its past versions from the index as well. -func (r *RecordVersionsRepository) DeleteByURI(ctx context.Context, uri string) error { - sqlStr := fmt.Sprintf("DELETE FROM record_version WHERE uri = %s", r.db.Placeholder(1)) - if _, err := r.db.Exec(ctx, sqlStr, []database.Value{database.Text(uri)}); err != nil { - return fmt.Errorf("delete record versions for %s: %w", uri, err) - } - return nil -} - // SeedBaseline adds a `baseline` version for every current record in a // matched collection that has no history yet, stamped with the record's // indexed_at. Idempotent: records that already have versions are skipped, so diff --git a/internal/database/repositories/record_versions_test.go b/internal/database/repositories/record_versions_test.go index cd9c5a4..b71b6db 100644 --- a/internal/database/repositories/record_versions_test.go +++ b/internal/database/repositories/record_versions_test.go @@ -80,11 +80,13 @@ func TestRecordVersions_AppendDedupesAndListsOldestFirst(t *testing.T) { if versions[1].JSON == nil || versions[1].CID != "cid2" { t.Errorf("update version = %+v", versions[1]) } - if err := repo.DeleteByURI(ctx, uri); err != nil { - t.Fatalf("DeleteByURI: %v", err) + // Deleting the record (even one missing from the record table) + // removes its history. + if err := db.Records.Delete(ctx, uri); err != nil { + t.Fatalf("Records.Delete: %v", err) } if remaining, err := repo.ListByURI(ctx, uri, 0, 500); err != nil || len(remaining) != 0 { - t.Fatalf("history after DeleteByURI = %d versions (err %v), want none", len(remaining), err) + t.Fatalf("history after Records.Delete = %d versions (err %v), want none", len(remaining), err) } if versions[0].Live == nil || !*versions[0].Live { t.Errorf("live flag not kept: %+v", versions[0].Live) diff --git a/internal/database/repositories/records.go b/internal/database/repositories/records.go index 82012e1..1a390b7 100644 --- a/internal/database/repositories/records.go +++ b/internal/database/repositories/records.go @@ -2620,21 +2620,36 @@ func (r *RecordsRepository) GetByDID(ctx context.Context, did string) ([]*Record // missing row returns (nil, nil), allowing duplicate delete deliveries to stay // idempotent without reusing stale pre-delete visibility metadata. func (r *RecordsRepository) DeleteReturning(ctx context.Context, uri string) (*Record, error) { - sqlStr := fmt.Sprintf("DELETE FROM record WHERE uri = %s RETURNING %s", r.db.Placeholder(1), r.recordColumns()) + tx, err := r.db.BeginTx(ctx, nil) + if err != nil { + return nil, fmt.Errorf("failed to begin transaction: %w", err) + } + defer tx.Rollback() // Rollback is a no-op after successful commit. + if err := r.deleteVersionHistoryTx(ctx, tx, "uri", uri); err != nil { + return nil, err + } + + sqlStr := fmt.Sprintf("DELETE FROM record WHERE uri = %s RETURNING %s", r.db.Placeholder(1), r.recordColumns()) var rec Record var indexedAtStr string var validationStatus string var validationError, validatedAtStr, lexiconHash sql.NullString - err := r.db.QueryRow(ctx, sqlStr, []database.Value{database.Text(uri)}, + err = tx.QueryRowContext(ctx, sqlStr, r.db.ConvertParams([]database.Value{database.Text(uri)})...).Scan( &rec.URI, &rec.CID, &rec.DID, &rec.Collection, &rec.JSON, &indexedAtStr, &rec.RKey, &validationStatus, &validationError, &validatedAtStr, &lexiconHash) + deleted := true if errors.Is(err, sql.ErrNoRows) { - return nil, nil - } - if err != nil { + deleted = false + } else if err != nil { return nil, err } + if err := tx.Commit(); err != nil { + return nil, fmt.Errorf("failed to commit transaction: %w", err) + } + if !deleted { + return nil, nil + } rec.IndexedAt = atproto.ParseTimestamp(indexedAtStr) rec.ValidationStatus = validation.Status(validationStatus) @@ -2642,18 +2657,49 @@ func (r *RecordsRepository) DeleteReturning(ctx context.Context, uri string) (*R return &rec, nil } -// Delete removes a record by URI. +// Delete removes a record by URI, together with its version history. func (r *RecordsRepository) Delete(ctx context.Context, uri string) error { - sqlStr := fmt.Sprintf("DELETE FROM record WHERE uri = %s", r.db.Placeholder(1)) - _, err := r.db.Exec(ctx, sqlStr, []database.Value{database.Text(uri)}) - return err + return r.deleteWithHistory(ctx, "uri", uri) } -// DeleteByDID removes all records for a specific DID. +// DeleteByDID removes all records for a specific DID, together with their +// version history. func (r *RecordsRepository) DeleteByDID(ctx context.Context, did string) error { - sqlStr := fmt.Sprintf("DELETE FROM record WHERE did = %s", r.db.Placeholder(1)) - _, err := r.db.Exec(ctx, sqlStr, []database.Value{database.Text(did)}) - return err + return r.deleteWithHistory(ctx, "did", did) +} + +// deleteWithHistory deletes records matching column = value and their +// record_version history in one transaction, so a deleted record never +// leaves past versions queryable. +func (r *RecordsRepository) deleteWithHistory(ctx context.Context, column, value string) error { + tx, err := r.db.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("failed to begin transaction: %w", err) + } + defer tx.Rollback() // Rollback is a no-op after successful commit. + + if err := r.deleteVersionHistoryTx(ctx, tx, column, value); err != nil { + return err + } + sqlStr := fmt.Sprintf("DELETE FROM record WHERE %s = %s", column, r.db.Placeholder(1)) + if _, err := tx.ExecContext(ctx, sqlStr, value); err != nil { + return fmt.Errorf("failed to delete records by %s: %w", column, err) + } + if err := tx.Commit(); err != nil { + return fmt.Errorf("failed to commit transaction: %w", err) + } + return nil +} + +// deleteVersionHistoryTx removes record_version rows matching column = value +// ("uri" or "did"). Deleting a record, or purging its account, removes its +// history too. +func (r *RecordsRepository) deleteVersionHistoryTx(ctx context.Context, tx *sql.Tx, column, value string) error { + sqlStr := fmt.Sprintf("DELETE FROM record_version WHERE %s = %s", column, r.db.Placeholder(1)) + if _, err := tx.ExecContext(ctx, sqlStr, value); err != nil { + return fmt.Errorf("failed to delete record versions by %s: %w", column, err) + } + return nil } // PurgeActorData atomically removes all records and the actor row for a DID. @@ -2671,9 +2717,8 @@ func (r *RecordsRepository) PurgeActorData(ctx context.Context, did string) erro // Record version history belongs to the account too: a deleted, // deactivated or taken-down account leaves no queryable past versions. - deleteVersionsSQL := fmt.Sprintf("DELETE FROM record_version WHERE did = %s", r.db.Placeholder(1)) - if _, err := tx.ExecContext(ctx, deleteVersionsSQL, did); err != nil { - return fmt.Errorf("failed to delete record versions by did: %w", err) + if err := r.deleteVersionHistoryTx(ctx, tx, "did", did); err != nil { + return err } deleteActorSQL := fmt.Sprintf("DELETE FROM actor WHERE did = %s", r.db.Placeholder(1)) diff --git a/internal/tap/handler.go b/internal/tap/handler.go index 9769918..9eabd5e 100644 --- a/internal/tap/handler.go +++ b/internal/tap/handler.go @@ -25,10 +25,9 @@ type IndexHandler struct { historyCollections repositories.CollectionMatcher } -// WithRecordHistory gives the handler the version store. New versions are -// recorded for the collections the matcher selects (an empty matcher records -// none); deletes always purge a record's stored history, even for -// collections no longer tracked. +// WithRecordHistory records every distinct version of records in the +// collections the matcher selects. Deleting a record removes its history in +// the records repository, whether or not history is enabled. func (h *IndexHandler) WithRecordHistory(versions *repositories.RecordVersionsRepository, collections repositories.CollectionMatcher) *IndexHandler { h.versions = versions h.historyCollections = collections @@ -142,13 +141,6 @@ func (h *IndexHandler) HandleRecord(ctx context.Context, event *RecordEvent) err if err != nil { return fmt.Errorf("failed to delete record: %w", err) } - // A deleted record takes its history with it. Runs on every delivery, - // so a failed purge is retried when Tap redelivers the event. - if h.versions != nil { - if err := h.versions.DeleteByURI(ctx, uri); err != nil { - return fmt.Errorf("failed to delete record version history: %w", err) - } - } if h.pubsub != nil { previousCID := "" deleteDID := event.DID diff --git a/internal/tap/handler_history_test.go b/internal/tap/handler_history_test.go index 0c56b68..6b3ce14 100644 --- a/internal/tap/handler_history_test.go +++ b/internal/tap/handler_history_test.go @@ -82,8 +82,7 @@ func TestIndexHandler_DeletePurgesHistoryAfterTrackingStops(t *testing.T) { }); err != nil { t.Fatalf("Append: %v", err) } - // ...and tracking has since been switched off (empty matcher). - handler.WithRecordHistory(db.RecordVersions, repositories.CollectionMatcher{}) + // ...and tracking has since been switched off (no WithRecordHistory). if err := handler.HandleRecord(ctx, occurrenceEvent(tap.ActionCreate, "cid2", `{"scientificName":"Quercus petraea"}`, true)); err != nil { t.Fatalf("create: %v", err) }