feat(etl): carry playlist removal history through the migration

cmd/genesis-writer/entities_playlist.go

@@ -8,6 +8,7 @@ import (
 
 	corev1 "github.com/OpenAudio/go-openaudio/pkg/api/core/v1"
 	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
 )
 
 type playlistMetadataWrapper struct {
@@ -39,11 +40,58 @@ type playlistMetadataInner struct {
 	RouteSlug        string `json:"route_slug,omitempty"`
 	RouteTitleSlug   string `json:"route_title_slug,omitempty"`
 	RouteCollisionID int    `json:"route_collision_id,omitempty"`
+	// The playlist's removal history: tracks that were once in it and have
+	// since left, one entry per source playlist_tracks row with
+	// is_removed = true. Carried rather than derived, because a Create replays
+	// a snapshot and the indexer only recognizes a removal as a transition
+	// between two states — see insertPlaylistTrackTombstones. Empty for the
+	// 308,635 of 312,552 playlists that never lost a track.
+	RemovedTracks []removedPlaylistTrack `json:"removed_tracks,omitempty"`
 	// Always serialized: `omitempty` would drop a false value and the indexer
 	// cannot tell "absent" from "false".
 	IsDelete bool `json:"is_delete"`
 }
 
+// removedPlaylistTrack is one tombstone in a playlist's removal history.
+// created_at is when the track joined the playlist, updated_at when it left —
+// the latter is the timestamp the API compares against a purchase date to
+// decide whether an album buyer still has access to a track that has since left
+// the album, so it is carried verbatim rather than stamped at replay time.
+type removedPlaylistTrack struct {
+	TrackID   int64  `json:"track_id"`
+	CreatedAt string `json:"created_at"`
+	UpdatedAt string `json:"updated_at"`
+}
+
+// preloadRemovedPlaylistTracks loads every tombstone in the source, keyed by
+// playlist. Small enough to hold in memory: 24,579 rows over 4,064 playlists on
+// a production clone.
+func preloadRemovedPlaylistTracks(ctx context.Context, db *pgxpool.Pool) (map[int64][]removedPlaylistTrack, error) {
+	rows, err := db.Query(ctx, `
+		SELECT playlist_id, track_id, created_at, updated_at
+		FROM playlist_tracks
+		WHERE is_removed = true
+		ORDER BY playlist_id, updated_at, track_id`)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	m := make(map[int64][]removedPlaylistTrack)
+	for rows.Next() {
+		var playlistID int64
+		var r removedPlaylistTrack
+		var createdAt, updatedAt time.Time
+		if err := rows.Scan(&playlistID, &r.TrackID, &createdAt, &updatedAt); err != nil {
+			return nil, err
+		}
+		r.CreatedAt = createdAt.UTC().Format(time.RFC3339)
+		r.UpdatedAt = updatedAt.UTC().Format(time.RFC3339)
+		m[playlistID] = append(m[playlistID], r)
+	}
+	return m, rows.Err()
+}
+
 type sourcePlaylist struct {
 	PlaylistID          int64
 	PlaylistOwnerID     int64
@@ -74,6 +122,14 @@ type sourcePlaylist struct {
 }
 
 func (w *Writer) writePlaylists(ctx context.Context) error {
+	// Pre-load the removal history. It lives in playlist_tracks, not in the
+	// playlist row, and a Create carrying only the final contents would lose it
+	// entirely.
+	removedTracks, err := preloadRemovedPlaylistTracks(ctx, w.srcDB)
+	if err != nil {
+		return fmt.Errorf("preload removed playlist tracks: %w", err)
+	}
+
 	return processBatched(ctx, w, "playlists",
 		// Deleted playlists are migrated too, carrying is_delete in the metadata, so
 		// a parity check can tell an intentional omission from real data loss.
@@ -136,6 +192,7 @@ func (w *Writer) writePlaylists(ctx context.Context) error {
 				RouteTitleSlug:         deref(p.RouteTitleSlug),
 				RouteCollisionID:       derefInt(p.RouteCollisionID),
 				IsDelete:               p.IsDelete,
+				RemovedTracks:          removedTracks[p.PlaylistID],
 			}
 
 			inner.PlaylistContents = unmarshalJSONB(p.PlaylistContents)

pkg/etl/processors/entity_manager/migration.go

@@ -188,11 +188,62 @@ func (h *migratedPlaylistCreateHandler) Handle(ctx context.Context, params *Para
 	}
 
 	return insertPlaylistAndRouteWithState(ctx, params, playlistState{
-		IsDelete: params.MetadataBoolOr("is_delete", false),
-		Route:    migratedPlaylistRoute(params),
+		IsDelete:      params.MetadataBoolOr("is_delete", false),
+		Route:         migratedPlaylistRoute(params),
+		RemovedTracks: migratedPlaylistRemovedTracks(params),
 	})
 }
 
+// migratedPlaylistRemovedTracks reads the removal history the writer carried
+// from the source's playlist_tracks rows with is_removed = true, under the
+// `removed_tracks` metadata key:
+//
+//	[{"track_id": 1, "created_at": "<rfc3339>", "updated_at": "<rfc3339>"}, ...]
+//
+// created_at is when the track joined the playlist, updated_at when it left.
+// Absent for a production create, and for the vast majority of migrated
+// playlists — 3,917 of 312,552 have any removal history at all.
+func migratedPlaylistRemovedTracks(params *Params) []removedPlaylistTrack {
+	raw, ok := params.MetadataJSON("removed_tracks")
+	if !ok {
+		return nil
+	}
+	entries, ok := raw.([]any)
+	if !ok {
+		return nil
+	}
+	out := make([]removedPlaylistTrack, 0, len(entries))
+	for _, entry := range entries {
+		obj, ok := entry.(map[string]any)
+		if !ok {
+			continue
+		}
+		trackID, ok := pickPlaylistTrackID(obj)
+		if !ok {
+			continue
+		}
+		removed := removedPlaylistTrack{TrackID: trackID}
+		if ts, ok := parseReleaseDate(stringField(obj, "updated_at")); ok {
+			removed.UpdatedAt = ts.Time
+		}
+		if ts, ok := parseReleaseDate(stringField(obj, "created_at")); ok {
+			removed.CreatedAt = ts.Time
+		}
+		out = append(out, removed)
+	}
+	if len(out) == 0 {
+		return nil
+	}
+	return out
+}
+
+// stringField reads a string out of a decoded JSON object, or "" if absent or
+// of another type.
+func stringField(obj map[string]any, key string) string {
+	s, _ := obj[key].(string)
+	return s
+}
+
 // migratedPlaylistRoute reads the route the writer carried from the source, or
 // returns nil so the slug is generated as it would be for a new playlist. See
 // migratedTrackRoute -- playlist slugs drifted the same way, and further: 60.7%

pkg/etl/processors/entity_manager/playlist_create.go

@@ -70,6 +70,13 @@ type playlistState struct {
 	// resolving entirely. A slug is a permanent URL, so the migration carries
 	// the real one rather than deriving a plausible one.
 	Route *playlistRoute
+
+	// RemovedTracks are the playlist's tombstones — tracks that were once in it
+	// and have since left. A Create can only produce live membership on its own,
+	// so the migration carries the removal history explicitly; see
+	// insertPlaylistTrackTombstones for why it cannot be derived and why it
+	// matters.
+	RemovedTracks []removedPlaylistTrack
 }
 
 // insertPlaylistAndRoute writes a newly created playlist. A new playlist is
@@ -128,9 +135,11 @@ func insertPlaylistAndRouteWithState(ctx context.Context, params *Params, state
 		CreatedAt:                   params.BlockTime,
 	}
 
+	trackIDs := extractPlaylistTrackIDs(params.Metadata)
+
 	// last_added_to is the block time of the most recent track add; set it when
 	// the playlist is created with tracks (matches the legacy indexer), else NULL.
-	if len(extractPlaylistTrackIDs(params.Metadata)) > 0 {
+	if len(trackIDs) > 0 {
 		row.LastAddedTo = pgTimestamp(params.BlockTime)
 	}
 
@@ -141,6 +150,12 @@ func insertPlaylistAndRouteWithState(ctx context.Context, params *Params, state
 	if err := updatePlaylistTracks(ctx, params.DBTX, params.EntityID, params.Metadata, params.BlockTime); err != nil {
 		return err
 	}
+	// Tombstones are written after the live rows so the live membership wins any
+	// contradiction between the two. Empty for a real create — only the
+	// migration carries removal history.
+	if err := insertPlaylistTrackTombstones(ctx, params.DBTX, params.EntityID, state.RemovedTracks, trackIDs, params.BlockTime); err != nil {
+		return err
+	}
 	if err := updateAlbumPriceHistory(ctx, params.DBTX, params.EntityID, params.BlockNumber, params.BlockTime, params.Metadata); err != nil {
 		return err
 	}

pkg/etl/processors/entity_manager/playlist_tombstones.go

@@ -0,0 +1,122 @@
+package entity_manager
+
+import (
+	"cmp"
+	"context"
+	"slices"
+	"time"
+
+	"github.com/OpenAudio/go-openaudio/pkg/etl/db"
+)
+
+// removedPlaylistTrack is a track that used to be in a playlist and is not in
+// it any more — a playlist_tracks row with is_removed = true. Because the
+// junction table's primary key is (playlist_id, track_id), a removal flips the
+// flag rather than deleting the row, so the row is a tombstone: the record that
+// the track was once a member, and when it left.
+//
+// CreatedAt is when the track joined; UpdatedAt is when it left.
+type removedPlaylistTrack struct {
+	TrackID   int64
+	CreatedAt time.Time
+	UpdatedAt time.Time
+}
+
+// insertPlaylistTrackTombstones writes the removal history a migrated playlist
+// carries on its Create.
+//
+// The tombstones cannot be derived the way production derives them.
+// updatePlaylistTracks recognizes a removal as a transition — a track that was
+// in playlist_contents before an update and is absent after it — and the
+// migration replays a snapshot: one Playlist/Create holding the playlist's
+// final contents, with no earlier state to differ from. The "was present, now
+// absent" loop has nothing to iterate, so a Create alone can only ever produce
+// live rows. On a production clone that is 20,763 tombstones across 3,917
+// playlists that would simply not exist in the migrated data.
+//
+// They are not bookkeeping. They are the input to
+// tracks.playlists_previously_containing_track, which the API reads to decide
+// whether someone who bought an album still has access to a track that later
+// left it — 19,441 tracks in the source have such a record. So each tombstone
+// updates the reverse index here too, through the same function the production
+// removal path uses, and carries the source's own timestamps: the removal time
+// is compared against a purchase date, so it has to be when the track actually
+// left, not when the migration replayed it.
+//
+// currentTrackIDs is the playlist's live membership. A track appearing in both
+// is a contradiction in the source; the live row wins and the tombstone is
+// dropped, so a replay can never mark a present track removed.
+func insertPlaylistTrackTombstones(
+	ctx context.Context,
+	dbtx db.DBTX,
+	playlistID int64,
+	removals []removedPlaylistTrack,
+	currentTrackIDs []int64,
+	blockTime time.Time,
+) error {
+	if len(removals) == 0 {
+		return nil
+	}
+
+	seen := make(map[int64]struct{}, len(currentTrackIDs)+len(removals))
+	for _, id := range currentTrackIDs {
+		seen[id] = struct{}{}
+	}
+
+	pending := make([]removedPlaylistTrack, 0, len(removals))
+	for _, r := range removals {
+		if _, dup := seen[r.TrackID]; dup {
+			continue
+		}
+		seen[r.TrackID] = struct{}{}
+		if r.UpdatedAt.IsZero() {
+			r.UpdatedAt = blockTime
+		}
+		if r.CreatedAt.IsZero() {
+			r.CreatedAt = r.UpdatedAt
+		}
+		pending = append(pending, r)
+	}
+	if len(pending) == 0 {
+		return nil
+	}
+
+	// Sorted by removal time so the reverse-index statements below batch into
+	// runs, and so a replay of the same input issues the same statements.
+	slices.SortFunc(pending, func(a, b removedPlaylistTrack) int {
+		if c := a.UpdatedAt.Compare(b.UpdatedAt); c != 0 {
+			return c
+		}
+		return cmp.Compare(a.TrackID, b.TrackID)
+	})
+
+	for _, r := range pending {
+		if _, err := dbtx.Exec(ctx, `
+			INSERT INTO playlist_tracks (playlist_id, track_id, is_removed, created_at, updated_at)
+			VALUES ($1, $2, true, $3, $4)
+			ON CONFLICT (playlist_id, track_id) DO NOTHING
+		`, playlistID, r.TrackID, r.CreatedAt, r.UpdatedAt); err != nil {
+			return err
+		}
+	}
+
+	// One reverse-index write per distinct removal time: the statement stamps a
+	// single timestamp across every track it touches, so tracks that left at
+	// different moments cannot share one.
+	for start := 0; start < len(pending); {
+		end := start + 1
+		for end < len(pending) && pending[end].UpdatedAt.Equal(pending[start].UpdatedAt) {
+			end++
+		}
+		ids := make([]int64, 0, end-start)
+		for _, r := range pending[start:end] {
+			ids = append(ids, r.TrackID)
+		}
+		if err := updateTrackPlaylistIndex(ctx, dbtx, playlistID, nil, ids, pending[start].UpdatedAt); err != nil {
+			return err
+		}
+		start = end
+	}
+
+	return nil
+}