yellowjacket/backend/queue/persistence.go

package queue

import (
	"database/sql"
	"encoding/json"
	"fmt"
	"slices"
	"strings"

	"yellowjacket/backend/coverart"
	"yellowjacket/backend/database/sql/sqlcgen"
	"yellowjacket/backend/profiling"
)

// persistAddTrack inserts a single track at the end of the queue.
// No position shifting is needed because this is always an append.
// The caller must hold q.mu.
func (q *Queue) persistAddTrack(track Track) {
	q.submitWrite(func() {
		_, err := q.db.Queries.InsertQueueTrack(q.db.Ctx, sqlcgen.InsertQueueTrackParams{
			AudioFileID: track.AudioFileID,
			Position:    track.Position,
		})
		if err != nil {
			q.logger.Error("Failed to persist added track", "err", err)
		}
	})
}

// persistAddTracks inserts multiple tracks at the end of the queue
// atomically in a transaction. No position shifting is needed because
// these are always appends.
// The caller must hold q.mu.
func (q *Queue) persistAddTracks(tracks []Track) {
	if len(tracks) == 0 {
		return
	}

	// Cloned because the caller's slice is usually a window onto
	// q.tracks, which mutates the moment the lock is released.
	snapshot := slices.Clone(tracks)

	q.submitWrite(func() { q.writeAddTracks(snapshot) })
}

func (q *Queue) writeAddTracks(tracks []Track) {
	tx, err := q.db.BeginTx()
	if err != nil {
		q.logger.Error("Failed to begin transaction", "err", err)

		return
	}

	committed := false

	defer func() {
		if !committed {
			if rbErr := tx.Rollback(); rbErr != nil {
				q.logger.Error(
					"Failed to rollback transaction",
					"err", rbErr,
				)
			}
		}
	}()

	txQueries := q.db.Queries.WithTx(tx)

	for _, track := range tracks {
		_, insertErr := txQueries.InsertQueueTrack(q.db.Ctx, sqlcgen.InsertQueueTrackParams{
			AudioFileID: track.AudioFileID,
			Position:    track.Position,
		})
		if insertErr != nil {
			q.logger.Error("Failed to insert track", "err", insertErr)

			return
		}
	}

	if commitErr := tx.Commit(); commitErr != nil {
		q.logger.Error("Failed to commit transaction", "err", commitErr)

		return
	}

	committed = true
}

// persistInsertTracks inserts multiple tracks at a given position,
// shifting existing tracks to make room. Uses a transaction for atomicity.
// The caller must hold q.mu.
func (q *Queue) persistInsertTracks(tracks []Track, insertPos int) {
	if len(tracks) == 0 {
		return
	}

	snapshot := slices.Clone(tracks)

	q.submitWrite(func() { q.writeInsertTracks(snapshot, insertPos) })
}

func (q *Queue) writeInsertTracks(tracks []Track, insertPos int) {
	tx, err := q.db.BeginTx()
	if err != nil {
		q.logger.Error("Failed to begin transaction", "err", err)

		return
	}

	committed := false

	defer func() {
		if !committed {
			if rbErr := tx.Rollback(); rbErr != nil {
				q.logger.Error(
					"Failed to rollback transaction",
					"err", rbErr,
				)
			}
		}
	}()

	// SAFETY: Multi-row position shift by variable N unsupported by sqlc
	// (ShiftQueuePositionsUp only shifts by 1). Bind variables match args;
	// no string interpolation.
	_, err = tx.ExecContext(
		q.db.Ctx,
		"UPDATE queue_tracks SET position = position + ? WHERE position >= ?",
		len(tracks), insertPos,
	)
	if err != nil {
		q.logger.Error("Failed to shift positions up", "err", err)

		return
	}

	txQueries := q.db.Queries.WithTx(tx)

	for i, track := range tracks {
		_, insertErr := txQueries.InsertQueueTrack(q.db.Ctx, sqlcgen.InsertQueueTrackParams{
			AudioFileID: track.AudioFileID,
			Position:    int64(insertPos + i),
		})
		if insertErr != nil {
			q.logger.Error("Failed to insert track", "err", insertErr)

			return
		}
	}

	if commitErr := tx.Commit(); commitErr != nil {
		q.logger.Error("Failed to commit transaction", "err", commitErr)

		return
	}

	committed = true
}

// persistRemoveTrack deletes a single track at the given position and
// shifts subsequent positions down to close the gap.
// The caller must hold q.mu.
func (q *Queue) persistRemoveTrack(position int) {
	q.submitWrite(func() { q.writeRemoveTrack(position) })
}

