102 lines
3.1 KiB
Go
102 lines
3.1 KiB
Go
|
package store
|
||
|
|
|
||
|
|
import (
|
||
|
|
"context"
|
||
|
|
"errors"
|
||
|
|
"fmt"
|
||
|
|
"time"
|
||
|
|
|
||
|
|
"github.com/jackc/pgx/v5"
|
||
|
|
)
|
||
|
|
|
||
|
|
// SyncRun records one library import.
|
||
|
|
type SyncRun struct {
|
||
|
|
ID int64 `json:"id"`
|
||
|
|
Kind string `json:"kind"`
|
||
|
|
Trigger string `json:"trigger"`
|
||
|
|
Status string `json:"status"`
|
||
|
|
StartedAt time.Time `json:"startedAt"`
|
||
|
|
FinishedAt *time.Time `json:"finishedAt"`
|
||
|
|
ItemsSeen int `json:"itemsSeen"`
|
||
|
|
ItemsUpserted int `json:"itemsUpserted"`
|
||
|
|
ItemsRemoved int `json:"itemsRemoved"`
|
||
|
|
Error string `json:"error"`
|
||
|
|
}
|
||
|
|
|
||
|
|
const (
|
||
|
|
SyncStatusRunning = "running"
|
||
|
|
SyncStatusSuccess = "success"
|
||
|
|
SyncStatusFailed = "failed"
|
||
|
|
)
|
||
|
|
|
||
|
|
func (s *Store) StartSyncRun(ctx context.Context, kind, trigger string) (int64, error) {
|
||
|
|
var id int64
|
||
|
|
err := s.pool.QueryRow(ctx,
|
||
|
|
`INSERT INTO sync_runs (kind, trigger, status) VALUES ($1, $2, $3) RETURNING id`,
|
||
|
|
kind, trigger, SyncStatusRunning).Scan(&id)
|
||
|
|
if err != nil {
|
||
|
|
return 0, fmt.Errorf("store: start sync run: %w", err)
|
||
|
|
}
|
||
|
|
return id, nil
|
||
|
|
}
|
||
|
|
|
||
|
|
func (s *Store) FinishSyncRun(ctx context.Context, id int64, run SyncRun) error {
|
||
|
|
_, err := s.pool.Exec(ctx, `
|
||
|
|
UPDATE sync_runs
|
||
|
|
SET status = $2, finished_at = now(), items_seen = $3,
|
||
|
|
items_upserted = $4, items_removed = $5, error = $6
|
||
|
|
WHERE id = $1`,
|
||
|
|
id, run.Status, run.ItemsSeen, run.ItemsUpserted, run.ItemsRemoved, run.Error)
|
||
|
|
if err != nil {
|
||
|
|
return fmt.Errorf("store: finish sync run: %w", err)
|
||
|
|
}
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
|
||
|
|
func (s *Store) RecentSyncRuns(ctx context.Context, limit int) ([]SyncRun, error) {
|
||
|
|
rows, err := s.pool.Query(ctx, `
|
||
|
|
SELECT id, kind, trigger, status, started_at, finished_at,
|
||
|
|
items_seen, items_upserted, items_removed, error
|
||
|
|
FROM sync_runs ORDER BY started_at DESC LIMIT $1`, limit)
|
||
|
|
if err != nil {
|
||
|
|
return nil, fmt.Errorf("store: recent sync runs: %w", err)
|
||
|
|
}
|
||
|
|
defer rows.Close()
|
||
|
|
|
||
|
|
runs := []SyncRun{}
|
||
|
|
for rows.Next() {
|
||
|
|
var run SyncRun
|
||
|
|
if err := rows.Scan(&run.ID, &run.Kind, &run.Trigger, &run.Status, &run.StartedAt,
|
||
|
|
&run.FinishedAt, &run.ItemsSeen, &run.ItemsUpserted, &run.ItemsRemoved, &run.Error); err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
runs = append(runs, run)
|
||
|
|
}
|
||
|
|
return runs, rows.Err()
|
||
|
|
}
|
||
|
|
|
||
|
|
// LastSuccessfulSyncAt is the watermark an incremental import asks Emby about: "what has
|
||
|
|
// changed since?" Nil means nothing has ever completed, so a full import is required.
|
||
|
|
func (s *Store) LastSuccessfulSyncAt(ctx context.Context) (*time.Time, error) {
|
||
|
|
var at *time.Time
|
||
|
|
err := s.pool.QueryRow(ctx,
|
||
|
|
`SELECT max(started_at) FROM sync_runs WHERE status = $1`, SyncStatusSuccess).Scan(&at)
|
||
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
||
|
|
return nil, nil
|
||
|
|
}
|
||
|
|
if err != nil {
|
||
|
|
return nil, fmt.Errorf("store: last successful sync: %w", err)
|
||
|
|
}
|
||
|
|
return at, nil
|
||
|
|
}
|
||
|
|
|
||
|
|
// MarkStaleRunsFailed cleans up runs left "running" by a crash or a restart mid-import.
|
||
|
|
func (s *Store) MarkStaleRunsFailed(ctx context.Context) error {
|
||
|
|
_, err := s.pool.Exec(ctx, `
|
||
|
|
UPDATE sync_runs
|
||
|
|
SET status = $1, finished_at = now(),
|
||
|
|
error = 'interrupted — the gateway restarted while this import was running'
|
||
|
|
WHERE status = $2`, SyncStatusFailed, SyncStatusRunning)
|
||
|
|
return err
|
||
|
|
}
|