deps: pin pkg/etl v1.6.4 and backfill users + save/repost types

Makefile

@@ -41,8 +41,8 @@ apidiff::
 	open http://localhost:1323/apidiff.html
 
 test-schema::
-	@set -a; \
-	writeDbUrl=postgresql://postgres:example@localhost:21300/postgres; \
+	@set -a -e; \
+	writeDbUrl='postgresql://postgres:example@localhost:21300/postgres?sslmode=disable'; \
 	echo "\033[0;32mBringing down any existing containers to start fresh...\033[0m"; \
 	docker compose down --volumes; \
 	docker compose up -d --wait; \

ddl/migrations/0236_saves_reposts_album_to_playlist.sql

@@ -0,0 +1,164 @@
+-- Collapse save_type / repost_type 'album' back into 'playlist'.
+--
+-- An album is a playlist with is_album = true. The indexer briefly derived a
+-- separate 'album' type by reading playlists.is_album at index time — a side
+-- effect of a fix for entity-id collisions, live from 2026-05-28. It no longer
+-- does (OpenAudio/go-openaudio#428). is_album is mutable while save_type is
+-- written once, so the same chain history indexed at different times produced
+-- different rows. Nothing reads the distinction: every consumer is
+-- track / not-track, or ORs the two together.
+--
+-- THE BINDING CONSTRAINT IS NOT THE PRIMARY KEY. saves_pkey is
+-- (user_id, save_item_id, save_type, txhash) and no two rows collide on it. But
+-- pkg/etl migration 0030 also added
+--
+--   saves_current_uniq_idx    ON saves   (user_id, save_item_id, save_type)     WHERE is_current
+--   reposts_current_uniq_idx  ON reposts (user_id, repost_item_id, repost_type) WHERE is_current
+--
+-- which carry no txhash. Two current rows for the same (user, item) are legal
+-- today precisely because one is 'album' and one is 'playlist'; collapsing the
+-- type makes them duplicates. A blind UPDATE therefore aborts with
+--
+--   duplicate key value violates unique constraint "saves_current_uniq_idx"
+--
+-- taking the whole pre-roll migrate Job with it. 43 save pairs and 34 repost
+-- pairs collide this way, so the pairs are resolved first and the remainder
+-- retyped.
+--
+-- Winner is the highest blocknumber. Verified against a production clone that
+-- this is decidable wherever it matters: of the 43 save pairs, 38 agree on
+-- is_delete (either row would do) and 5 disagree — and in every disagreeing
+-- pair the blocknumbers differ, with the later row holding the correct state
+-- (e.g. user 9014 unfavourited item 613011280 in 2026; the newer row is the
+-- delete). Blocknumber ties occur only among pairs that agree, where the choice
+-- is immaterial: zero rows both disagree and tie. Reposts are the same shape —
+-- 33 agree, 1 disagrees, no tie coincides with a disagreement. The created_at /
+-- type tiebreaks below are therefore never load-bearing; they exist only to
+-- make the statement deterministic.
+--
+-- Losers are demoted, not deleted: unlike users these tables do keep superseded
+-- history (344 saves, 161 reposts on the clone), so is_current = false is the
+-- shape they already use.
+--
+-- Aggregates: this is a net correction. reconcile_aggregates counts these rows
+-- with count(*) and ORs both types, so every colliding pair has been
+-- double-counting aggregate_playlist.save_count / repost_count.
+--
+-- Triggers: on_save / on_repost are suppressed for the whole operation. Their
+-- notification group_id embeds the type ('save:<id>:type:<save_type>'), so both
+-- the demote and the retype would emit fresh favourite/repost notifications
+-- that ON CONFLICT could not dedupe against the existing ':type:album' rows.
+-- trg_saves / trg_reposts stay enabled so the search indexer sees the change.
+-- Aggregate counts are unaffected by the retype either way: handle_save's delta
+-- is transition-aware and evaluates to 0 when is_delete does not change.
+--
+-- ALTER TABLE ... DISABLE TRIGGER takes ShareRowExclusiveLock — it blocks
+-- concurrent writes but not reads, and is held until commit. Blocking writers
+-- is the property we want: they wait rather than silently running trigger-free.
+-- Each table gets its own transaction so the two locks are never held at once.
+--
+-- The DO blocks guard on the trigger existing: pg_migrate.sh applies
+-- migrations/ before functions/, so on a database bootstrapped from ddl/ alone
+-- these triggers do not exist yet.
+--
+-- Re-running is a no-op once no 'album' rows remain.
+
+BEGIN;
+SET LOCAL lock_timeout = '5s';
+
+DO $$
+BEGIN
+    IF EXISTS (SELECT 1 FROM pg_trigger
+               WHERE tgrelid = 'saves'::regclass AND tgname = 'on_save' AND NOT tgisinternal) THEN
+        ALTER TABLE saves DISABLE TRIGGER on_save;
+    END IF;
+END $$;
+
+-- 1. demote the older row of each (user, item) pair holding both types
+WITH ranked AS (
+    SELECT
+        user_id, save_item_id, save_type, txhash,
+        row_number() OVER (
+            PARTITION BY user_id, save_item_id
+            ORDER BY blocknumber DESC NULLS LAST, created_at DESC NULLS LAST, save_type DESC
+        ) AS rn
+    FROM saves
+    WHERE is_current = true
+      AND save_type IN ('playlist', 'album')
+      AND (user_id, save_item_id) IN (
+          SELECT user_id, save_item_id FROM saves
+          WHERE is_current = true AND save_type IN ('playlist', 'album')
+          GROUP BY user_id, save_item_id
+          HAVING count(DISTINCT save_type) > 1
+      )
+)
+UPDATE saves s
+SET is_current = false
+FROM ranked r
+WHERE s.user_id = r.user_id
+  AND s.save_item_id = r.save_item_id
+  AND s.save_type = r.save_type
+  AND s.txhash = r.txhash
+  AND r.rn > 1;
+
+-- 2. retype the survivors
+UPDATE saves SET save_type = 'playlist' WHERE is_current = true AND save_type = 'album';
+
+DO $$
+BEGIN
+    IF EXISTS (SELECT 1 FROM pg_trigger
+               WHERE tgrelid = 'saves'::regclass AND tgname = 'on_save' AND NOT tgisinternal) THEN
+        ALTER TABLE saves ENABLE TRIGGER on_save;
+    END IF;
+END $$;
+
+COMMIT;
+
+BEGIN;
+SET LOCAL lock_timeout = '5s';
+
+DO $$
+BEGIN
+    IF EXISTS (SELECT 1 FROM pg_trigger
+               WHERE tgrelid = 'reposts'::regclass AND tgname = 'on_repost' AND NOT tgisinternal) THEN
+        ALTER TABLE reposts DISABLE TRIGGER on_repost;
+    END IF;
+END $$;
+
+WITH ranked AS (
+    SELECT
+        user_id, repost_item_id, repost_type, txhash,
+        row_number() OVER (
+            PARTITION BY user_id, repost_item_id
+            ORDER BY blocknumber DESC NULLS LAST, created_at DESC NULLS LAST, repost_type DESC
+        ) AS rn
+    FROM reposts
+    WHERE is_current = true
+      AND repost_type IN ('playlist', 'album')
+      AND (user_id, repost_item_id) IN (
+          SELECT user_id, repost_item_id FROM reposts
+          WHERE is_current = true AND repost_type IN ('playlist', 'album')
+          GROUP BY user_id, repost_item_id
+          HAVING count(DISTINCT repost_type) > 1
+      )
+)
+UPDATE reposts s
+SET is_current = false
+FROM ranked r
+WHERE s.user_id = r.user_id
+  AND s.repost_item_id = r.repost_item_id
+  AND s.repost_type = r.repost_type
+  AND s.txhash = r.txhash
+  AND r.rn > 1;
+
+UPDATE reposts SET repost_type = 'playlist' WHERE is_current = true AND repost_type = 'album';
+
+DO $$
+BEGIN
+    IF EXISTS (SELECT 1 FROM pg_trigger
+               WHERE tgrelid = 'reposts'::regclass AND tgname = 'on_repost' AND NOT tgisinternal) THEN
+        ALTER TABLE reposts ENABLE TRIGGER on_repost;
+    END IF;
+END $$;
+
+COMMIT;

ddl/migrations/0237_users_one_current_row_backfill.sql

@@ -0,0 +1,77 @@
+-- Remove duplicate is_current rows from users.
+--
+-- users_pkey is (user_id, txhash), which admits a second is_current row under a
+-- different txhash. Five users have one, spread from 2025-06 to 2026-06 — a
+-- slow drip, not a one-off. Each pair shares created_at and sits a few blocks
+-- apart.
+--
+-- How is not established. Both indexer create paths reject an existing user
+-- (validateUserCreate and migratedUserCreateHandler both call userExists, which
+-- tests is_current), so a single writer cannot produce these: its check and its
+-- insert share a transaction. A second writer can, since check-then-act is not
+-- atomic across transactions. Three of the five pairs put a bare-hex txhash
+-- next to a 0x-prefixed one, which fits that reading without proving it.
+--
+-- Five rows, but the blast radius is not five rows: anything joining an entity
+-- to its owner's wallet fans out and silently duplicates every entity those
+-- users own — measured at 18 extra tracks and 787 extra follows on a clone.
+--
+-- This is the backfill half of pkg/etl migration 0035, which adds
+-- users_current_uniq_idx to stop it recurring. That index cannot be created
+-- while duplicates exist, so this must run first. It does: ddl migrations run
+-- in the pre-roll `migrate` Job that every serving Deployment depends on, while
+-- the ETL's run at indexer start, after the Job has completed. The delete lives
+-- here rather than in the ETL migration so that bumping a Go module can never
+-- silently remove rows from this database.
+--
+-- Deleted rather than demoted: users keeps no versioned history — the indexer
+-- writes it in place, and production has zero is_current = false rows — so
+-- demoting would leave a category of row nothing reads. Deleting is safe here:
+-- no foreign key references users, and its triggers are INSERT (on_user) or
+-- INSERT OR UPDATE (trg_users), so neither fires.
+--
+-- Winner is the highest blocknumber, matching how consumers already pick the
+-- live row. Verified against a production clone: in all five cases blocknumber
+-- and updated_at agree, and for user 666149592 it keeps is_deactivated = true,
+-- the later of that pair's two states. Re-running is a no-op.
+
+BEGIN;
+SET LOCAL lock_timeout = '5s';
+
+-- Restricted to the offending user_ids so row_number() is computed over a
+-- handful of rows rather than all 3.15M current ones. Identifying them still
+-- costs a scan of users: this runs before users_current_uniq_idx exists (that
+-- is pkg/etl 0035, which runs later, at indexer start) and no other index
+-- covers is_current. A few seconds in a one-time pre-roll migration, and it
+-- takes only RowExclusiveLock, so concurrent DML is not blocked.
+--
+-- txhash is compared under the C collation so the tiebreak cannot vary with
+-- database collation. It never fires in practice — blocknumber decides every
+-- real case — but should be deterministic if it ever does.
+WITH dupes AS MATERIALIZED (
+    SELECT user_id
+    FROM users
+    WHERE is_current = true
+    GROUP BY user_id
+    HAVING count(*) > 1
+),
+ranked AS (
+    SELECT
+        user_id,
+        txhash,
+        row_number() OVER (
+            PARTITION BY user_id
+            ORDER BY blocknumber DESC NULLS LAST, updated_at DESC NULLS LAST,
+                     txhash COLLATE "C" DESC
+        ) AS rn
+    FROM users
+    WHERE is_current = true
+      AND user_id IN (SELECT user_id FROM dupes)
+)
+DELETE FROM users u
+USING ranked r
+WHERE u.user_id = r.user_id
+  AND u.txhash = r.txhash
+  AND r.rn > 1;
+
+COMMIT;

go.mod

@@ -7,7 +7,7 @@ require (
 	github.com/AlecAivazis/survey/v2 v2.3.7
 	github.com/Doist/unfurlist v0.0.0-20250409100812-515f2735f8e5
 	github.com/OpenAudio/go-openaudio v1.8.2-0.20260727214803-1d9f69772e87
-	github.com/OpenAudio/go-openaudio/pkg/etl v1.6.3-0.20260727214803-1d9f69772e87
+	github.com/OpenAudio/go-openaudio/pkg/etl v1.6.4
 	github.com/aquasecurity/esquery v0.2.0
 	github.com/axiomhq/axiom-go v0.23.0
 	github.com/axiomhq/hyperloglog v0.2.5
@@ -45,6 +45,7 @@ require (
 	github.com/urfave/cli/v3 v3.5.0
 	go.uber.org/zap v1.27.0
 	golang.org/x/crypto v0.48.0
+	golang.org/x/net v0.50.0
 	golang.org/x/sync v0.19.0
 	google.golang.org/grpc v1.71.1
 	google.golang.org/protobuf v1.36.11
@@ -219,7 +220,6 @@ require (
 	go.uber.org/multierr v1.11.0 // indirect
 	go.uber.org/ratelimit v0.2.0 // indirect
 	golang.org/x/exp v0.0.0-20250408133849-7e4ce0ab07d0 // indirect
-	golang.org/x/net v0.50.0 // indirect
 	golang.org/x/sys v0.41.0 // indirect
 	golang.org/x/term v0.40.0 // indirect
 	golang.org/x/text v0.34.0 // indirect

go.sum

@@ -22,8 +22,8 @@ github.com/Nvveen/Gotty v0.0.0-20120604004816-cd527374f1e5 h1:TngWCqHvy9oXAN6lEV
 github.com/Nvveen/Gotty v0.0.0-20120604004816-cd527374f1e5/go.mod h1:lmUJ/7eu/Q8D7ML55dXQrVaamCz2vxCfdQBasLZfHKk=
 github.com/OpenAudio/go-openaudio v1.8.2-0.20260727214803-1d9f69772e87 h1:HiLw4qrUkVWRADJjGsdHLlFMdJjhU5WCVNAfeKfww4s=
 github.com/OpenAudio/go-openaudio v1.8.2-0.20260727214803-1d9f69772e87/go.mod h1:lLRvUF5oWkxOyZx8rp/ecqxuMo3yzPvuvJbLSfvxguQ=
-github.com/OpenAudio/go-openaudio/pkg/etl v1.6.3-0.20260727214803-1d9f69772e87 h1:E3HoYAxTNrZ0x1H+PJ+sijy1yh3jMtnO6JTyZqWHJ9U=
-github.com/OpenAudio/go-openaudio/pkg/etl v1.6.3-0.20260727214803-1d9f69772e87/go.mod h1:6MIJhF06djJvPl6sfv3Vqp0f4IsyBsWPB4F+BsIamTc=
+github.com/OpenAudio/go-openaudio/pkg/etl v1.6.4 h1:SkKNhfvPWOlEGBw6i1LvAF1iNA0Gdf5rf/9izr2Kbro=
+github.com/OpenAudio/go-openaudio/pkg/etl v1.6.4/go.mod h1:z7X/5RziXEpASGTz7tD+P3pg4W5iCA1cKW7ogw8GShY=
 github.com/StackExchange/wmi v1.2.1 h1:VIkavFPXSjcnS+O8yTq7NI32k0R5Aj+v39y29VYDOSA=
 github.com/StackExchange/wmi v1.2.1/go.mod h1:rcmrprowKIVzvc+NUiLncP2uuArMWLCbu9SBzvHz7e8=
 github.com/VictoriaMetrics/fastcache v1.12.2 h1:N0y9ASrJ0F6h0QaC3o6uJb3NIZ9VKLjCM7NQbSmF7WI=

indexer/indexer.go

@@ -53,12 +53,27 @@ func NewIndexer(cfg config.Config) *CoreIndexer {
 	aggregatesCalculator := NewAggregatesCalculator(cfg)
 
 	// ETL needs the Connect/gRPC Core client (for block fetching) and a DB URL.
-	// SkipMigrations stays false (default): ETL's migrations are idempotent
-	// against api/'s schema — every migration uses CREATE TABLE IF NOT EXISTS /
-	// ADD COLUMN IF NOT EXISTS, and tracks state in its own `etl_db_migrations`
-	// table separate from api/'s `schema_version`. Verified by applying all 21
-	// current ETL migrations to a fresh DB seeded with api/'s schema: zero
-	// errors, only NOTICE messages for already-existing relations.
+	// SkipMigrations stays false (default), so bumping the pkg/etl module runs
+	// whatever migrations it brought with it against api/'s database, on the
+	// next indexer start. They track state in their own `etl_db_migrations`
+	// table, separate from api/'s `schema_version`, and only up-migrations run
+	// (RunDownMigrations is left false).
+	//
+	// None of them touches row data, which is the line being held: a module bump
+	// reaches this database automatically, so anything that repairs data belongs
+	// in ddl/, where it goes through review and the pre-roll migrate Job. 0035
+	// is the worked example — it adds users_current_uniq_idx, while the delete
+	// that makes that index creatable lives in ddl 0237.
+	//
+	// They are not, however, purely additive: 0026 drops a constraint and 0027
+	// drops and recreates playlist_seen's primary key. A bump can alter the
+	// schema here, so read what a version brings before taking it.
+	//
+	// The corollary is that an ETL migration can depend on a ddl one having
+	// run. 0035 fails with a unique violation if ddl 0237 has not removed the
+	// duplicates yet, which would stop the indexer starting. That ordering
+	// holds because ddl migrations run in the pre-roll Job that every serving
+	// Deployment depends on, and the ETL's run later, at indexer start.
 	//
 	// Two optional ETL components are disabled here because they don't fit
 	// api/'s deployment:

main.go

@@ -15,7 +15,9 @@ import (
 	"api.audius.co/esindexer"
 	eth_indexer "api.audius.co/eth/indexer"
 	core_indexer "api.audius.co/indexer"
+	"api.audius.co/logging"
 	solana_indexer "api.audius.co/solana/indexer"
+	etldb "github.com/OpenAudio/go-openaudio/pkg/etl/db"
 )
 
 func main() {
@@ -108,7 +110,29 @@ func main() {
 		}
 	case "migrate":
 		{
-			// no-op, handled prior to switch/case
+			// ddl migrations already ran above. Apply the ETL module's too, so a
+			// database migrated by this command matches a deployed one.
+			//
+			// They are otherwise only ever applied at indexer start, which left
+			// `make test-schema` unable to produce them: that target seeds from
+			// sql/01_schema.sql, runs this command, and dumps the result, so
+			// ETL-created objects could never enter the loop no matter how often
+			// it was regenerated. The four partial unique indexes from pkg/etl
+			// 0030 were consequently absent under test while present in every
+			// deployment — which is how ddl 0236 came to be written against the
+			// wrong constraint and still passed CI.
+			//
+			// Ordering is the useful side effect: ddl migrations run before these,
+			// in one process, so a ddl migration that has to precede an ETL one
+			// (0237 before 0035) is sequenced here rather than across two pods.
+			//
+			// Idempotent and tracked separately in etl_db_migrations, so the
+			// indexer applying them again at startup is a no-op.
+			logger := logging.NewZapLogger(config.Cfg).Named("migrate")
+			if err := etldb.RunMigrations(logger, config.Cfg.WriteDbUrl, false); err != nil {
+				fmt.Println("etl migration failed:", err)
+				os.Exit(1)
+			}
 			os.Exit(0)
 		}
 	default: