vidarchive

vidarchive/internal/repository/subscription.go
subscription.go
⎇
Raw
1package repository
2
3import (
4 "database/sql"
5 "fmt"
6 "time"
7
8 "vidarchive/internal/models"
9)
10
11type SubscriptionRepository struct {
12 db *sql.DB
13}
14
15func NewSubscriptionRepository(db *sql.DB) *SubscriptionRepository {
16 return &SubscriptionRepository{db: db}
17}
18
19const subscriptionColumns = `id, name, url, enabled, refresh_mode, schedule_kind, cron_expr,
20 preset_id, format_override, custom_flags, output_dir, prune_removed,
21 last_run_at, next_run_at, last_status, created_at`
22
23func (r *SubscriptionRepository) Create(s *models.Subscription) error {
24 result, err := r.db.Exec(
25 `INSERT INTO subscriptions
26 (name, url, enabled, refresh_mode, schedule_kind, cron_expr, preset_id,
27 format_override, custom_flags, output_dir, prune_removed, next_run_at)
28 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
29 s.Name, s.URL, boolToInt(s.Enabled), s.RefreshMode, s.ScheduleKind, s.CronExpr,
30 s.PresetID, s.FormatOverride, s.CustomFlags, s.OutputDir, boolToInt(s.PruneRemoved),
31 s.NextRunAt,
32 )
33 if err != nil {
34 return err
35 }
36 id, err := result.LastInsertId()
37 if err != nil {
38 return fmt.Errorf("subscription last insert id: %w", err)
39 }
40 s.ID = id
41 return nil
42}
43
44func (r *SubscriptionRepository) GetByID(id int64) (*models.Subscription, error) {
45 row := r.db.QueryRow(`SELECT `+subscriptionColumns+` FROM subscriptions WHERE id = ?`, id)
46 return scanSubscription(row)
47}
48
49func (r *SubscriptionRepository) GetAll() ([]*models.Subscription, error) {
50 rows, err := r.db.Query(`SELECT ` + subscriptionColumns + ` FROM subscriptions ORDER BY created_at DESC`)
51 if err != nil {
52 return nil, err
53 }
54 defer rows.Close()
55 return scanSubscriptions(rows)
56}
57
58// GetDue returns enabled subscriptions due at or before now, or never run.
59func (r *SubscriptionRepository) GetDue(now time.Time) ([]*models.Subscription, error) {
60 rows, err := r.db.Query(
61 `SELECT `+subscriptionColumns+`
62 FROM subscriptions
63 WHERE enabled = 1 AND (next_run_at IS NULL OR next_run_at <= ?)
64 ORDER BY created_at ASC`,
65 now,
66 )
67 if err != nil {
68 return nil, err
69 }
70 defer rows.Close()
71 return scanSubscriptions(rows)
72}
73
74func (r *SubscriptionRepository) Update(s *models.Subscription) error {
75 _, err := r.db.Exec(
76 `UPDATE subscriptions SET name=?, url=?, enabled=?, refresh_mode=?, schedule_kind=?,
77 cron_expr=?, preset_id=?, format_override=?, custom_flags=?, output_dir=?,
78 prune_removed=?, next_run_at=?
79 WHERE id=?`,
80 s.Name, s.URL, boolToInt(s.Enabled), s.RefreshMode, s.ScheduleKind, s.CronExpr,
81 s.PresetID, s.FormatOverride, s.CustomFlags, s.OutputDir, boolToInt(s.PruneRemoved),
82 s.NextRunAt, s.ID,
83 )
84 return err
85}
86
87func (r *SubscriptionRepository) SetEnabled(id int64, enabled bool) error {
88 _, err := r.db.Exec(`UPDATE subscriptions SET enabled = ? WHERE id = ?`, boolToInt(enabled), id)
89 return err
90}
91
92func (r *SubscriptionRepository) MarkRun(id int64, lastRunAt, nextRunAt time.Time, status string) error {
93 _, err := r.db.Exec(
94 `UPDATE subscriptions SET last_run_at = ?, next_run_at = ?, last_status = ? WHERE id = ?`,
95 lastRunAt, nextRunAt, status, id,
96 )
97 return err
98}
99
100// MarkManualRun records a run started by hand. Unlike MarkRun it leaves
101// next_run_at alone: running a subscription now must not move its schedule.
102func (r *SubscriptionRepository) MarkManualRun(id int64, lastRunAt time.Time, status string) error {
103 _, err := r.db.Exec(
104 `UPDATE subscriptions SET last_run_at = ?, last_status = ? WHERE id = ?`,
105 lastRunAt, status, id,
106 )
107 return err
108}
109
110// SetLastStatus leaves the run times alone. The worker calls it as the run
111// progresses, so the subscription does not stay on the status it was queued with.
112func (r *SubscriptionRepository) SetLastStatus(id int64, status string) error {
113 _, err := r.db.Exec(`UPDATE subscriptions SET last_status = ? WHERE id = ?`, status, id)
114 return err
115}
116
117func (r *SubscriptionRepository) Delete(id int64) error {
118 _, err := r.db.Exec(`DELETE FROM subscriptions WHERE id = ?`, id)
119 return err
120}
121
122func scanSubscriptions(rows *sql.Rows) ([]*models.Subscription, error) {
123 var subs []*models.Subscription
124 for rows.Next() {
125 s, err := scanSubscription(rows)
126 if err != nil {
127 return nil, err
128 }
129 subs = append(subs, s)
130 }
131 return subs, rows.Err()
132}
133
134func scanSubscription(row interface{ Scan(...interface{}) error }) (*models.Subscription, error) {
135 var s models.Subscription
136 var enabled, pruneRemoved int
137 var refreshMode, scheduleKind, cronExpr, formatOverride, customFlags, outputDir sql.NullString
138 err := row.Scan(
139 &s.ID, &s.Name, &s.URL, &enabled, &refreshMode, &scheduleKind, &cronExpr,
140 &s.PresetID, &formatOverride, &customFlags, &outputDir, &pruneRemoved,
141 &s.LastRunAt, &s.NextRunAt, &s.LastStatus, &s.CreatedAt,
142 )
143 if err != nil {
144 return nil, err
145 }
146 s.Enabled = enabled == 1
147 s.PruneRemoved = pruneRemoved == 1
148 s.RefreshMode = refreshMode.String
149 s.ScheduleKind = scheduleKind.String
150 s.CronExpr = cronExpr.String
151 s.FormatOverride = formatOverride.String
152 s.CustomFlags = customFlags.String
153 s.OutputDir = outputDir.String
154 return &s, nil
155}
156