darroyo
84b185be39
Phase A - Stability: - Engine.Start(): recovers orphaned jobs (running/queued/waking_up) after crash - Engine.Stop(): graceful shutdown - cancels all in-flight jobs and waits - Queue keyed by jobID (not syncPairID): cancel now targets exact job - Local rsync uses jobCtx (context.Background() replaced) - Migrations wrapped in transactions; checksums stored Phase B - Security: - admin/admin default removed: SYNCSERVER_ADMIN_PASSWORD required on first run - Path validation: rejects .., leading -, null bytes in sync pair paths - Rsync flags allowlist: dangerous flags blocked (--rsync-path, -e, --files-from) - Shell concat in RunRemote replaced with proper sh -c escaping - knownhosts: replaced custom parser with golang.org/x/crypto/ssh/knownhosts - RequireAdmin wired: machine CRUD, SSH key ops, settings require admin role - deploy-keys: uses authorized_keys only (no private key upload) - Hardcoded /var/lib/syncserver/ssh paths replaced with cfg.SSHDir() Phase C - Operational: - /readyz health check: DB query + SSH dir accessibility - /metrics endpoint: Prometheus text format (jobs, queue, machines) - Event struct JSON tags: job_id, machine_id, type (snake_case) - EventBus broadcast: fanned out to all subscribers - SQLite VACUUM INTO backup: scheduled before cleanup if BackupDir set - Filesystem job log cleanup: removes .log files for purged jobs - Backup retention: old backups auto-purged Phase D - Frontend: - Schedules page: REST API + full CRUD UI for cron schedules - Dashboard: cancel button for running/queued jobs - JobDetail: server-side log download via API - Settings: displays data_dir from server - 404 page: proper NotFound component Phase E - Tests: - auth_test.go: JWT, bcrypt, middleware, seed (18 tests) - models_test.go: Job, Machine, SyncPair, Schedule repos (18 tests) - go test -race: no data races found
119 lines
2.6 KiB
Go
119 lines
2.6 KiB
Go
package db
|
|
|
|
import (
|
|
"crypto/sha256"
|
|
"database/sql"
|
|
"encoding/hex"
|
|
"fmt"
|
|
"os"
|
|
"path/filepath"
|
|
"sort"
|
|
"strings"
|
|
|
|
_ "modernc.org/sqlite"
|
|
)
|
|
|
|
type DB struct {
|
|
*sql.DB
|
|
}
|
|
|
|
func (d *DB) SQLDB() *sql.DB {
|
|
return d.DB
|
|
}
|
|
|
|
func Open(dbPath string) (*DB, error) {
|
|
if err := os.MkdirAll(filepath.Dir(dbPath), 0700); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
db, err := sql.Open("sqlite", dbPath+"?_pragma=busy_timeout(5000)&_pragma=journal_mode(WAL)")
|
|
if err != nil {
|
|
return nil, fmt.Errorf("opening db: %w", err)
|
|
}
|
|
|
|
if _, err := db.Exec("PRAGMA foreign_keys = ON"); err != nil {
|
|
db.Close()
|
|
return nil, fmt.Errorf("enabling foreign_keys: %w", err)
|
|
}
|
|
|
|
return &DB{db}, nil
|
|
}
|
|
|
|
func (db *DB) Close() error {
|
|
return db.DB.Close()
|
|
}
|
|
|
|
func (db *DB) RunMigrations() error {
|
|
return db.runMigrationsInternal(migrationsFS, "migrations")
|
|
}
|
|
|
|
func (db *DB) runMigrationsInternal(mfs embedFS, migrationsRoot string) error {
|
|
entries, err := mfs.ReadDir(migrationsRoot)
|
|
if err != nil {
|
|
return fmt.Errorf("reading migrations dir: %w", err)
|
|
}
|
|
|
|
var names []string
|
|
for _, e := range entries {
|
|
if !e.IsDir() && strings.HasSuffix(e.Name(), ".sql") {
|
|
names = append(names, e.Name())
|
|
}
|
|
}
|
|
sort.Strings(names)
|
|
|
|
if _, err := db.Exec(`
|
|
CREATE TABLE IF NOT EXISTS schema_migrations (
|
|
version TEXT PRIMARY KEY,
|
|
checksum TEXT,
|
|
applied_at DATETIME DEFAULT CURRENT_TIMESTAMP
|
|
)
|
|
`); err != nil {
|
|
return fmt.Errorf("creating schema_migrations table: %w", err)
|
|
}
|
|
|
|
for _, name := range names {
|
|
var storedChecksum string
|
|
row := db.QueryRow("SELECT checksum FROM schema_migrations WHERE version = ?", name)
|
|
if err := row.Scan(&storedChecksum); err == nil && storedChecksum != "" {
|
|
continue
|
|
}
|
|
|
|
data, err := mfs.ReadFile(filepath.Join(migrationsRoot, name))
|
|
if err != nil {
|
|
return fmt.Errorf("reading migration %s: %w", name, err)
|
|
}
|
|
|
|
checksum := sha256.Sum256(data)
|
|
checksumHex := hex.EncodeToString(checksum[:])
|
|
|
|
tx, err := db.Begin()
|
|
if err != nil {
|
|
return fmt.Errorf("starting transaction for migration %s: %w", name, err)
|
|
}
|
|
|
|
if _, err := tx.Exec(string(data)); err != nil {
|
|
tx.Rollback()
|
|
return fmt.Errorf("applying migration %s: %w", name, err)
|
|
}
|
|
|
|
if _, err := tx.Exec(
|
|
"INSERT INTO schema_migrations (version, checksum) VALUES (?, ?)",
|
|
name, checksumHex,
|
|
); err != nil {
|
|
tx.Rollback()
|
|
return fmt.Errorf("recording migration %s: %w", name, err)
|
|
}
|
|
|
|
if err := tx.Commit(); err != nil {
|
|
return fmt.Errorf("committing migration %s: %w", name, err)
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
type embedFS interface {
|
|
ReadDir(name string) ([]os.DirEntry, error)
|
|
ReadFile(name string) ([]byte, error)
|
|
}
|