pipu-music/services/api/internal/uploads/release_store.go

package uploads

import (
	"context"
	"encoding/json"
	"errors"
	"time"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
	"pipu/api/internal/library/meta"
	"pipu/api/internal/musicbrainz"
	"pipu/api/internal/platform/apperr"
)

func (st *Store) GetMetadataCache(ctx context.Context, key string) ([]byte, bool, error) {
	var data []byte
	err := st.pool.QueryRow(ctx, `SELECT payload FROM metadata_lookup_cache WHERE cache_key=$1 AND expires_at>now()`, key).Scan(&data)
	if errors.Is(err, pgx.ErrNoRows) {
		return nil, false, nil
	}
	return data, err == nil, err
}
func (st *Store) PutMetadataCache(ctx context.Context, key string, data []byte, ttl time.Duration) error {
	_, err := st.pool.Exec(ctx, `INSERT INTO metadata_lookup_cache(cache_key,payload,expires_at) VALUES($1,$2,now()+$3*interval '1 second')
 ON CONFLICT(cache_key) DO UPDATE SET payload=EXCLUDED.payload,expires_at=EXCLUDED.expires_at`, key, data, ttl.Seconds())
	return err
}

func (st *Store) ExpireMetadataCache(ctx context.Context) error {
	_, err := st.pool.Exec(ctx, `DELETE FROM metadata_lookup_cache WHERE expires_at<now()`)
	return err
}
func (st *Store) ExistingReleaseTracks(ctx context.Context, tracks []musicbrainz.Track) ([]musicbrainz.Track, error) {
	out := []musicbrainz.Track{}
	if len(tracks) == 0 {
		return out, nil
	}
	artists, titles := []string{}, []string{}
	durations := []int32{}
	for _, t := range tracks {
		artists = append(artists, meta.Key(t.Artist))
		titles = append(titles, meta.Key(t.Title))
		durations = append(durations, int32(t.DurationMs))
	}
	rows, err := st.pool.Query(ctx, `SELECT DISTINCT e.n::int FROM unnest($1::text[],$2::text[],$3::int[]) WITH ORDINALITY e(artist,title,duration,n)
 JOIN tracks t ON t.title_norm=e.title AND abs(t.duration_ms-e.duration)<=2000
 JOIN track_artists ta ON ta.track_id=t.id JOIN artists a ON a.id=ta.artist_id AND a.name_norm=e.artist
 WHERE e.duration>0 AND EXISTS(SELECT 1 FROM media_files mf JOIN library_sources s ON s.id=mf.source_id
 WHERE mf.track_id=t.id AND mf.missing_since IS NULL AND mf.scan_error IS NULL AND NOT library_path_excluded(mf.rel_path,s.excluded_dirs))
 ORDER BY e.n::int`, artists, titles, durations)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	for rows.Next() {
		var i int
		if err = rows.Scan(&i); err != nil {
			return nil, err
		}
		out = append(out, tracks[i-1])
	}
	return out, rows.Err()
}
func (st *Store) AllInBatch(ctx context.Context, id uuid.UUID) ([]File, error) {
	rows, err := st.pool.Query(ctx, `SELECT `+fileColumns+fileFrom+` WHERE f.batch_id=$1 ORDER BY f.id`, id)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	out := []File{}
	for rows.Next() {
		f, err := scanFile(rows)
		if err != nil {
			return nil, err
		}
		out = append(out, f)
	}
	if err = rows.Err(); err != nil {
		return nil, err
	}
	if len(out) == 0 {
		return nil, ErrNotFound
	}
	return out, nil
}
func (st *Store) ReleaseReport(ctx context.Context, id uuid.UUID, scope ...string) (ReleaseReport, error) {
	r := ReleaseReport{State: "queued", Candidates: []ReleaseCandidate{}}
	key, err := st.resolveGroupKey(ctx, id, firstGroup(scope))
	if err != nil {
		return r, err
	}
	r.GroupID = key
	var raw []byte
	var audio []byte
	err = st.pool.QueryRow(ctx, `SELECT state,input_hash,candidates,selected_release::text,audio_results FROM upload_release_jobs WHERE batch_id=$1 AND group_key=$2`, id, key).Scan(&r.State, &r.InputHash, &raw, &r.SelectedRelease, &audio)
	if errors.Is(err, pgx.ErrNoRows) {
		return r, nil
	}
	if err != nil {
		return r, err
	}
	err = json.Unmarshal(raw, &r.Candidates)
	if err == nil {
		err = json.Unmarshal(audio, &r.AudioResults)
	}
	return r, err
}
func (st *Store) QueueRelease(ctx context.Context, id uuid.UUID, q musicbrainz.Query) error {
	key, err := st.resolveGroupKey(ctx, id, q.GroupID)
	if err != nil {
		return err
	}
	raw, _ := json.Marshal(q)
	_, err = st.pool.Exec(ctx, `INSERT INTO upload_release_jobs(batch_id,request,group_key) VALUES($1,$2,$3)
 ON CONFLICT(batch_id,group_key) DO UPDATE SET state='queued',request=EXCLUDED.request,audio_results='[]',lease=NULL,attempts=0,selected_release=NULL,updated_at=now()`, id, raw, key)
	return err
}