func (q *Queue) writeRemoveTrack(position int) {
	tx, err := q.db.BeginTx()
	if err != nil {
		q.logger.Error("Failed to begin transaction", "err", err)

		return
	}

	committed := false

	defer func() {
		if !committed {
			if rbErr := tx.Rollback(); rbErr != nil {
				q.logger.Error(
					"Failed to rollback transaction",
					"err", rbErr,
				)
			}
		}
	}()

	txQueries := q.db.Queries.WithTx(tx)

	if removeErr := txQueries.RemoveQueueTrackByPosition(
		q.db.Ctx, int64(position),
	); removeErr != nil {
		q.logger.Error("Failed to remove track by position", "err", removeErr)

		return
	}

	if shiftErr := txQueries.ShiftQueuePositionsDown(
		q.db.Ctx, int64(position),
	); shiftErr != nil {
		q.logger.Error("Failed to shift positions down", "err", shiftErr)

		return
	}

	if commitErr := tx.Commit(); commitErr != nil {
		q.logger.Error("Failed to commit transaction", "err", commitErr)

		return
	}

	committed = true
}

// lookupTrackMetaBatch fetches audio file IDs and metadata for a batch of
// file paths using a single query per chunk (instead of 2 queries per track).
// Returns a map keyed by file path. This is safe to call without holding q.mu.
func (q *Queue) lookupTrackMetaBatch(
	filePaths []string,
) map[string]trackMeta {
	result := make(map[string]trackMeta, len(filePaths))

	// Deduplicate paths to avoid redundant work.
	unique := make([]string, 0, len(filePaths))
	seen := make(map[string]bool, len(filePaths))

	for _, fp := range filePaths {
		if !seen[fp] {
			seen[fp] = true

			unique = append(unique, fp)
		}
	}

	// Process in chunks to stay under the SQLite bind variable limit.
	for i := 0; i < len(unique); i += maxSQLiteVars {
		end := i + maxSQLiteVars
		if end > len(unique) {
			end = len(unique)
		}

		chunk := unique[i:end]
		q.lookupChunk(chunk, result)
	}

	return result
}

// lookupChunk executes a single batch query for a chunk of file paths
// using the sqlc-generated LookupTrackMetaByPaths query against the
// track_metadata VIEW.
func (q *Queue) lookupChunk(
	paths []string,
	result map[string]trackMeta,
) {
	if len(paths) == 0 {
		return
	}

	rows, err := q.db.ReadQueries.LookupTrackMetaByPaths(q.db.Ctx, paths)
	if err != nil {
		q.logger.Error("Batch metadata lookup failed", "err", err)

		return
	}

	for _, row := range rows {
		result[row.FilePath] = trackMeta{
			AudioFileID:      row.ID,
			FilePath:         row.FilePath,
			Title:            row.Title,
			Artist:           row.ArtistName,
			Album:            row.Album,
			CoverArtPath:     row.CoverArtPath,
			ArtistMBID:       row.ArtistMbid,
			ReleaseGroupMBID: row.ReleaseGroupMbid,
			RecordingMBID:    row.RecordingMbid,
		}
	}
}

// persistTracks writes the current queue tracks to the database atomically
// using a transaction with batched multi-row inserts.
func (q *Queue) persistTracks() {
	snapshot := slices.Clone(q.tracks)

	q.submitWrite(func() { q.writeTracks(snapshot) })
}

func (q *Queue) writeTracks(tracks []Track) {
	tx, err := q.db.BeginTx()
	if err != nil {
		q.logger.Error("Failed to begin transaction", "err", err)

		return
	}

	committed := false

	defer func() {
		if !committed {
			if rbErr := tx.Rollback(); rbErr != nil {
				q.logger.Error(
					"Failed to rollback transaction",
					"err", rbErr,
				)
			}
		}
	}()

	// Clear existing tracks.
	txQueries := q.db.Queries.WithTx(tx)

	if clearErr := txQueries.ClearQueueTracks(q.db.Ctx); clearErr != nil {
		q.logger.Error("Failed to clear queue tracks", "err", clearErr)

		return
	}

	// Batch insert tracks. Each row needs 2 bind vars (audio_file_id, position).
	const varsPerRow = 2

	batchSize := maxSQLiteVars / varsPerRow

	for i := 0; i < len(tracks); i += batchSize {
		end := i + batchSize
		if end > len(tracks) {
			end = len(tracks)
		}

		batch := tracks[i:end]

		if insertErr := q.insertTrackBatch(tx, batch); insertErr != nil {
			q.logger.Error(
				"Failed to batch insert queue tracks",
				"err", insertErr,
			)

			return
		}
	}

	if commitErr := tx.Commit(); commitErr != nil {
		q.logger.Error("Failed to commit transaction", "err", commitErr)

		return
	}

	committed = true
}

