fix(etl): playlist_seen PK, ErrNoRows on soft-deletes, attribute dispatch errors

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)
 );

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.

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);

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",

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
 }

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, &copyright, &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
 	}

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
 	}

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
 	}