vidarchive

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