92 lines
2.7 KiB
Go
92 lines
2.7 KiB
Go
|
package store
|
||
|
|
|
||
|
|
import (
|
||
|
|
"context"
|
||
|
|
"database/sql"
|
||
|
|
"fmt"
|
||
|
|
)
|
||
|
|
|
||
|
|
const (
|
||
|
|
SyncKindBackfill = "backfill"
|
||
|
|
SyncKindIncremental = "incremental"
|
||
|
|
|
||
|
|
SyncStatusRunning = "running"
|
||
|
|
SyncStatusSuccess = "success"
|
||
|
|
SyncStatusError = "error"
|
||
|
|
)
|
||
|
|
|
||
|
|
// SyncRun records one backfill or incremental sync attempt.
|
||
|
|
type SyncRun struct {
|
||
|
|
ID int64
|
||
|
|
Kind string
|
||
|
|
StartedAt string
|
||
|
|
FinishedAt *string
|
||
|
|
ActivitiesFetched int
|
||
|
|
Status string
|
||
|
|
ErrorMessage *string
|
||
|
|
}
|
||
|
|
|
||
|
|
// StartSyncRun records a new in-progress sync run and returns its id.
|
||
|
|
func (db *DB) StartSyncRun(ctx context.Context, kind string) (int64, error) {
|
||
|
|
res, err := db.ExecContext(ctx, `
|
||
|
|
INSERT INTO sync_runs (kind, started_at, status) VALUES (?, datetime('now'), ?)`,
|
||
|
|
kind, SyncStatusRunning)
|
||
|
|
if err != nil {
|
||
|
|
return 0, fmt.Errorf("start sync run: %w", err)
|
||
|
|
}
|
||
|
|
return res.LastInsertId()
|
||
|
|
}
|
||
|
|
|
||
|
|
// FinishSyncRun marks a sync run as finished, recording how many activities
|
||
|
|
// were fetched and whether it succeeded.
|
||
|
|
func (db *DB) FinishSyncRun(ctx context.Context, id int64, activitiesFetched int, errMsg *string) error {
|
||
|
|
status := SyncStatusSuccess
|
||
|
|
if errMsg != nil {
|
||
|
|
status = SyncStatusError
|
||
|
|
}
|
||
|
|
_, err := db.ExecContext(ctx, `
|
||
|
|
UPDATE sync_runs SET finished_at = datetime('now'), activities_fetched = ?, status = ?, error_message = ?
|
||
|
|
WHERE id = ?`, activitiesFetched, status, errMsg, id)
|
||
|
|
if err != nil {
|
||
|
|
return fmt.Errorf("finish sync run %d: %w", id, err)
|
||
|
|
}
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
|
||
|
|
// LatestSyncRun returns the most recent sync run, if any.
|
||
|
|
func (db *DB) LatestSyncRun(ctx context.Context) (SyncRun, bool, error) {
|
||
|
|
row := db.QueryRowContext(ctx, `
|
||
|
|
SELECT id, kind, started_at, finished_at, activities_fetched, status, error_message
|
||
|
|
FROM sync_runs ORDER BY id DESC LIMIT 1`)
|
||
|
|
var r SyncRun
|
||
|
|
err := row.Scan(&r.ID, &r.Kind, &r.StartedAt, &r.FinishedAt, &r.ActivitiesFetched, &r.Status, &r.ErrorMessage)
|
||
|
|
if err == sql.ErrNoRows {
|
||
|
|
return SyncRun{}, false, nil
|
||
|
|
}
|
||
|
|
if err != nil {
|
||
|
|
return SyncRun{}, false, fmt.Errorf("latest sync run: %w", err)
|
||
|
|
}
|
||
|
|
return r, true, nil
|
||
|
|
}
|
||
|
|
|
||
|
|
// ListSyncRuns returns recent sync runs, newest first.
|
||
|
|
func (db *DB) ListSyncRuns(ctx context.Context, limit int) ([]SyncRun, error) {
|
||
|
|
rows, err := db.QueryContext(ctx, `
|
||
|
|
SELECT id, kind, started_at, finished_at, activities_fetched, status, error_message
|
||
|
|
FROM sync_runs ORDER BY id DESC LIMIT ?`, limit)
|
||
|
|
if err != nil {
|
||
|
|
return nil, fmt.Errorf("list sync runs: %w", err)
|
||
|
|
}
|
||
|
|
defer rows.Close()
|
||
|
|
|
||
|
|
runs := []SyncRun{}
|
||
|
|
for rows.Next() {
|
||
|
|
var r SyncRun
|
||
|
|
if err := rows.Scan(&r.ID, &r.Kind, &r.StartedAt, &r.FinishedAt, &r.ActivitiesFetched, &r.Status, &r.ErrorMessage); err != nil {
|
||
|
|
return nil, fmt.Errorf("scan sync run row: %w", err)
|
||
|
|
}
|
||
|
|
runs = append(runs, r)
|
||
|
|
}
|
||
|
|
return runs, rows.Err()
|
||
|
|
}
|