package store import ( "context" "database/sql" "fmt" "strings" "time" "coda/internal/models" "github.com/google/uuid" ) type ListJobsParams struct { Status *models.JobStatus Limit int Offset int } type JobRepository struct { db *sql.DB } func (r *JobRepository) Create(ctx context.Context, job *models.Job) error { if job.ID == "" { job.ID = uuid.New().String() } if job.CreatedAt.IsZero() { job.CreatedAt = time.Now().UTC() } _, err := r.db.ExecContext(ctx, `INSERT INTO jobs (id, artist, album, path, status, source, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)`, job.ID, job.Artist, job.Album, job.Path, job.Status, job.Source, job.CreatedAt, ) if err != nil { return fmt.Errorf("insert job: %w", err) } return nil } func (r *JobRepository) Get(ctx context.Context, id string) (*models.Job, error) { row := r.db.QueryRowContext(ctx, `SELECT id, artist, album, path, status, source, created_at, finished_at, duration_ms, error_msg, files_total, files_processed FROM jobs WHERE id = ?`, id) var job models.Job err := row.Scan(&job.ID, &job.Artist, &job.Album, &job.Path, &job.Status, &job.Source, &job.CreatedAt, &job.FinishedAt, &job.DurationMs, &job.ErrorMsg, &job.FilesTotal, &job.FilesProcessed) if err == sql.ErrNoRows { return nil, ErrNotFound } if err != nil { return nil, fmt.Errorf("get job: %w", err) } return &job, nil } func (r *JobRepository) List(ctx context.Context, params ListJobsParams) ([]models.Job, error) { query := `SELECT id, artist, album, path, status, source, created_at, finished_at, duration_ms, error_msg, files_total, files_processed FROM jobs` args := []any{} if params.Status != nil { query += ` WHERE status = ?` args = append(args, *params.Status) } query += ` ORDER BY created_at DESC` if params.Limit > 0 { query += ` LIMIT ?` args = append(args, params.Limit) } if params.Offset > 0 { query += ` OFFSET ?` args = append(args, params.Offset) } rows, err := r.db.QueryContext(ctx, query, args...) if err != nil { return nil, fmt.Errorf("list jobs: %w", err) } defer rows.Close() var jobs []models.Job for rows.Next() { var job models.Job if err := rows.Scan(&job.ID, &job.Artist, &job.Album, &job.Path, &job.Status, &job.Source, &job.CreatedAt, &job.FinishedAt, &job.DurationMs, &job.ErrorMsg, &job.FilesTotal, &job.FilesProcessed); err != nil { return nil, fmt.Errorf("scan job: %w", err) } jobs = append(jobs, job) } return jobs, rows.Err() } func (r *JobRepository) Count(ctx context.Context, status *models.JobStatus) (int, error) { query := `SELECT COUNT(*) FROM jobs` args := []any{} if status != nil { query += ` WHERE status = ?` args = append(args, *status) } var count int err := r.db.QueryRowContext(ctx, query, args...).Scan(&count) if err != nil { return 0, fmt.Errorf("count jobs: %w", err) } return count, nil } func (r *JobRepository) UpdateProgress(ctx context.Context, id string, filesProcessed, filesTotal int) error { result, err := r.db.ExecContext(ctx, `UPDATE jobs SET files_processed = ?, files_total = ? WHERE id = ?`, filesProcessed, filesTotal, id, ) if err != nil { return fmt.Errorf("update job progress: %w", err) } rows, err := result.RowsAffected() if err != nil { return fmt.Errorf("rows affected: %w", err) } if rows == 0 { return ErrNotFound } return nil } func (r *JobRepository) UpdateStatus(ctx context.Context, id string, status models.JobStatus) error { result, err := r.db.ExecContext(ctx, `UPDATE jobs SET status = ? WHERE id = ?`, status, id) if err != nil { return fmt.Errorf("update job status: %w", err) } rows, err := result.RowsAffected() if err != nil { return fmt.Errorf("rows affected: %w", err) } if rows == 0 { return ErrNotFound } return nil } func (r *JobRepository) Finish(ctx context.Context, id string, status models.JobStatus, durationMs int64, errMsg string) error { result, err := r.db.ExecContext(ctx, `UPDATE jobs SET status = ?, finished_at = ?, duration_ms = ?, error_msg = ? WHERE id = ?`, status, time.Now().UTC(), durationMs, errMsg, id, ) if err != nil { return fmt.Errorf("finish job: %w", err) } rows, err := result.RowsAffected() if err != nil { return fmt.Errorf("rows affected: %w", err) } if rows == 0 { return ErrNotFound } return nil } func (r *JobRepository) ExpireStuck(ctx context.Context, statuses []models.JobStatus, olderThan time.Duration) (int, error) { if len(statuses) == 0 { return 0, nil } placeholders := make([]string, len(statuses)) args := make([]any, 0, len(statuses)+1) for i, status := range statuses { placeholders[i] = "?" args = append(args, status) } args = append(args, time.Now().UTC().Add(-olderThan)) query := fmt.Sprintf( `UPDATE jobs SET status = ?, finished_at = ?, error_msg = ? WHERE status IN (%s) AND created_at < ?`, strings.Join(placeholders, ","), ) finalArgs := []any{models.StatusTimeout, time.Now().UTC(), "job timed out"} finalArgs = append(finalArgs, args...) result, err := r.db.ExecContext(ctx, query, finalArgs...) if err != nil { return 0, fmt.Errorf("expire stuck jobs: %w", err) } rows, err := result.RowsAffected() if err != nil { return 0, fmt.Errorf("rows affected: %w", err) } return int(rows), nil }