Opt-in record version history (recordHistory query)

.agents/skills/hyperindex/SKILL.md

@@ -257,6 +257,22 @@ Variables:
 
 Use typed collection queries when the caller needs collection-specific filters, sorting, exact totals, or typed fields. Use `recordTimeline` when the primary requirement is one stable newest-first page across selected collections.
 
+## Record version history
+
+For collections the deployment tracks, `recordHistory(uri: "at://…")` returns every observed version of one record, oldest first, with the full record body in `value`. Use it to show how a record changed, for example who renamed an observation:
+
+```graphql
+{
+  recordHistory(uri: "at://did:plc:example/app.gainforest.dwc.occurrence/3kabc") {
+    action
+    observedAt
+    value
+  }
+}
+```
+
+The first entry may be a `baseline`: the version indexed when history was switched on. Untracked collections return an empty list.
+
 ## External labeler filtering
 
 Use external label filtering only after confirming that the target endpoint exposes external label support.

.agents/skills/hyperindex/references/schema-reference.md

@@ -50,6 +50,7 @@ Last updated on 2026-07-30 for pending record validation gate and record author
 | `orgHypercertsWorkscopeTag` | after: `String`, before: `String`, first: `Int`, last: `Int`, sortBy: `OrgHypercertsWorkscopeTagSortField`, sortDirection: `SortDirection`, where: `OrgHypercertsWorkscopeTagWhereInput` | Query org.hypercerts.workscope.tag records |
 | `orgHypercertsWorkscopeTagByUri` | uri: `String!` | Get a single org.hypercerts.workscope.tag by AT-URI |
 | `records` | after: `String`, before: `String`, collection: `String!`, first: `Int`, last: `Int` | Query raw records from any collection, including records hidden from typed GraphQL by validation metadata. |
+| `recordHistory` | first: `Int`, uri: `String!` | Every version the indexer has observed for one record, oldest first. Empty unless the record's collection is configured for history. |
 | `recordTimeline` | after: `String`, first: `Int`, where: `RecordTimelineWhereInput!` | Query a newest-first page of current records across selected collections, optionally filtered by author DIDs. |
 | `externalLabels` | activeOnly: `Boolean`, sources: `[String!]`, subjects: `[String!]!`, values: `[String!]` | Query locally ingested external ATProto labels by DID or AT-URI subject. |
 | `search` | after: `String`, collection: `String`, first: `Int`, query: `String!` | Search records by text content |
@@ -346,6 +347,24 @@ External label predicates bound by the containing filter field.
 | `has` | `ExternalLabelPredicateInput` | Keep records whose bound label subject has a matching external label. |
 | `none` | `ExternalLabelPredicateInput` | Keep records whose bound label subject does not have a matching external label. |
 
+## Record version history
+
+`recordHistory(uri: String!, first: Int = 100)` returns `[RecordVersion!]!`, oldest first, for collections the deployment tracks (`RECORD_HISTORY_COLLECTIONS`; production tracks `app.gainforest.dwc.occurrence`). Other collections return an empty list. A `baseline` version is what was indexed when history was switched on; earlier edits are not available.
+
+### `RecordVersion`
+
+| Field | Type | Description |
+| --- | --- | --- |
+| `action` | `String!` | `baseline`, `create`, `update`, or `delete`. |
+| `cid` | `String!` | CID of this version (for deletes, the version that was deleted). |
+| `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. |
+
 ## External label support
 
 Production exposes external ATProto labels through the root `externalLabels` query, each generated record type's virtual `externalLabels` field, `where.externalLabels.has` / `where.externalLabels.none` predicates for record AT-URI labels, and `where.authorLabels.has` / `where.authorLabels.none` predicates for author DID labels on typed list queries.

.changes/unreleased/add-record-version-history.yaml

@@ -0,0 +1,5 @@
+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) GraphQL query returns them oldest first. History is off by default; migration 015 adds the table.
+time: 2026-09-23T03:10:00.000000+02:00
+custom:
+  Affects: operator

AGENTS.md

@@ -91,6 +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.
 - 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.

README.md

@@ -120,6 +120,9 @@ TAP_SIGNAL_COLLECTION=app.bsky.feed.post docker compose -f docker-compose.tap.ym
 | `TAP_DISABLE_ACKS` | Disable ack-based delivery (useful for debugging) | `false` |
 | `TAP_SIGNAL_COLLECTION` | Collection NSID for auto-discovery of repos | *(empty)* |
 | `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 every version is kept in `record_version` and served 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://…")`.
 
 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.
 

cmd/hyperindex/main.go

