diff --git a/pkg/etl/db/sql/migrations/0005_notification_tables.up.sql b/pkg/etl/db/sql/migrations/0005_notification_tables.up.sql index 01cbd02f..ce55cf1a 100644 --- a/pkg/etl/db/sql/migrations/0005_notification_tables.up.sql +++ b/pkg/etl/db/sql/migrations/0005_notification_tables.up.sql @@ -26,6 +26,13 @@ CREATE TABLE IF NOT EXISTS notification_seen ( CONSTRAINT notification_seen_pkey PRIMARY KEY (user_id, seen_at) ); +-- playlist_seen PK matches prod's schema: (user_id, playlist_id, seen_at). +-- The `is_current` column is retained for backward compatibility with the +-- legacy schema (always written as `true` by the playlist-seen handler), but +-- is NOT part of the uniqueness contract — including it in the PK would +-- mismatch prod and cause 42P10 on any ON CONFLICT clause that names the +-- prod target columns. Older databases that were created with the wrong PK +-- get fixed by migration 0027_fix_playlist_seen_pkey.up.sql. CREATE TABLE IF NOT EXISTS playlist_seen ( is_current boolean NOT NULL, user_id integer NOT NULL, @@ -34,5 +41,5 @@ CREATE TABLE IF NOT EXISTS playlist_seen ( blocknumber integer, blockhash character varying, txhash character varying, - CONSTRAINT playlist_seen_pkey PRIMARY KEY (is_current, user_id, playlist_id, seen_at) + CONSTRAINT playlist_seen_pkey PRIMARY KEY (user_id, playlist_id, seen_at) ); diff --git a/pkg/etl/db/sql/migrations/0027_fix_playlist_seen_pkey.down.sql b/pkg/etl/db/sql/migrations/0027_fix_playlist_seen_pkey.down.sql new file mode 100644 index 00000000..1bed2485 --- /dev/null +++ b/pkg/etl/db/sql/migrations/0027_fix_playlist_seen_pkey.down.sql @@ -0,0 +1,3 @@ +-- Reverting to the broken (is_current,...) PK is intentionally a no-op — +-- it would re-introduce the 42P10 incompatibility with prod and break the +-- ON CONFLICT in the playlist_seen handler. Leave the down side empty. diff --git a/pkg/etl/db/sql/migrations/0027_fix_playlist_seen_pkey.up.sql b/pkg/etl/db/sql/migrations/0027_fix_playlist_seen_pkey.up.sql new file mode 100644 index 00000000..ba4b6049 --- /dev/null +++ b/pkg/etl/db/sql/migrations/0027_fix_playlist_seen_pkey.up.sql @@ -0,0 +1,17 @@ +-- playlist_seen primary key was originally declared as +-- (is_current, user_id, playlist_id, seen_at) — `is_current` was a column +-- but never a meaningful part of the uniqueness contract (the handler +-- always writes is_current=true). Prod's schema has the PK without +-- is_current. ON CONFLICT clauses that name the prod target — which is +-- the right thing to do — get rejected with 42P10 against the older +-- in-house PK. +-- +-- Fresh databases skip this migration via the corrected 0005. This +-- migration converts older databases (that already ran the bad 0005) to +-- the prod-compatible shape. + +ALTER TABLE playlist_seen + DROP CONSTRAINT IF EXISTS playlist_seen_pkey; + +ALTER TABLE playlist_seen + ADD CONSTRAINT playlist_seen_pkey PRIMARY KEY (user_id, playlist_id, seen_at); diff --git a/pkg/etl/indexer.go b/pkg/etl/indexer.go index 312ae3bb..ef96e2b8 100644 --- a/pkg/etl/indexer.go +++ b/pkg/etl/indexer.go @@ -422,7 +422,18 @@ func (e *Indexer) indexBlocks() error { zap.String("hash", tx.Hash), ) } else { - e.logger.Error("entity manager dispatch error", zap.Error(dErr)) + // Include entity_type / action / hash so we can attribute + // non-validation handler errors. Without this context the + // log line is unreadable (e.g. a bare "no rows in result + // set" with no clue which handler emitted it). + e.logger.Error("entity manager dispatch error", + zap.String("entity_type", me.GetEntityType()), + zap.String("action", me.GetAction()), + zap.Int64("entity_id", me.GetEntityId()), + zap.Int64("user_id", me.GetUserId()), + zap.String("hash", tx.Hash), + zap.Error(dErr), + ) } } else { e.logger.Debug("tx indexed", diff --git a/pkg/etl/processors/entity_manager/notification.go b/pkg/etl/processors/entity_manager/notification.go index 41020a19..3e69d951 100644 --- a/pkg/etl/processors/entity_manager/notification.go +++ b/pkg/etl/processors/entity_manager/notification.go @@ -93,11 +93,16 @@ func (h *notificationViewPlaylistHandler) Handle(ctx context.Context, params *Pa return NewValidationError("playlist %d does not exist", params.EntityID) } + // PK on prod is (user_id, playlist_id, seen_at) — `is_current` is not + // part of the unique index, so it must not appear in the ON CONFLICT + // target (Postgres 42P10 otherwise). Observed firing during the + // prod-clone validation run: same user seeing the same playlist twice + // in the indexed window. _, err = params.DBTX.Exec(ctx, ` INSERT INTO playlist_seen ( is_current, user_id, playlist_id, seen_at, txhash, blocknumber ) VALUES (true, $1, $2, $3, $4, $5) - ON CONFLICT (is_current, user_id, playlist_id, seen_at) DO NOTHING + ON CONFLICT (user_id, playlist_id, seen_at) DO NOTHING `, params.UserID, params.EntityID, params.BlockTime, params.TxHash, params.BlockNumber) return err } diff --git a/pkg/etl/processors/entity_manager/playlist_row.go b/pkg/etl/processors/entity_manager/playlist_row.go index 68900ac4..264baa50 100644 --- a/pkg/etl/processors/entity_manager/playlist_row.go +++ b/pkg/etl/processors/entity_manager/playlist_row.go @@ -4,9 +4,11 @@ import ( "context" "database/sql" "encoding/json" + "errors" "time" "github.com/OpenAudio/go-openaudio/pkg/etl/db" + "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgtype" ) @@ -59,6 +61,11 @@ func loadCurrentPlaylistRow(ctx context.Context, dbtx db.DBTX, playlistID int64) &upc, &ddex, &ddexRel, &artists, ©right, &producerCopyright, &parentalWarning, &r.CreatedAt, ) + if errors.Is(err, pgx.ErrNoRows) { + // Same race as loadCurrentTrackRow: validation accepts is_delete=true + // rows but this loader filters them out. Convert to ValidationError. + return nil, NewValidationError("playlist %d is deleted or does not exist", playlistID) + } if err != nil { return nil, err } diff --git a/pkg/etl/processors/entity_manager/track_row.go b/pkg/etl/processors/entity_manager/track_row.go index cb169123..f00f55ea 100644 --- a/pkg/etl/processors/entity_manager/track_row.go +++ b/pkg/etl/processors/entity_manager/track_row.go @@ -4,9 +4,11 @@ import ( "context" "database/sql" "encoding/json" + "errors" "time" "github.com/OpenAudio/go-openaudio/pkg/etl/db" + "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgtype" ) @@ -68,6 +70,13 @@ func loadCurrentTrackRow(ctx context.Context, dbtx db.DBTX, trackID int64) (*tra &releaseDate, &r.IsScheduledRelease, &aiAttr, &r.IsPlaylistUpload, &ddex, &ddexRel, &r.IsAvailable, &segments, &r.CreatedAt, ) + if errors.Is(err, pgx.ErrNoRows) { + // Track was soft-deleted between validation and now (validateTrackUpdate + // uses trackExists which accepts is_delete=true; this function filters + // is_delete=false). Convert raw ErrNoRows to a ValidationError so the + // dispatcher logs it at WARN, not ERROR. + return nil, NewValidationError("track %d is deleted or does not exist", trackID) + } if err != nil { return nil, err } diff --git a/pkg/etl/processors/entity_manager/user_update.go b/pkg/etl/processors/entity_manager/user_update.go index d210e095..f0f71cfd 100644 --- a/pkg/etl/processors/entity_manager/user_update.go +++ b/pkg/etl/processors/entity_manager/user_update.go @@ -4,10 +4,12 @@ import ( "context" "database/sql" "encoding/json" + "errors" "strings" "time" "github.com/OpenAudio/go-openaudio/pkg/etl/db" + "github.com/jackc/pgx/v5" ) type userUpdateHandler struct{} @@ -258,6 +260,13 @@ func getCurrentUser(ctx context.Context, dbtx db.DBTX, userID int64) (*currentUs func getUserHandle(ctx context.Context, dbtx db.DBTX, userID int64) (string, error) { var handleLC sql.NullString err := dbtx.QueryRow(ctx, "SELECT handle_lc FROM users WHERE user_id = $1 AND is_current = true LIMIT 1", userID).Scan(&handleLC) + if errors.Is(err, pgx.ErrNoRows) { + // User has no current row (deleted between validation and now, or + // never existed under is_current=true). Treat as empty handle — + // callers compare against the new handle, so empty here means + // "different, run the uniqueness check". + return "", nil + } if handleLC.Valid { return handleLC.String, err }