restic-manager/internal/store/schedules.go

package store

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"time"
)

// CreateSchedule inserts a new schedule and bumps the host's
// schedule_version atomically. Returns the inserted row's
// CreatedAt / UpdatedAt timestamps written into s.
func (st *Store) CreateSchedule(ctx context.Context, s *Schedule) error {
	if s.ID == "" || s.HostID == "" {
		return errors.New("store: schedule id and host_id required")
	}
	now := time.Now().UTC()
	s.CreatedAt = now
	s.UpdatedAt = now
	if s.Paths == nil {
		s.Paths = []string{}
	}
	if s.Excludes == nil {
		s.Excludes = []string{}
	}
	if s.Tags == nil {
		s.Tags = []string{}
	}
	pathsJSON, _ := json.Marshal(s.Paths)
	excludesJSON, _ := json.Marshal(s.Excludes)
	tagsJSON, _ := json.Marshal(s.Tags)
	retentionJSON, _ := json.Marshal(s.RetentionPolicy)
	optionsJSON, _ := json.Marshal(s.Options)

	tx, err := st.db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("store: begin tx: %w", err)
	}
	defer func() { _ = tx.Rollback() }()

	if _, err := tx.ExecContext(ctx,
		`INSERT INTO schedules (
			id, host_id, kind, cron_expr, paths, excludes, tags,
			retention_policy, options, pre_hook, post_hook, enabled,
			created_at, updated_at
		) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
		s.ID, s.HostID, s.Kind, s.CronExpr,
		string(pathsJSON), string(excludesJSON), string(tagsJSON),
		string(retentionJSON), string(optionsJSON),
		s.PreHook, s.PostHook, boolToInt(s.Enabled),
		now.Format(time.RFC3339Nano), now.Format(time.RFC3339Nano),
	); err != nil {
		return fmt.Errorf("store: create schedule: %w", err)
	}
	if err := bumpHostScheduleVersionTx(ctx, tx, s.HostID); err != nil {
		return err
	}
	return tx.Commit()
}

// UpdateSchedule replaces every editable field on an existing row
// and bumps host_schedule_version. ID and HostID must match an
// existing row; kind is immutable (creating a new schedule is
// cheaper than re-keying retention/hooks).
func (st *Store) UpdateSchedule(ctx context.Context, s *Schedule) error {
	if s.ID == "" || s.HostID == "" {
		return errors.New("store: schedule id and host_id required")
	}
	if s.Paths == nil {
		s.Paths = []string{}
	}
	if s.Excludes == nil {
		s.Excludes = []string{}
	}
	if s.Tags == nil {
		s.Tags = []string{}
	}
	pathsJSON, _ := json.Marshal(s.Paths)
	excludesJSON, _ := json.Marshal(s.Excludes)
	tagsJSON, _ := json.Marshal(s.Tags)
	retentionJSON, _ := json.Marshal(s.RetentionPolicy)
	optionsJSON, _ := json.Marshal(s.Options)
	now := time.Now().UTC()

	tx, err := st.db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("store: begin tx: %w", err)
	}
	defer func() { _ = tx.Rollback() }()

	res, err := tx.ExecContext(ctx,
		`UPDATE schedules SET
			cron_expr = ?, paths = ?, excludes = ?, tags = ?,
			retention_policy = ?, options = ?,
			pre_hook = ?, post_hook = ?, enabled = ?,
			updated_at = ?
		 WHERE id = ? AND host_id = ?`,
		s.CronExpr,
		string(pathsJSON), string(excludesJSON), string(tagsJSON),
		string(retentionJSON), string(optionsJSON),
		s.PreHook, s.PostHook, boolToInt(s.Enabled),
		now.Format(time.RFC3339Nano),
		s.ID, s.HostID,
	)
	if err != nil {
		return fmt.Errorf("store: update schedule: %w", err)
	}
	n, _ := res.RowsAffected()
	if n == 0 {
		return ErrNotFound
	}
	s.UpdatedAt = now
	if err := bumpHostScheduleVersionTx(ctx, tx, s.HostID); err != nil {
		return err
	}
	return tx.Commit()
}

// DeleteSchedule removes a schedule and bumps host_schedule_version.
// Returns ErrNotFound if no row matched.
func (st *Store) DeleteSchedule(ctx context.Context, hostID, scheduleID string) error {
	tx, err := st.db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("store: begin tx: %w", err)
	}
	defer func() { _ = tx.Rollback() }()

	res, err := tx.ExecContext(ctx,
		`DELETE FROM schedules WHERE id = ? AND host_id = ?`,
		scheduleID, hostID)
	if err != nil {
		return fmt.Errorf("store: delete schedule: %w", err)
	}
	n, _ := res.RowsAffected()
	if n == 0 {
		return ErrNotFound
	}
	if err := bumpHostScheduleVersionTx(ctx, tx, hostID); err != nil {
		return err
	}
	return tx.Commit()
}

// GetSchedule returns one schedule by (host_id, id). Returns
// ErrNotFound on miss.
func (st *Store) GetSchedule(ctx context.Context, hostID, scheduleID string) (*Schedule, error) {
	row := st.db.QueryRowContext(ctx,
		`SELECT id, host_id, kind, cron_expr, paths, excludes, tags,
			retention_policy, options, pre_hook, post_hook, enabled,
			created_at, updated_at
		 FROM schedules WHERE id = ? AND host_id = ?`,
		scheduleID, hostID)
	s, err := scanSchedule(row)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, ErrNotFound
	}
	return s, err
}

// ListSchedulesByHost returns every schedule for a host, ordered
// by created_at. Empty slice on miss (not an error).
func (st *Store) ListSchedulesByHost(ctx context.Context, hostID string) ([]Schedule, error) {
	rows, err := st.db.QueryContext(ctx,
		`SELECT id, host_id, kind, cron_expr, paths, excludes, tags,
			retention_policy, options, pre_hook, post_hook, enabled,
			created_at, updated_at
		 FROM schedules WHERE host_id = ? ORDER BY created_at`,
		hostID)
	if err != nil {
		return nil, fmt.Errorf("store: list schedules: %w", err)
	}
	defer rows.Close()
	out := []Schedule{}
	for rows.Next() {
		s, err := scanScheduleRow(rows)
		if err != nil {
			return nil, err
		}
		out = append(out, *s)
	}
	return out, rows.Err()
}

// GetHostScheduleVersion returns the current version for a host,
// or 0 if no row exists yet.
func (st *Store) GetHostScheduleVersion(ctx context.Context, hostID string) (int64, error) {
	var v int64
	err := st.db.QueryRowContext(ctx,
		`SELECT version FROM host_schedule_version WHERE host_id = ?`, hostID).Scan(&v)
	if errors.Is(err, sql.ErrNoRows) {
		return 0, nil
	}
	if err != nil {
		return 0, fmt.Errorf("store: get schedule version: %w", err)
	}
	return v, nil
}

// SetHostAppliedScheduleVersion records the version the agent has
// confirmed via schedule.ack. Idempotent.
func (st *Store) SetHostAppliedScheduleVersion(ctx context.Context, hostID string, version int64) error {
	_, err := st.db.ExecContext(ctx,
		`UPDATE hosts SET applied_schedule_version = ? WHERE id = ?`,
		version, hostID)
	if err != nil {
		return fmt.Errorf("store: set applied schedule version: %w", err)
	}
	return nil
}

// bumpHostScheduleVersionTx upserts host_schedule_version, +1 each
// call. Caller owns the tx.
func bumpHostScheduleVersionTx(ctx context.Context, tx *sql.Tx, hostID string) error {
	if _, err := tx.ExecContext(ctx,
		`INSERT INTO host_schedule_version (host_id, version)
		 VALUES (?, 1)
		 ON CONFLICT(host_id) DO UPDATE SET version = version + 1`,
		hostID); err != nil {
		return fmt.Errorf("store: bump schedule version: %w", err)
	}
	return nil
}

// ----- scan helpers --------------------------------------------------

func scanSchedule(row *sql.Row) (*Schedule, error) {
	return scanScheduleRow(row)
}

type scheduleScanner interface {
	Scan(dest ...any) error
}

func scanScheduleRow(s scheduleScanner) (*Schedule, error) {
	var (
		out                                          Schedule
		paths, excludes, tags, retention, options    string
		createdAt, updatedAt                         string
		enabled                                      int
	)
	err := s.Scan(&out.ID, &out.HostID, &out.Kind, &out.CronExpr,
		&paths, &excludes, &tags, &retention, &options,
		&out.PreHook, &out.PostHook, &enabled,
		&createdAt, &updatedAt)
	if err != nil {
		return nil, err
	}
	if paths != "" {
		_ = json.Unmarshal([]byte(paths), &out.Paths)
	}
	if excludes != "" {
		_ = json.Unmarshal([]byte(excludes), &out.Excludes)
	}
	if tags != "" {
		_ = json.Unmarshal([]byte(tags), &out.Tags)
	}
	if retention != "" {
		_ = json.Unmarshal([]byte(retention), &out.RetentionPolicy)
	}
	if options != "" {
		_ = json.Unmarshal([]byte(options), &out.Options)
	}
	out.Enabled = enabled != 0
	if t, err := time.Parse(time.RFC3339Nano, createdAt); err == nil {
		out.CreatedAt = t
	}
	if t, err := time.Parse(time.RFC3339Nano, updatedAt); err == nil {
		out.UpdatedAt = t
	}
	return &out, nil
}

func boolToInt(b bool) int {
	if b {
		return 1
	}
	return 0
}