vidarchive

vidarchive/internal/repository/download.go
download.go
⎇
Raw
1package repository
2
3import (
4 "context"
5 "database/sql"
6 "fmt"
7
8 "vidarchive/internal/models"
9)
10
11// downloadColumns is the canonical select list, kept in the order scanDownload
12// expects so the two can't drift apart.
13const downloadColumns = `id, url, status, logs, error_message, preset_id, format_override,
14 custom_flags, output_dir, subscription_id, started_at, completed_at, created_at`
15
16type DownloadRepository struct {
17 db *sql.DB
18}
19
20func NewDownloadRepository(db *sql.DB) *DownloadRepository {
21 return &DownloadRepository{db: db}
22}
23
24func (r *DownloadRepository) Create(d *models.Download) error {
25 result, err := r.db.Exec(
26 `INSERT INTO downloads (url, status, preset_id, format_override, custom_flags, output_dir, subscription_id)
27 VALUES (?, ?, ?, ?, ?, ?, ?)`,
28 d.URL, d.Status, d.PresetID, d.FormatOverride, d.CustomFlags, d.OutputDir, d.SubscriptionID,
29 )
30 if err != nil {
31 return err
32 }
33 id, err := result.LastInsertId()
34 if err != nil {
35 return fmt.Errorf("download last insert id: %w", err)
36 }
37 d.ID = id
38 return nil
39}
40
41func (r *DownloadRepository) GetByID(id int64) (*models.Download, error) {
42 row := r.db.QueryRow(`SELECT `+downloadColumns+` FROM downloads WHERE id = ?`, id)
43 return scanDownload(row)
44}
45
46func (r *DownloadRepository) GetAll(status, sortBy string) ([]*models.Download, error) {
47 query := `SELECT ` + downloadColumns + ` FROM downloads WHERE 1=1`
48 var args []interface{}
49
50 if status != "" && status != "all" {
51 query += ` AND status = ?`
52 args = append(args, status)
53 }
54
55 switch sortBy {
56 case "status":
57 query += ` ORDER BY status, created_at DESC`
58 default:
59 query += ` ORDER BY created_at DESC`
60 }
61
62 return r.queryDownloads(query, args...)
63}
64
65func (r *DownloadRepository) GetQueued(limit int) ([]*models.Download, error) {
66 return r.queryDownloads(
67 `SELECT `+downloadColumns+` FROM downloads WHERE status = 'queued' ORDER BY created_at ASC LIMIT ?`,
68 limit,
69 )
70}
71
72func (r *DownloadRepository) Ping(ctx context.Context) error {
73 var one int
74 return r.db.QueryRowContext(ctx, `SELECT 1`).Scan(&one)
75}
76
77// IDsByStatus returns the ids of downloads in the given status.
78func (r *DownloadRepository) IDsByStatus(status string) ([]int64, error) {
79 rows, err := r.db.Query(`SELECT id FROM downloads WHERE status = ?`, status)
80 if err != nil {
81 return nil, err
82 }
83 defer rows.Close()
84
85 var ids []int64
86 for rows.Next() {
87 var id int64
88 if err := rows.Scan(&id); err != nil {
89 return nil, err
90 }
91 ids = append(ids, id)
92 }
93 return ids, rows.Err()
94}
95
96func (r *DownloadRepository) queryDownloads(query string, args ...interface{}) ([]*models.Download, error) {
97 rows, err := r.db.Query(query, args...)
98 if err != nil {
99 return nil, err
100 }
101 defer rows.Close()
102
103 var downloads []*models.Download
104 for rows.Next() {
105 d, err := scanDownload(rows)
106 if err != nil {
107 return nil, err
108 }
109 downloads = append(downloads, d)
110 }
111 return downloads, rows.Err()
112}
113
114func (r *DownloadRepository) AppendLogs(id int64, logs string) error {
115 _, err := r.db.Exec(
116 `UPDATE downloads SET logs = COALESCE(logs, '') || ? WHERE id = ?`,
117 logs, id,
118 )
119 return err
120}
121
122// MarkStarted atomically transitions a download from 'queued' to 'downloading'.
123// It reports whether this call actually claimed it: false means another worker
124// already started it, so the caller must not process it again.
125func (r *DownloadRepository) MarkStarted(id int64) (bool, error) {
126 res, err := r.db.Exec(
127 `UPDATE downloads SET status = 'downloading', started_at = CURRENT_TIMESTAMP WHERE id = ? AND status = 'queued'`,
128 id,
129 )
130 if err != nil {
131 return false, err
132 }
133 n, err := res.RowsAffected()
134 if err != nil {
135 return false, err
136 }
137 return n > 0, nil
138}
139
140func (r *DownloadRepository) MarkCompleted(id int64, status string) error {
141 _, err := r.db.Exec(
142 `UPDATE downloads SET status = ?, completed_at = CURRENT_TIMESTAMP WHERE id = ?`,
143 status, id,
144 )
145 return err
146}
147
148func (r *DownloadRepository) MarkError(id int64, errMsg string) error {
149 _, err := r.db.Exec(
150 `UPDATE downloads SET status = 'error', error_message = ? WHERE id = ?`,
151 errMsg, id,
152 )
153 return err
154}
155
156// HasActiveForSubscription reports whether the given subscription already has a
157// download that is queued or in progress, so the scheduler can avoid stacking a
158// second run on top of one that hasn't finished.
159func (r *DownloadRepository) HasActiveForSubscription(subID int64) (bool, error) {
160 var n int
161 err := r.db.QueryRow(
162 `SELECT COUNT(*) FROM downloads WHERE subscription_id = ? AND status IN ('queued', 'downloading')`,
163 subID,
164 ).Scan(&n)
165 if err != nil {
166 return false, err
167 }
168 return n > 0, nil
169}
170
171func (r *DownloadRepository) Delete(id int64) error {
172 _, err := r.db.Exec(`DELETE FROM downloads WHERE id = ?`, id)
173 return err
174}
175
176func (r *DownloadRepository) DeleteAll() error {
177 _, err := r.db.Exec(`DELETE FROM downloads`)
178 return err
179}
180
181func (r *DownloadRepository) UpdateStatusWhere(oldStatus, newStatus string) error {
182 _, err := r.db.Exec(
183 `UPDATE downloads SET status = ? WHERE status = ?`,
184 newStatus, oldStatus,
185 )
186 return err
187}
188
189func scanDownload(row interface{ Scan(...interface{}) error }) (*models.Download, error) {
190 var d models.Download
191 err := row.Scan(
192 &d.ID, &d.URL, &d.Status,
193 &d.Logs, &d.ErrorMessage, &d.PresetID, &d.FormatOverride, &d.CustomFlags, &d.OutputDir, &d.SubscriptionID,
194 &d.StartedAt, &d.CompletedAt, &d.CreatedAt,
195 )
196 if err != nil {
197 return nil, err
198 }
199 return &d, nil
200}
201