type releaseJob struct {
	ID, Lease uuid.UUID
	GroupKey  string
	Query     musicbrainz.Query
}

func (st *Store) ClaimRelease(ctx context.Context) (releaseJob, error) {
	j := releaseJob{Lease: newID()}
	// Empty-key jobs bootstrap memberships, then disappear. Actual lookups have independent leases.
	var batch uuid.UUID
	err := st.pool.QueryRow(ctx, `SELECT j.batch_id FROM upload_release_jobs j WHERE j.group_key='' AND j.state='queued'
 AND NOT EXISTS(SELECT 1 FROM upload_files f WHERE f.batch_id=j.batch_id AND f.status IN ('uploading','processing')) ORDER BY j.updated_at LIMIT 1`).Scan(&batch)
	if err == nil {
		if err = st.EnsureReleaseGroups(ctx, batch); err != nil {
			return j, err
		}
	} else if !errors.Is(err, pgx.ErrNoRows) {
		return j, err
	}
	var raw []byte
	err = st.pool.QueryRow(ctx, `UPDATE upload_release_jobs SET state='searching',lease=$1,attempts=attempts+1,updated_at=now()
 WHERE (batch_id,group_key)=(SELECT j.batch_id,j.group_key FROM upload_release_jobs j
 WHERE j.group_key<>'' AND (j.state='queued' OR j.state='error' AND j.attempts<3 AND j.updated_at<now()-interval '10 minutes'
 OR j.state='searching' AND j.updated_at<now()-interval '5 minutes')
 AND EXISTS(SELECT 1 FROM upload_files f WHERE f.batch_id=j.batch_id AND f.release_group_id::text=j.group_key AND f.status IN ('pending_review','import_failed'))
 AND NOT EXISTS(SELECT 1 FROM upload_files f WHERE f.batch_id=j.batch_id AND f.status IN ('uploading','processing'))
 ORDER BY j.updated_at FOR UPDATE SKIP LOCKED LIMIT 1) RETURNING batch_id,group_key,request`, j.Lease).Scan(&j.ID, &j.GroupKey, &raw)
	if errors.Is(err, pgx.ErrNoRows) {
		return j, ErrNotFound
	}
	if err != nil {
		return j, err
	}
	err = json.Unmarshal(raw, &j.Query)
	return j, err
}
func (st *Store) SaveRelease(ctx context.Context, j releaseJob, r ReleaseReport) error {
	raw, _ := json.Marshal(r.Candidates)
	audio := r.AudioResults
	if audio == nil {
		audio = []AudioRecognition{}
	}
	encoded, _ := json.Marshal(audio)
	_, err := st.pool.Exec(ctx, `UPDATE upload_release_jobs SET state=$3,input_hash=$4,candidates=$5,audio_results=$6,lease=NULL,updated_at=now()
 WHERE batch_id=$1 AND lease=$2 AND group_key=$7`, j.ID, j.Lease, r.State, r.InputHash, raw, encoded, j.GroupKey)
	return err
}

