package repository import ( "context" "fmt" "github.com/jackc/pgx/v5" "github.com/prabhupadasays/spsays-bff/internal/model" ) // AudioProgressRepo wraps Postgres access for audio_progress — the audio // player's per-file resume points. One row per (user, audio_key). type AudioProgressRepo struct{ pg *Postgres } func NewAudioProgressRepo(pg *Postgres) *AudioProgressRepo { return &AudioProgressRepo{pg: pg} } // ListByUser returns all of the user's saved resume points, newest first. // Used to hydrate the player's in-memory map on sign-in. func (r *AudioProgressRepo) ListByUser(ctx context.Context, userID string) ([]model.AudioProgress, error) { rows, err := r.pg.Pool.Query(ctx, ` SELECT audio_key, position_seconds, duration_seconds, updated_at FROM audio_progress WHERE user_id = $1 ORDER BY updated_at DESC`, userID, ) if err != nil { return nil, fmt.Errorf("list audio progress: %w", err) } defer rows.Close() var out []model.AudioProgress for rows.Next() { var p model.AudioProgress if err := rows.Scan(&p.Key, &p.Position, &p.Duration, &p.UpdatedAt); err != nil { return nil, fmt.Errorf("scan audio progress: %w", err) } out = append(out, p) } return out, rows.Err() } // Upsert writes a batch of resume points. Each row is inserted or updated // only when the incoming updated_at is at least as new as the stored one // (newest-wins), so concurrent tabs/devices and the on-login sync of // anonymous IndexedDB entries never clobber a fresher position. func (r *AudioProgressRepo) Upsert(ctx context.Context, userID string, items []model.AudioProgress) error { if len(items) == 0 { return nil } batch := &pgx.Batch{} for _, it := range items { batch.Queue(` INSERT INTO audio_progress (user_id, audio_key, position_seconds, duration_seconds, updated_at) VALUES ($1, $2, $3, $4, $5) ON CONFLICT (user_id, audio_key) DO UPDATE SET position_seconds = EXCLUDED.position_seconds, duration_seconds = EXCLUDED.duration_seconds, updated_at = EXCLUDED.updated_at WHERE EXCLUDED.updated_at >= audio_progress.updated_at`, userID, it.Key, it.Position, it.Duration, it.UpdatedAt) } br := r.pg.Pool.SendBatch(ctx, batch) defer br.Close() for range items { if _, err := br.Exec(); err != nil { return fmt.Errorf("upsert audio progress: %w", err) } } return nil } // Delete removes a single resume point (called when a track plays to // completion so it restarts from the beginning next time). func (r *AudioProgressRepo) Delete(ctx context.Context, userID, key string) error { _, err := r.pg.Pool.Exec(ctx, `DELETE FROM audio_progress WHERE user_id = $1 AND audio_key = $2`, userID, key, ) if err != nil { return fmt.Errorf("delete audio progress: %w", err) } return nil }