@@ -71,6 +71,7 @@ type services struct {
 	labelDefinitions  *repositories.LabelDefinitionsRepository
 	labelPreferences  *repositories.LabelPreferencesRepository
 	reports           *repositories.ReportsRepository
+	recordVersions    *repositories.RecordVersionsRepository
 	validator         *validation.Validator
 	validationRefresh *validationrefresh.Scheduler
 }
@@ -211,6 +212,7 @@ func initServices(cfg *config.Config) (*services, error) {
 		labelDefinitions: repositories.NewLabelDefinitionsRepository(db),
 		labelPreferences: repositories.NewLabelPreferencesRepository(db),
 		reports:          repositories.NewReportsRepository(db),
+		recordVersions:   repositories.NewRecordVersionsRepository(db),
 	}
 
 	if cfg.PLCDirectoryURL != "" {
@@ -792,6 +794,7 @@ func setupGraphQL(r *chi.Mux, cfg *config.Config, svc *services, pubsub *subscri
 		Actors:         svc.actors,
 		Lexicons:       svc.lexicons,
 		ExternalLabels: svc.externalLabels,
+		RecordVersions: svc.recordVersions,
 	}
 
 	graphqlHandler, err := hgraphql.NewHandler(registry, repos)
@@ -951,6 +954,19 @@ func startTap(
 
 	// Create handler that stores records and publishes to subscriptions.
 	handler := tap.NewIndexHandler(svc.records, svc.actors, svc.activity, pubsub, svc.validator)
+	if historyCollections := repositories.NewCollectionMatcher(cfg.RecordHistoryCollections); !historyCollections.Empty() {
+		handler.WithRecordHistory(svc.recordVersions, historyCollections)
+		// Seed the current version of every opted-in record before live events
+		// start, so each history begins with what was indexed at switch-on.
+		seedCtx, cancelSeed := context.WithTimeout(context.Background(), 10*time.Minute)
+		seeded, err := svc.recordVersions.SeedBaseline(seedCtx, historyCollections)
+		cancelSeed()
+		if err != nil {
+			slog.Error("Failed to seed record version baseline; history starts with the next change", "error", err)
+		} else {
+			slog.Info("Record version history enabled", "collections", cfg.RecordHistoryCollections, "baseline_seeded", seeded)
+		}
+	}
 
 	// Create consumer.
 	consumer := tap.NewConsumer(tap.ConsumerConfig{

internal/config/config.go

@@ -63,6 +63,11 @@ type Config struct {
 	TapDisableAcks   bool   // Fire-and-forget mode (default: false)
 	TapEnabled       bool   // Use Tap instead of Jetstream+Backfill (default: false)
 
+	// Record version history (Tap mode): comma-separated collection NSIDs or
+	// `prefix.*` patterns whose every version is kept in record_version.
+	// Empty (the default) keeps no history.
+	RecordHistoryCollections string
+
 	// Labeler subscriptions
 	LabelerSubscribeEnabled      bool
 	LabelerSubscribeURLs         string // Comma-separated com.atproto.label.subscribeLabels websocket URLs
@@ -127,6 +132,9 @@ func Load() (*Config, error) {
 		TapDisableAcks:   getEnvBool("TAP_DISABLE_ACKS", false),
 		TapEnabled:       getEnvBool("TAP_ENABLED", false),
 
+		// Record version history
+		RecordHistoryCollections: getEnv("RECORD_HISTORY_COLLECTIONS", ""),
+
 		// Labeler subscriptions
 		LabelerSubscribeEnabled:      getEnvBool("LABELER_SUBSCRIBE_ENABLED", false),
 		LabelerSubscribeURLs:         getEnv("LABELER_SUBSCRIBE_URLS", ""),

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{"014", "013", "012", "011"} {
+	for _, version := range []string{"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,43 @@ func TestMigrations_Rollback(t *testing.T) {
 	}
 }
 
+// rollbackRecordVersionMigration undoes 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)
+	}
+}
+
+func TestMigrations_RecordVersionUpAndDown(t *testing.T) {
+	exec := newTestExecutor(t)
+	ctx := context.Background()
+	if err := migrations.Run(ctx, exec); err != nil {
+		t.Fatalf("Run() error = %v", err)
+	}
+	if !sqliteIndexExists(t, exec, "idx_record_version_uri_cid") || !migrationVersionExists(t, exec, "015") {
+		t.Fatal("migration 015 did not create record_version")
+	}
+	rollbackRecordVersionMigration(ctx, t, exec)
+	if sqliteIndexExists(t, exec, "idx_record_version_uri_cid") || migrationVersionExists(t, exec, "015") {
+		t.Fatal("migration 015 schema/version remained after rollback")
+	}
+	if err := migrations.Run(ctx, exec); err != nil {
+		t.Fatalf("re-Run() error = %v", err)
+	}
+	if !migrationVersionExists(t, exec, "015") {
+		t.Fatal("migration 015 was not re-applied")
+	}
+}
+
 func TestMigrations_Rollback014Then013PreservesLexiconData(t *testing.T) {
 	exec := newTestExecutor(t)
 	ctx := context.Background()
 	if err := migrations.Run(ctx, exec); err != nil {
 		t.Fatalf("Run() error = %v", err)
 	}
+	rollbackRecordVersionMigration(ctx, t, exec)
 	const raw = `{"lexicon":1,"id":"com.example.saved","defs":{}}`
 	if _, err := exec.DB().ExecContext(ctx, `INSERT INTO lexicon (id, json, raw_json) VALUES (?, ?, ?)`, "com.example.saved", raw, raw); err != nil {
 		t.Fatalf("insert Lexicon before downgrade: %v", err)
@@ -529,6 +560,7 @@ func TestMigrations_ApplyAndVersionRecordRollbackTogether(t *testing.T) {
 	if err := migrations.Run(ctx, exec); err != nil {
 		t.Fatalf("Run() error = %v", err)
 	}
+	rollbackRecordVersionMigration(ctx, t, exec)
 	if err := migrations.Rollback(ctx, exec); err != nil {
 		t.Fatalf("Rollback(014) error = %v", err)
 	}
@@ -559,6 +591,7 @@ func TestMigrations_DownAndVersionDeleteRollbackTogether(t *testing.T) {
 	if err := migrations.Run(ctx, exec); err != nil {
 		t.Fatalf("Run() error = %v", err)
 	}
+	rollbackRecordVersionMigration(ctx, t, exec)
 	if err := migrations.Rollback(ctx, exec); err != nil {
 		t.Fatalf("Rollback(014) error = %v", err)
 	}

internal/database/migrations/postgres/015_add_record_version.down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS record_version;

internal/database/migrations/postgres/015_add_record_version.up.sql

@@ -0,0 +1,22 @@
+-- Append-only version history for opted-in collections
+-- (RECORD_HISTORY_COLLECTIONS). One row per distinct record version (CID) the
+-- indexer observes, plus delete tombstones. `baseline` rows seed the version
+-- that was current when history was switched on for a collection.
+CREATE TABLE IF NOT EXISTS record_version (
+  id BIGSERIAL PRIMARY KEY,
+  uri TEXT NOT NULL,
+  cid TEXT NOT NULL,
+  did TEXT NOT NULL,
+  collection TEXT NOT NULL,
+  action TEXT NOT NULL CHECK (action IN ('baseline', 'create', 'update', 'delete')),
+  json JSONB,
+  live BOOLEAN,
+  observed_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW()
+);
+
+-- A version is recorded once, however often Tap redelivers or resyncs it.
+CREATE UNIQUE INDEX IF NOT EXISTS idx_record_version_uri_cid
+ON record_version(uri, cid) WHERE action <> 'delete';
+CREATE INDEX IF NOT EXISTS idx_record_version_uri_id ON record_version(uri, id);
+CREATE INDEX IF NOT EXISTS idx_record_version_collection_observed
+ON record_version(collection, observed_at DESC);

internal/database/migrations/sqlite/015_add_record_version.down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS record_version;

internal/database/migrations/sqlite/015_add_record_version.up.sql

@@ -0,0 +1,19 @@
+-- Append-only version history for opted-in collections
+-- (RECORD_HISTORY_COLLECTIONS). See the PostgreSQL migration for details.
+CREATE TABLE IF NOT EXISTS record_version (
+  id INTEGER PRIMARY KEY AUTOINCREMENT,
+  uri TEXT NOT NULL,
+  cid TEXT NOT NULL,
+  did TEXT NOT NULL,
+  collection TEXT NOT NULL,
+  action TEXT NOT NULL CHECK (action IN ('baseline', 'create', 'update', 'delete')),
+  json TEXT,
+  live INTEGER,
+  observed_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now'))
+);
+
+CREATE UNIQUE INDEX IF NOT EXISTS idx_record_version_uri_cid
+ON record_version(uri, cid) WHERE action <> 'delete';
+CREATE INDEX IF NOT EXISTS idx_record_version_uri_id ON record_version(uri, id);
+CREATE INDEX IF NOT EXISTS idx_record_version_collection_observed
+ON record_version(collection, observed_at DESC);