// ApplyRelease locks the complete batch and rejects stale suggestions. No network
// or filesystem IO occurs inside the transaction; decisions and quota stay intact.
func (st *Store) ApplyRelease(ctx context.Context, batch uuid.UUID, expected string, changes []releaseChange, cover *string, next ReleaseReport, reviewer uuid.UUID) error {
	key, err := st.resolveGroupKey(ctx, batch, next.GroupID)
	if err != nil {
		return err
	}
	tx, err := st.pool.Begin(ctx)
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback(ctx) }()
	var id uuid.UUID
	if err = tx.QueryRow(ctx, `SELECT id FROM upload_batches WHERE id=$1 FOR UPDATE`, batch).Scan(&id); err != nil {
		return err
	}
	rows, err := tx.Query(ctx, `SELECT `+fileColumns+fileFrom+` WHERE f.batch_id=$1 AND f.release_group_id::text=$2 ORDER BY f.id FOR UPDATE OF f`, batch, key)
	if err != nil {
		return err
	}
	current := []File{}
	editable := false
	for rows.Next() {
		f, e := scanFile(rows)
		if e != nil {
			rows.Close()
			return e
		}
		current = append(current, f)
		editable = editable || f.Status == StatusPendingReview || f.Status == StatusImportFailed
	}
	err = rows.Err()
	rows.Close()
	if err != nil {
		return err
	}
	if !editable {
		return ErrWrongStatus
	}
	if releaseHash(current) != expected {
		return apperr.Conflict("uploads.release_stale", "batch changed during review")
	}
	var state, hash string
	if err = tx.QueryRow(ctx, `SELECT state,input_hash FROM upload_release_jobs WHERE batch_id=$1 AND group_key=$2 FOR UPDATE`, batch, key).Scan(&state, &hash); err != nil {
		return err
	}
	if state != "ready" || hash != expected {
		return apperr.Conflict("uploads.release_stale", "release lookup changed during review")
	}
	for _, c := range changes {
		metadata, _ := json.Marshal(c.Meta)
		tags, _ := json.Marshal(c.Tags)
		result, e := tx.Exec(ctx, `UPDATE upload_files SET metadata=$2,tags=$3,metadata_edited=true,target_source_id=$4,target_rel_path=$5,
   status_reason=$6,matched_artist_id=$7,matched_album_id=$8,updated_at=now() WHERE id=$1 AND release_group_id::text=$9 AND status IN ('pending_review','import_failed')`,
			c.File.ID, metadata, tags, c.Placement.SourceID, c.Placement.RelPath, nullable(c.Placement.Reason), c.Placement.ArtistID, c.Placement.AlbumID, key)
		if e != nil {
			return e
		}
		if result.RowsAffected() != 1 {
			return ErrWrongStatus
		}
	}
	if cover != nil {
		if _, err = tx.Exec(ctx, `UPDATE upload_release_groups SET cover_path=$2 WHERE batch_id=$1 AND id::text=$3`, batch, *cover, key); err != nil {
			return err
		}
	}
	if len(changes) > 0 {
		m := changes[0].Meta
		if _, err = tx.Exec(ctx, `UPDATE upload_release_groups SET artist=$3,album=$4 WHERE batch_id=$1 AND id::text=$2`, batch, key, m.AlbumArtist, m.Album); err != nil {
			return err
		}
	}
	raw, _ := json.Marshal(next.Candidates)
	audio := next.AudioResults
	if audio == nil {
		audio = []AudioRecognition{}
	}
	encoded, _ := json.Marshal(audio)
	if _, err = tx.Exec(ctx, `UPDATE upload_release_jobs SET state='ready',input_hash=$2,candidates=$3,selected_release=$4,audio_results=$5,lease=NULL,updated_at=now() WHERE batch_id=$1 AND group_key=$6`, batch, next.InputHash, raw, next.SelectedRelease, encoded, key); err != nil {
		return err
	}
	if err = tx.Commit(ctx); err != nil {
		return err
	}
	st.Audit(ctx, &reviewer, "upload.release.apply", batch, map[string]any{"group": key, "release": next.SelectedRelease, "files": len(changes), "cover": cover != nil})
	return nil
}