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
46// GetAll lists one page of downloads. created_at is indexed, so the ORDER BY
47// doesn't sort the whole table to serve a page.
48func (r *DownloadRepository) GetAll(status, sortBy string, limit, offset int) ([]*models.Download, error) {
49 query := `SELECT ` + downloadColumns + ` FROM downloads WHERE 1=1`
50 var args []interface{}
51
52 if status != "" && status != "all" {
53 query += ` AND status = ?`
54 args = append(args, status)
55 }
56
57 switch sortBy {
58 case "status":
59 query += ` ORDER BY status, created_at DESC`
60 default:
61 query += ` ORDER BY created_at DESC`
62 }
63
64 query += ` LIMIT ? OFFSET ?`
65 args = append(args, limit, offset)
66
67 return r.queryDownloads(query, args...)
68}
69
70func (r *DownloadRepository) GetQueued(limit int) ([]*models.Download, error) {
71 return r.queryDownloads(
72 `SELECT `+downloadColumns+` FROM downloads WHERE status = 'queued' ORDER BY created_at ASC LIMIT ?`,
73 limit,
74 )
75}
76
77func (r *DownloadRepository) Ping(ctx context.Context) error {
78 var one int
79 return r.db.QueryRowContext(ctx, `SELECT 1`).Scan(&one)
80}
81
82// IDsByStatus returns the ids of downloads in the given status.
83func (r *DownloadRepository) IDsByStatus(status string) ([]int64, error) {
84 rows, err := r.db.Query(`SELECT id FROM downloads WHERE status = ?`, status)
85 if err != nil {
86 return nil, err
87 }
88 defer rows.Close()
89
90 var ids []int64
91 for rows.Next() {
92 var id int64
93 if err := rows.Scan(&id); err != nil {
94 return nil, err
95 }
96 ids = append(ids, id)
97 }
98 return ids, rows.Err()
99}
100
101func (r *DownloadRepository) queryDownloads(query string, args ...interface{}) ([]*models.Download, error) {
102 rows, err := r.db.Query(query, args...)
103 if err != nil {
104 return nil, err
105 }
106 defer rows.Close()
107
108 var downloads []*models.Download
109 for rows.Next() {
110 d, err := scanDownload(rows)
111 if err != nil {
112 return nil, err
113 }
114 downloads = append(downloads, d)
115 }
116 return downloads, rows.Err()
117}
118
119func (r *DownloadRepository) AppendLogs(id int64, logs string) error {
120 _, err := r.db.Exec(
121 `UPDATE downloads SET logs = COALESCE(logs, '') || ? WHERE id = ?`,
122 logs, id,
123 )
124 return err
125}
126
127// MarkStarted atomically transitions a download from 'queued' to 'downloading'.
128// It reports whether this call actually claimed it: false means another worker
129// already started it, so the caller must not process it again.
130func (r *DownloadRepository) MarkStarted(id int64) (bool, error) {
131 return r.affected(
132 `UPDATE downloads SET status = 'downloading', started_at = CURRENT_TIMESTAMP WHERE id = ? AND status = 'queued'`,
133 id,
134 )
135}
136
137// Requeue moves a finished-but-unsuccessful download back to 'queued' so the
138// queue checker picks it up again. Only 'error' and 'cancelled' rows qualify:
139// re-queuing a running or completed one would duplicate work. The bool reports
140// whether a row actually changed.
141func (r *DownloadRepository) Requeue(id int64) (bool, error) {
142 return r.affected(
143 `UPDATE downloads SET status = 'queued', error_message = NULL, logs = NULL,
144 started_at = NULL, completed_at = NULL
145 WHERE id = ? AND status IN ('error', 'cancelled')`,
146 id,
147 )
148}
149
150// CancelQueued marks a not-yet-started download as cancelled. A running one is
151// stopped by cancelling its context instead (see DownloadService.Cancel), which
152// is why the status guard matters: it must not overwrite a row a worker owns.
153func (r *DownloadRepository) CancelQueued(id int64) (bool, error) {
154 return r.affected(
155 `UPDATE downloads SET status = 'cancelled', completed_at = CURRENT_TIMESTAMP
156 WHERE id = ? AND status = 'queued'`,
157 id,
158 )
159}
160
161// affected runs an UPDATE and reports whether it matched a row.
162func (r *DownloadRepository) affected(query string, args ...interface{}) (bool, error) {
163 res, err := r.db.Exec(query, args...)
164 if err != nil {
165 return false, err
166 }
167 n, err := res.RowsAffected()
168 if err != nil {
169 return false, err
170 }
171 return n > 0, nil
172}
173
174func (r *DownloadRepository) MarkCompleted(id int64, status string) error {
175 _, err := r.db.Exec(
176 `UPDATE downloads SET status = ?, completed_at = CURRENT_TIMESTAMP WHERE id = ?`,
177 status, id,
178 )
179 return err
180}
181
182func (r *DownloadRepository) MarkError(id int64, errMsg string) error {
183 _, err := r.db.Exec(
184 `UPDATE downloads SET status = 'error', error_message = ? WHERE id = ?`,
185 errMsg, id,
186 )
187 return err
188}
189
190// HasActiveForSubscription reports whether the given subscription already has a
191// download that is queued or in progress, so the scheduler can avoid stacking a
192// second run on top of one that hasn't finished.
193func (r *DownloadRepository) HasActiveForSubscription(subID int64) (bool, error) {
194 var n int
195 err := r.db.QueryRow(
196 `SELECT COUNT(*) FROM downloads WHERE subscription_id = ? AND status IN ('queued', 'downloading')`,
197 subID,
198 ).Scan(&n)
199 if err != nil {
200 return false, err
201 }
202 return n > 0, nil
203}
204
205func (r *DownloadRepository) Delete(id int64) error {
206 _, err := r.db.Exec(`DELETE FROM downloads WHERE id = ?`, id)
207 return err
208}
209
210func (r *DownloadRepository) DeleteAll() error {
211 _, err := r.db.Exec(`DELETE FROM downloads`)
212 return err
213}
214
215// DeleteByStatus removes every download in one status, backing the queue page's
216// "Clear completed" / "Clear failed" actions.
217func (r *DownloadRepository) DeleteByStatus(status string) error {
218 _, err := r.db.Exec(`DELETE FROM downloads WHERE status = ?`, status)
219 return err
220}
221
222func (r *DownloadRepository) UpdateStatusWhere(oldStatus, newStatus string) error {
223 _, err := r.db.Exec(
224 `UPDATE downloads SET status = ? WHERE status = ?`,
225 newStatus, oldStatus,
226 )
227 return err
228}
229
230func scanDownload(row interface{ Scan(...interface{}) error }) (*models.Download, error) {
231 var d models.Download
232 err := row.Scan(
233 &d.ID, &d.URL, &d.Status,
234 &d.Logs, &d.ErrorMessage, &d.PresetID, &d.FormatOverride, &d.CustomFlags, &d.OutputDir, &d.SubscriptionID,
235 &d.StartedAt, &d.CompletedAt, &d.CreatedAt,
236 )
237 if err != nil {
238 return nil, err
239 }
240 return &d, nil
241}
242