package db import ( "database/sql" "fmt" ) func scanStorageConfig(row interface { Scan(dest ...any) error }) (StorageConfig, error) { var c StorageConfig var cleanMacos, removeSource, inplace int var createdAt string if err := row.Scan( &c.ID, &c.MoverSource, &c.MoverDest, &cleanMacos, &removeSource, &inplace, &c.MoverRsyncOptions, &c.SnapraidContent, &c.SnapraidDataDirs, &c.SnapraidParityDir, &c.SnapraidScrubPlan, &createdAt, ); err != nil { return StorageConfig{}, err } c.MoverCleanMacOS = cleanMacos != 0 c.MoverRemoveSource = removeSource != 0 c.MoverInplace = inplace != 0 c.UpdatedAt = createdAt return c, nil } func (d *DB) GetStorageConfig() (StorageConfig, error) { row := d.conn.QueryRow(`SELECT id, mover_source, mover_dest, mover_clean_macos, mover_remove_source, mover_inplace, mover_rsync_options, snapraid_content, snapraid_data_dirs, snapraid_parity_dir, snapraid_scrub_plan, updated_at FROM storage_config WHERE id = 1`) c, err := scanStorageConfig(row) if err == sql.ErrNoRows { return StorageConfig{ID: 1, MoverCleanMacOS: true, MoverRemoveSource: true, MoverInplace: true, SnapraidScrubPlan: 8}, nil } if err != nil { return StorageConfig{}, fmt.Errorf("get storage config: %w", err) } return c, nil } func (d *DB) UpdateStorageConfig(c StorageConfig) error { _, err := d.conn.Exec(` UPDATE storage_config SET mover_source=?, mover_dest=?, mover_clean_macos=?, mover_remove_source=?, mover_inplace=?, mover_rsync_options=?, snapraid_content=?, snapraid_data_dirs=?, snapraid_parity_dir=?, snapraid_scrub_plan=?, updated_at=datetime('now') WHERE id=1`, c.MoverSource, c.MoverDest, toInt(c.MoverCleanMacOS), toInt(c.MoverRemoveSource), toInt(c.MoverInplace), c.MoverRsyncOptions, c.SnapraidContent, c.SnapraidDataDirs, c.SnapraidParityDir, c.SnapraidScrubPlan, c.ID, ) if err != nil { return fmt.Errorf("update storage config: %w", err) } return nil } func (d *DB) EnsureStorageConfig() error { _, err := d.conn.Exec(`INSERT OR IGNORE INTO storage_config (id) VALUES (1)`) if err != nil { return fmt.Errorf("ensure storage config: %w", err) } return nil } func scanStorageJob(row interface { Scan(dest ...any) error }) (StorageJob, error) { var j StorageJob var startedAt, finishedAt sql.NullString if err := row.Scan( &j.ID, &j.Kind, &j.Status, &j.ArgsJSON, &j.PID, &startedAt, &finishedAt, &j.ExitCode, &j.Output, &j.Error, &j.CreatedAt, ); err != nil { return StorageJob{}, err } if startedAt.Valid { j.StartedAt = &startedAt.String } if finishedAt.Valid { j.FinishedAt = &finishedAt.String } return j, nil } func (d *DB) CreateStorageJob(kind string, argsJSON string) (StorageJob, error) { result, err := d.conn.Exec( `INSERT INTO storage_jobs (kind, status, args_json) VALUES (?, 'queued', ?)`, kind, argsJSON, ) if err != nil { return StorageJob{}, fmt.Errorf("create storage job: %w", err) } id, err := result.LastInsertId() if err != nil { return StorageJob{}, fmt.Errorf("last insert id: %w", err) } return d.GetStorageJob(id) } func (d *DB) GetStorageJob(id int64) (StorageJob, error) { row := d.conn.QueryRow(` SELECT id, kind, status, args_json, pid, started_at, finished_at, exit_code, output, error, created_at FROM storage_jobs WHERE id=?`, id) j, err := scanStorageJob(row) if err == sql.ErrNoRows { return StorageJob{}, fmt.Errorf("storage job not found") } if err != nil { return StorageJob{}, fmt.Errorf("get storage job: %w", err) } return j, nil } func (d *DB) MarkJobRunning(id int64, pid int) error { _, err := d.conn.Exec( `UPDATE storage_jobs SET status='running', pid=?, started_at=datetime('now') WHERE id=?`, pid, id, ) if err != nil { return fmt.Errorf("mark job running: %w", err) } return nil } func (d *DB) MarkJobFinished(id int64, status string, exitCode int, output, errMsg string) error { _, err := d.conn.Exec( `UPDATE storage_jobs SET status=?, exit_code=?, output=?, error=?, finished_at=datetime('now') WHERE id=?`, status, exitCode, output, errMsg, id, ) if err != nil { return fmt.Errorf("mark job finished: %w", err) } return nil } func (d *DB) AppendJobOutput(id int64, line string) error { _, err := d.conn.Exec( `UPDATE storage_jobs SET output = output || ? || char(10) WHERE id=?`, line, id, ) if err != nil { return fmt.Errorf("append job output: %w", err) } return nil } func (d *DB) TrimJobOutput(id int64, maxLines int) error { _ = maxLines return nil } func (d *DB) ListStorageJobs(limit int) ([]StorageJob, error) { if limit <= 0 { limit = 50 } rows, err := d.conn.Query(` SELECT id, kind, status, args_json, pid, started_at, finished_at, exit_code, output, error, created_at FROM storage_jobs ORDER BY created_at DESC LIMIT ?`, limit) if err != nil { return nil, fmt.Errorf("list storage jobs: %w", err) } defer rows.Close() jobs := make([]StorageJob, 0) for rows.Next() { j, err := scanStorageJob(rows) if err != nil { return nil, fmt.Errorf("scan storage job: %w", err) } jobs = append(jobs, j) } return jobs, rows.Err() } func (d *DB) GetRunningJob() (*StorageJob, error) { row := d.conn.QueryRow(` SELECT id, kind, status, args_json, pid, started_at, finished_at, exit_code, output, error, created_at FROM storage_jobs WHERE status IN ('queued','running') ORDER BY created_at ASC LIMIT 1`) j, err := scanStorageJob(row) if err == sql.ErrNoRows { return nil, nil } if err != nil { return nil, fmt.Errorf("get running job: %w", err) } return &j, nil } func (d *DB) ResetOrphanedJobs() error { _, err := d.conn.Exec( `UPDATE storage_jobs SET status='interrupted', finished_at=datetime('now') WHERE status IN ('queued','running')`, ) if err != nil { return fmt.Errorf("reset orphaned jobs: %w", err) } return nil } func (d *DB) SetJobCancelled(id int64) error { _, err := d.conn.Exec( `UPDATE storage_jobs SET status='cancelled', finished_at=datetime('now') WHERE id=?`, id, ) if err != nil { return fmt.Errorf("set job cancelled: %w", err) } return nil } func toInt(b bool) int { if b { return 1 } return 0 }