// insertTrackBatch inserts a batch of tracks in a single multi-row INSERT.
func (q *Queue) insertTrackBatch(tx *sql.Tx, batch []Track) error {
	if len(batch) == 0 {
		return nil
	}

	valuePlaceholders := make([]string, len(batch))
	args := make([]any, 0, len(batch)*2)

	for i, track := range batch {
		valuePlaceholders[i] = "(?, ?)"

		args = append(args, track.AudioFileID, track.Position)
	}

	// SAFETY: Multi-row INSERT with variable row count unsupported by sqlc. Placeholder count matches args length; no string interpolation.
	query := "INSERT INTO queue_tracks (audio_file_id, position) VALUES " +
		strings.Join(valuePlaceholders, ",")

	_, err := tx.ExecContext(q.db.Ctx, query, args...)
	if err != nil {
		return fmt.Errorf("batch insert failed: %w", err)
	}

	return nil
}

// persistState writes the queue metadata to the database.
func (q *Queue) persistState() {
	var shuffleOrderJSON sql.NullString

	if len(q.shuffleOrder) > 0 {
		data, err := json.Marshal(q.shuffleOrder)
		if err != nil {
			q.logger.Error(
				"Failed to marshal shuffle order",
				"err", err,
			)
		} else {
			shuffleOrderJSON = sql.NullString{
				String: string(data),
				Valid:  true,
			}
		}
	}

	params := sqlcgen.UpdateQueueStateParams{
		CurrentPosition: int64(q.currentIndex),
		ShuffleMode:     q.shuffleMode,
		RepeatMode:      string(q.repeatMode),
		ShuffleOrder:    shuffleOrderJSON,
		SourceType:      q.source.Type,
		SourceID:        q.source.ID,
		SourceLabel:     q.source.Label,
	}

	q.submitWrite(func() {
		if err := q.db.Queries.UpdateQueueState(q.db.Ctx, params); err != nil {
			q.logger.Error("Failed to persist queue state", "err", err)
		}
	})
}

// SaveState persists the queue state to the database.  Unlike every
// other write here it waits for the writer: its callers are shutdown
// and the tests, both of which need the row to exist on return.
func (q *Queue) SaveState() {
	q.mu.Lock()
	defer q.mu.Unlock()

	q.persistTracks()
	q.persistState()
	q.flushWrites()
	q.logger.Info("Queue state saved",
		"trackCount", len(q.tracks),
		"currentIndex", q.currentIndex,
		"shuffleMode", q.shuffleMode,
		"repeatMode", q.repeatMode,
	)
}

// RestoreState loads the queue state from the database.
func (q *Queue) RestoreState() {
	defer profiling.TimeOp(q.logger, "queue.RestoreState")()

	q.mu.Lock()
	defer q.mu.Unlock()

	// The other read-back: at startup there is nothing pending, but a
	// restore after any mutation must see it.
	q.flushWrites()

	// Restore queue metadata.
	state, err := q.db.ReadQueries.GetQueueState(q.db.Ctx)
	if err != nil {
		q.logger.Error("Failed to load queue state", "err", err)

		return
	}

	q.currentIndex = int(state.CurrentPosition)
	q.shuffleMode = state.ShuffleMode
	q.repeatMode = RepeatMode(state.RepeatMode)

	q.source = Source{
		Type:  state.SourceType,
		ID:    state.SourceID,
		Label: state.SourceLabel,
	}

	// Restore shuffle order.
	if state.ShuffleOrder.Valid && state.ShuffleOrder.String != "" {
		var order []int

		if err := json.Unmarshal(
			[]byte(state.ShuffleOrder.String), &order,
		); err != nil {
			q.logger.Warn("Failed to parse shuffle order", "err", err)
		} else {
			q.shuffleOrder = order
		}
	}

	// Restore queue tracks.
	rows, err := q.db.ReadQueries.GetQueueTracks(q.db.Ctx)
	if err != nil {
		q.logger.Error("Failed to load queue tracks", "err", err)

		return
	}

	q.tracks = make([]Track, 0, len(rows))

	for _, row := range rows {
		var coverArtURL string
		if row.CoverArtPath != "" {
			coverArtURL = coverart.ResolveURLs(row.CoverArtPath).Small
		}

		q.tracks = append(q.tracks, Track{
			ID:               row.ID,
			AudioFileID:      row.AudioFileID,
			FilePath:         row.FilePath,
			Position:         row.Position,
			Title:            row.Title,
			Artist:           row.Artist,
			Album:            row.Album,
			CoverArtPath:     coverArtURL,
			ArtistMBID:       row.ArtistMbid,
			ReleaseGroupMBID: row.ReleaseGroupMbid,
			RecordingMBID:    row.RecordingMbid,
		})
	}

	// Clamp current index. A value of -1 is valid and means "no current
	// track" (e.g. the queue was exhausted before shutdown). Only clamp
	// when the index exceeds the restored track count.
	if q.currentIndex >= len(q.tracks) && len(q.tracks) > 0 {
		q.currentIndex = len(q.tracks) - 1
	}

	q.logger.Info("Queue state restored",
		"trackCount", len(q.tracks),
		"currentIndex", q.currentIndex,
		"shuffleMode", q.shuffleMode,
		"repeatMode", q.repeatMode,
	)
}