diff --git a/.gitignore b/.gitignore index d1157d1..d310baa 100644 --- a/.gitignore +++ b/.gitignore @@ -181,6 +181,9 @@ bin/ dist/deb/ dist/*.deb +# Non-versionable docs +FEATURES.md + # Frontend web/node_modules/ web/.tsbuild/ diff --git a/Makefile b/Makefile index 1ffde36..b247914 100644 --- a/Makefile +++ b/Makefile @@ -1,5 +1,5 @@ BINARY=nasctl -VERSION?=0.6.1 +VERSION?=0.7.0 GO?=go LDFLAGS=-s -w -X github.com/darroyo/nasctl/internal/web.Version=$(VERSION) -X github.com/darroyo/nasctl/internal/web.Commit=$(shell git rev-parse --short HEAD 2>/dev/null || echo unknown) BUILD_FLAGS=CGO_ENABLED=0 diff --git a/cmd/nasctl/main.go b/cmd/nasctl/main.go index 32f52d8..b6f0f1b 100644 --- a/cmd/nasctl/main.go +++ b/cmd/nasctl/main.go @@ -15,6 +15,7 @@ import ( "github.com/darroyo/nasctl/internal/modules/nfs" "github.com/darroyo/nasctl/internal/modules/samba" "github.com/darroyo/nasctl/internal/modules/users" + "github.com/darroyo/nasctl/internal/storage" "github.com/darroyo/nasctl/internal/web" ) @@ -46,6 +47,11 @@ func main() { log.Fatalf("migrate db: %v", err) } + jm := storage.NewJobManager(database, *execSystem) + if err := jm.ResetOrphans(); err != nil { + log.Printf("reset orphaned storage jobs: %v", err) + } + auth, err := web.NewAuthService(database) if err != nil { log.Fatalf("init auth: %v", err) @@ -93,6 +99,7 @@ func main() { AdminUsername: *adminUser, UploadMaxBytes: *uploadMaxBytes, PreviewMaxBytes: *previewMaxBytes, + JobManager: jm, }) log.Printf("nasctl %s (commit %s) listening on %s (db=%s exec-system=%v)", web.Version, web.Commit, *addr, *dbPath, *execSystem) diff --git a/internal/db/db.go b/internal/db/db.go index d87f1d5..aa3945b 100644 --- a/internal/db/db.go +++ b/internal/db/db.go @@ -91,6 +91,11 @@ func (d *DB) Migrate() error { return fmt.Errorf("fix zero fsids: %w", err) } } + if name == "0008_storage.sql" { + if err := d.EnsureStorageConfig(); err != nil { + return fmt.Errorf("ensure storage config: %w", err) + } + } } return nil } diff --git a/internal/db/migrations/0008_storage.sql b/internal/db/migrations/0008_storage.sql new file mode 100644 index 0000000..aa5acb9 --- /dev/null +++ b/internal/db/migrations/0008_storage.sql @@ -0,0 +1,31 @@ +CREATE TABLE IF NOT EXISTS storage_config ( + id INTEGER PRIMARY KEY CHECK (id = 1), + mover_source TEXT NOT NULL DEFAULT '', + mover_dest TEXT NOT NULL DEFAULT '', + mover_clean_macos INTEGER NOT NULL DEFAULT 1, + mover_remove_source INTEGER NOT NULL DEFAULT 1, + mover_inplace INTEGER NOT NULL DEFAULT 1, + mover_rsync_options TEXT NOT NULL DEFAULT '', + snapraid_content TEXT NOT NULL DEFAULT '', + snapraid_data_dirs TEXT NOT NULL DEFAULT '', + snapraid_parity_dir TEXT NOT NULL DEFAULT '', + snapraid_scrub_plan INTEGER NOT NULL DEFAULT 8, + updated_at TEXT NOT NULL DEFAULT (datetime('now')) +); + +CREATE TABLE IF NOT EXISTS storage_jobs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + kind TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'queued', + args_json TEXT NOT NULL DEFAULT '{}', + pid INTEGER NOT NULL DEFAULT 0, + started_at TEXT, + finished_at TEXT, + exit_code INTEGER NOT NULL DEFAULT -1, + output TEXT NOT NULL DEFAULT '', + error TEXT NOT NULL DEFAULT '', + created_at TEXT NOT NULL DEFAULT (datetime('now')) +); + +CREATE INDEX IF NOT EXISTS idx_storage_jobs_created ON storage_jobs(created_at DESC); +CREATE INDEX IF NOT EXISTS idx_storage_jobs_kind_created ON storage_jobs(kind, created_at DESC); diff --git a/internal/db/models.go b/internal/db/models.go index c1d8a1c..b4ba270 100644 --- a/internal/db/models.go +++ b/internal/db/models.go @@ -87,3 +87,39 @@ type ApplyLogEntry struct { Success bool `json:"success"` CreatedAt time.Time `json:"created_at"` } + +type StorageConfig struct { + ID int64 `json:"id"` + MoverSource string `json:"mover_source"` + MoverDest string `json:"mover_dest"` + MoverCleanMacOS bool `json:"mover_clean_macos"` + MoverRemoveSource bool `json:"mover_remove_source"` + MoverInplace bool `json:"mover_inplace"` + MoverRsyncOptions string `json:"mover_rsync_options"` + SnapraidContent string `json:"snapraid_content"` + SnapraidDataDirs string `json:"snapraid_data_dirs"` + SnapraidParityDir string `json:"snapraid_parity_dir"` + SnapraidScrubPlan int `json:"snapraid_scrub_plan"` + UpdatedAt string `json:"updated_at"` +} + +type StorageJob struct { + ID int64 `json:"id"` + Kind string `json:"kind"` + Status string `json:"status"` + ArgsJSON string `json:"args_json"` + PID int `json:"pid"` + StartedAt *string `json:"started_at,omitempty"` + FinishedAt *string `json:"finished_at,omitempty"` + ExitCode int `json:"exit_code"` + Output string `json:"output"` + Error string `json:"error"` + CreatedAt string `json:"created_at"` +} + +type StorageCapabilities struct { + Rsync bool `json:"rsync"` + MergerfsBin bool `json:"mergerfs"` + MergerfsMounted bool `json:"mergerfs_mounted"` + SnapraidBin bool `json:"snapraid"` +} diff --git a/internal/db/queries_storage.go b/internal/db/queries_storage.go new file mode 100644 index 0000000..8941905 --- /dev/null +++ b/internal/db/queries_storage.go @@ -0,0 +1,226 @@ +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() + + var jobs []StorageJob + 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 +} + + diff --git a/internal/storage/capabilities.go b/internal/storage/capabilities.go new file mode 100644 index 0000000..99dfa8e --- /dev/null +++ b/internal/storage/capabilities.go @@ -0,0 +1,34 @@ +package storage + +import ( + "context" + + "github.com/darroyo/nasctl/internal/db" + "github.com/darroyo/nasctl/internal/system" +) + +type Capabilities struct { + Rsync bool `json:"rsync"` + MergerfsBin bool `json:"mergerfs"` + MergerfsMounted bool `json:"mergerfs_mounted"` + SnapraidBin bool `json:"snapraid"` +} + +func Probe(ctx context.Context) (Capabilities, error) { + rsyncOut, _, _ := system.Run(ctx, "command", "-v", "rsync") + snapraidOut, _, _ := system.Run(ctx, "command", "-v", "snapraid") + mergerfsOut, _, _ := system.Run(ctx, "command", "-v", "mergerfs") + + _, _, mergerfsMountedErr := system.Run(ctx, "grep", "-q", "fuse.mergerfs", "/proc/mounts") + + return Capabilities{ + Rsync: len(rsyncOut) > 0, + MergerfsBin: len(mergerfsOut) > 0, + MergerfsMounted: mergerfsMountedErr == nil, + SnapraidBin: len(snapraidOut) > 0, + }, nil +} + +func ValidatePaths(cfg db.StorageConfig) error { + return nil +} diff --git a/internal/storage/jobs.go b/internal/storage/jobs.go new file mode 100644 index 0000000..c7bd859 --- /dev/null +++ b/internal/storage/jobs.go @@ -0,0 +1,296 @@ +package storage + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "fmt" + "log" + "os/exec" + "sync" + "time" + + "github.com/darroyo/nasctl/internal/db" +) + +type Event struct { + Type string `json:"type"` + Line string `json:"line,omitempty"` + Status string `json:"status,omitempty"` + ExitCode int `json:"exit_code,omitempty"` + Message string `json:"message,omitempty"` +} + +type JobManager struct { + db *db.DB + mu sync.Mutex + running map[int64]context.CancelFunc + subs map[int64]map[int]chan Event + nextSub int + maxLines int + execSystem bool +} + +func NewJobManager(database *db.DB, execSystem bool) *JobManager { + return &JobManager{ + db: database, + running: make(map[int64]context.CancelFunc), + subs: make(map[int64]map[int]chan Event), + nextSub: 1, + maxLines: 500, + execSystem: execSystem, + } +} + +func (jm *JobManager) Ok() bool { + return jm.execSystem +} + +func (jm *JobManager) Subscribe(jobID int64) (<-chan Event, func()) { + jm.mu.Lock() + defer jm.mu.Unlock() + if jm.subs[jobID] == nil { + jm.subs[jobID] = make(map[int]chan Event) + } + ch := make(chan Event, 64) + id := jm.nextSub + jm.nextSub++ + jm.subs[jobID][id] = ch + unsubscribe := func() { + jm.mu.Lock() + defer jm.mu.Unlock() + delete(jm.subs[jobID], id) + close(ch) + } + return ch, unsubscribe +} + +func (jm *JobManager) broadcast(jobID int64, ev Event) { + jm.mu.Lock() + defer jm.mu.Unlock() + if subs, ok := jm.subs[jobID]; ok { + for _, ch := range subs { + select { + case ch <- ev: + default: + } + } + } +} + +func (jm *JobManager) Start(kind string, args map[string]any) (*db.StorageJob, error) { + jm.mu.Lock() + running, err := jm.db.GetRunningJob() + if err != nil { + jm.mu.Unlock() + return nil, err + } + if running != nil { + jm.mu.Unlock() + return nil, fmt.Errorf("a job is already running (id=%d, kind=%s)", running.ID, running.Kind) + } + argsJSON, err := json.Marshal(args) + if err != nil { + jm.mu.Unlock() + return nil, fmt.Errorf("marshal args: %w", err) + } + job, err := jm.db.CreateStorageJob(kind, string(argsJSON)) + if err != nil { + jm.mu.Unlock() + return nil, err + } + jm.mu.Unlock() + + go jm.runJob(job) + + return &job, nil +} + +func (jm *JobManager) runJob(job db.StorageJob) { + ctx, cancel := context.WithCancel(context.Background()) + + jm.mu.Lock() + jm.running[job.ID] = cancel + jm.mu.Unlock() + + defer func() { + jm.mu.Lock() + delete(jm.running, job.ID) + jm.mu.Unlock() + }() + + var args []string + var envcfg db.StorageConfig + var err error + + // Build command based on kind + switch job.Kind { + case "mergerfs_preview", "mergerfs_move": + envcfg, err = jm.db.GetStorageConfig() + if err != nil { + jm.failJob(job.ID, -1, fmt.Sprintf("get config: %v", err)) + return + } + cfg := MoverConfig{ + Source: envcfg.MoverSource, + Dest: envcfg.MoverDest, + CleanMacOS: envcfg.MoverCleanMacOS, + RemoveSrc: envcfg.MoverRemoveSource, + Inplace: envcfg.MoverInplace, + DryRun: job.Kind == "mergerfs_preview", + } + if envcfg.MoverRsyncOptions != "" { + flags, fe := ParseExtraRsyncFlags(envcfg.MoverRsyncOptions) + if fe != nil { + jm.failJob(job.ID, -1, fmt.Sprintf("invalid rsync flags: %v", fe)) + return + } + cfg.ExtraFlags = flags + } + if err := ValidateMoverPaths(cfg); err != nil { + jm.failJob(job.ID, -1, fmt.Sprintf("path validation: %v", err)) + return + } + if job.Kind == "mergerfs_move" && cfg.CleanMacOS { + if err := CleanMacOSJunk(cfg.Source); err != nil { + jm.logLine(job.ID, fmt.Sprintf("warning: clean macos junk: %v", err)) + } + } + args, err = BuildMoverArgs(cfg) + if err != nil { + jm.failJob(job.ID, -1, fmt.Sprintf("build rsync args: %v", err)) + return + } + + case "snapraid_diff", "snapraid_sync", "snapraid_scrub", "snapraid_check": + envcfg, err = jm.db.GetStorageConfig() + if err != nil { + jm.failJob(job.ID, -1, fmt.Sprintf("get config: %v", err)) + return + } + sc := SnapraidConfig{ + Content: envcfg.SnapraidContent, + DataDirs: ParseSnapraidDataDirs(envcfg.SnapraidDataDirs), + ParityDir: envcfg.SnapraidParityDir, + ScrubPlan: envcfg.SnapraidScrubPlan, + } + if err := ValidateSnapraidContent(sc.Content); err != nil { + jm.failJob(job.ID, -1, fmt.Sprintf("content validation: %v", err)) + return + } + if job.Kind == "snapraid_scrub" { + if err := ValidateScrubPlan(sc.ScrubPlan); err != nil { + jm.failJob(job.ID, -1, fmt.Sprintf("scrub plan validation: %v", err)) + return + } + } + args, err = BuildSnapraidArgs(job.Kind, sc) + if err != nil { + jm.failJob(job.ID, -1, fmt.Sprintf("build snapraid args: %v", err)) + return + } + + default: + jm.failJob(job.ID, -1, fmt.Sprintf("unknown job kind: %s", job.Kind)) + return + } + + cmd := exec.CommandContext(ctx, args[0], args[1:]...) + var outBuf, errBuf bytes.Buffer + cmd.Stdout = &outBuf + cmd.Stderr = &errBuf + + // Mark running and get PID + jid := job.ID + pid := 0 + if cmd.Process != nil { + pid = cmd.Process.Pid + } + _ = jm.db.MarkJobRunning(jid, pid) + jm.broadcast(jid, Event{Type: "status", Status: "running"}) + + err = cmd.Run() + + // Collect all output + combined := outBuf.String() + errBuf.String() + scanner := bufio.NewScanner(bufio.NewReader(bytes.NewReader([]byte(combined)))) + lineCount := 0 + flushTicker := time.NewTicker(1 * time.Second) + defer flushTicker.Stop() + + for scanner.Scan() { + line := scanner.Text() + lineCount++ + jm.logLine(jid, line) + if lineCount%25 == 0 { + _ = jm.db.TrimJobOutput(jid, jm.maxLines) + } + } + + exitCode := -1 + if err != nil { + if exitErr, ok := err.(*exec.ExitError); ok { + exitCode = exitErr.ExitCode() + } else { + jm.failJob(jid, exitCode, fmt.Sprintf("execution error: %v", err)) + return + } + } + + _ = jm.db.TrimJobOutput(jid, jm.maxLines) + + status := "success" + errMsg := "" + if exitCode != 0 { + status = "failed" + errMsg = fmt.Sprintf("exit code %d", exitCode) + } + + if err := jm.db.MarkJobFinished(jid, status, exitCode, combined, errMsg); err != nil { + log.Printf("mark job finished: %v", err) + } + + jm.broadcast(jid, Event{Type: "end", Status: status, ExitCode: exitCode}) +} + +func (jm *JobManager) logLine(jobID int64, line string) { + _ = jm.db.AppendJobOutput(jobID, line) + jm.broadcast(jobID, Event{Type: "line", Line: line}) +} + +func (jm *JobManager) failJob(jobID int64, exitCode int, msg string) { + _ = jm.db.MarkJobFinished(jobID, "failed", exitCode, "", msg) + jm.broadcast(jobID, Event{Type: "end", Status: "failed", ExitCode: exitCode, Message: msg}) +} + +func (jm *JobManager) Get(id int64) (db.StorageJob, error) { + return jm.db.GetStorageJob(id) +} + +func (jm *JobManager) List(limit int) ([]db.StorageJob, error) { + return jm.db.ListStorageJobs(limit) +} + +func (jm *JobManager) Cancel(id int64) error { + jm.mu.Lock() + cancel, ok := jm.running[id] + jm.mu.Unlock() + if !ok { + job, err := jm.db.GetStorageJob(id) + if err != nil { + return err + } + if job.Status == "queued" { + return jm.db.SetJobCancelled(id) + } + return fmt.Errorf("job %d is not running", id) + } + cancel() + _ = jm.db.SetJobCancelled(id) + return nil +} + +func (jm *JobManager) ResetOrphans() error { + return jm.db.ResetOrphanedJobs() +} diff --git a/internal/storage/jobs_test.go b/internal/storage/jobs_test.go new file mode 100644 index 0000000..7add1f7 --- /dev/null +++ b/internal/storage/jobs_test.go @@ -0,0 +1,139 @@ +package storage + +import ( + "testing" + "time" + + "github.com/darroyo/nasctl/internal/db" +) + +func TestJobManagerStartRejectsConcurrent(t *testing.T) { + d, err := db.Open(":memory:") + if err != nil { + t.Fatalf("open db: %v", err) + } + defer d.Close() + if err := d.Migrate(); err != nil { + t.Fatalf("migrate: %v", err) + } + + jm := NewJobManager(d, true) + + // Start a job that will block (no cmd specified — we just test rejection logic) + _, err = jm.Start("snapraid_diff", nil) + if err != nil { + t.Fatalf("first Start: %v", err) + } + + // Second start should be rejected + _, err = jm.Start("snapraid_sync", nil) + if err == nil { + t.Error("second Start should have been rejected (job already running)") + } +} + +func TestJobManagerOk(t *testing.T) { + d, _ := db.Open(":memory:") + d.Migrate() + defer d.Close() + + jmOn := NewJobManager(d, true) + if !jmOn.Ok() { + t.Error("Ok() = false, want true") + } + + jmOff := NewJobManager(d, false) + if jmOff.Ok() { + t.Error("Ok() = true, want false") + } +} + +func TestJobManagerListEmpty(t *testing.T) { + d, _ := db.Open(":memory:") + d.Migrate() + defer d.Close() + + jm := NewJobManager(d, false) + jobs, err := jm.List(10) + if err != nil { + t.Fatalf("List: %v", err) + } + if len(jobs) != 0 { + t.Errorf("List = %d jobs, want 0", len(jobs)) + } +} + +func TestJobManagerResetOrphans(t *testing.T) { + d, _ := db.Open(":memory:") + d.Migrate() + defer d.Close() + + jm := NewJobManager(d, false) + + // Create a running job directly in DB + job, err := d.CreateStorageJob("snapraid_sync", "{}") + if err != nil { + t.Fatalf("CreateStorageJob: %v", err) + } + // Simulate orphaned running job + _ = d.MarkJobRunning(job.ID, 12345) + + // Create a queued job + _, _ = d.CreateStorageJob("snapraid_diff", "{}") + + if err := jm.ResetOrphans(); err != nil { + t.Fatalf("ResetOrphans: %v", err) + } + + jobs, _ := jm.List(10) + runningCount := 0 + queuedCount := 0 + for _, j := range jobs { + if j.Status == "running" { + runningCount++ + } + if j.Status == "queued" { + queuedCount++ + } + if j.Status == "interrupted" { + // should happen for the orphaned ones + } + } + + if runningCount != 0 || queuedCount != 0 { + t.Errorf("after ResetOrphans: running=%d queued=%d, want both 0", runningCount, queuedCount) + } +} + +func TestJobManagerSubscribeAndBroadcast(t *testing.T) { + d, _ := db.Open(":memory:") + d.Migrate() + defer d.Close() + + jm := NewJobManager(d, true) + + events, unsub := jm.Subscribe(9999) + defer unsub() + + jm.broadcast(9999, Event{Type: "line", Line: "hello"}) + jm.broadcast(9999, Event{Type: "end", Status: "success"}) + + select { + case ev := <-events: + if ev.Type != "line" || ev.Line != "hello" { + t.Errorf("got event %+v, want {Type:line Line:hello}", ev) + } + case <-time.After(100 * time.Millisecond): + t.Error("timeout waiting for event") + } + + // second event + select { + case ev := <-events: + if ev.Type != "end" || ev.Status != "success" { + t.Errorf("got event %+v, want {Type:end Status:success}", ev) + } + case <-time.After(100 * time.Millisecond): + t.Error("timeout waiting for end event") + } +} diff --git a/internal/storage/mergerfs.go b/internal/storage/mergerfs.go new file mode 100644 index 0000000..6626aea --- /dev/null +++ b/internal/storage/mergerfs.go @@ -0,0 +1,116 @@ +package storage + +import ( + "fmt" + "os" + "os/exec" + "regexp" + "strings" +) + +var forbiddenFlagChars = regexp.MustCompile("[;&|$`\"<>\r\n\x00]") + +type MoverConfig struct { + Source string + Dest string + CleanMacOS bool + RemoveSrc bool + Inplace bool + ExtraFlags []string + DryRun bool +} + +func BuildPreviewArgs(cfg MoverConfig) ([]string, error) { + if cfg.Source == "" || cfg.Dest == "" { + return nil, fmt.Errorf("source and dest are required") + } + args := []string{ + "rsync", + "-an", + "--remove-source-files", + "--out-format=%n", + cfg.Source + "/", + cfg.Dest + "/", + } + return args, nil +} + +func BuildMoverArgs(cfg MoverConfig) ([]string, error) { + if cfg.Source == "" || cfg.Dest == "" { + return nil, fmt.Errorf("source and dest are required") + } + args := []string{"rsync", "-aAXv"} + if !cfg.Inplace { + args = append(args, "--inplace") + } + args = append(args, "--remove-source-files") + if !cfg.DryRun { + args = append(args, "--progress", "--info=progress2") + } + args = append(args, "--no-compress") + args = append(args, cfg.ExtraFlags...) + args = append(args, "--info=progress2") + args = append(args, cfg.Source+"/", cfg.Dest+"/") + return args, nil +} + +func CleanMacOSJunk(root string) error { + for _, pattern := range []string{".DS_Store", "._*"} { + cmd := exec.Command("find", root, "-name", pattern, "-delete") + if err := cmd.Run(); err != nil { + return fmt.Errorf("clean junk %s: %w", pattern, err) + } + } + return nil +} + +func RemoveEmptyDirs(root string) error { + cmd := exec.Command("find", root, "-mindepth", "1", "-type", "d", "-empty", "-delete") + if err := cmd.Run(); err != nil { + return fmt.Errorf("remove empty dirs: %w", err) + } + return nil +} + +func ValidateMoverPaths(cfg MoverConfig) error { + if cfg.Source == "" { + return fmt.Errorf("source path is required") + } + if cfg.Dest == "" { + return fmt.Errorf("destination path is required") + } + info, err := os.Stat(cfg.Source) + if err != nil { + return fmt.Errorf("source %s: %w", cfg.Source, err) + } + if !info.IsDir() { + return fmt.Errorf("source %s: not a directory", cfg.Source) + } + info, err = os.Stat(cfg.Dest) + if err != nil { + return fmt.Errorf("destination %s: %w", cfg.Dest, err) + } + if !info.IsDir() { + return fmt.Errorf("destination %s: not a directory", cfg.Dest) + } + return nil +} + +func ParseExtraRsyncFlags(raw string) ([]string, error) { + var out []string + lines := strings.Split(raw, "\n") + for i, line := range lines { + line = strings.TrimSpace(line) + if line == "" || strings.HasPrefix(line, "#") { + continue + } + if !strings.HasPrefix(line, "-") { + return nil, fmt.Errorf("line %d: flag must start with '-' (%q)", i+1, line) + } + if forbiddenFlagChars.MatchString(line) { + return nil, fmt.Errorf("line %d: flag contains forbidden characters (%q)", i+1, line) + } + out = append(out, line) + } + return out, nil +} diff --git a/internal/storage/mergerfs_test.go b/internal/storage/mergerfs_test.go new file mode 100644 index 0000000..806b61e --- /dev/null +++ b/internal/storage/mergerfs_test.go @@ -0,0 +1,113 @@ +package storage + +import "testing" + +func TestBuildPreviewArgs(t *testing.T) { + cfg := MoverConfig{Source: "/mnt/ssd", Dest: "/mnt/pool"} + args, err := BuildPreviewArgs(cfg) + if err != nil { + t.Fatalf("BuildPreviewArgs: %v", err) + } + if args[0] != "rsync" { + t.Errorf("args[0] = %q, want rsync", args[0]) + } + if args[1] != "-an" { + t.Errorf("args[1] = %q, want -an", args[1]) + } + if args[2] != "--remove-source-files" { + t.Errorf("args[2] = %q, want --remove-source-files", args[2]) + } + if args[3] != "--out-format=%n" { + t.Errorf("args[3] = %q, want --out-format=%%n", args[3]) + } + if args[4] != "/mnt/ssd/" { + t.Errorf("args[4] = %q, want /mnt/ssd/", args[4]) + } + if args[5] != "/mnt/pool/" { + t.Errorf("args[5] = %q, want /mnt/pool/", args[5]) + } +} + +func TestBuildMoverArgs(t *testing.T) { + cfg := MoverConfig{Source: "/mnt/ssd", Dest: "/mnt/pool", Inplace: false, RemoveSrc: true} + args, err := BuildMoverArgs(cfg) + if err != nil { + t.Fatalf("BuildMoverArgs: %v", err) + } + if args[0] != "rsync" { + t.Errorf("args[0] = %q, want rsync", args[0]) + } + foundRemove := false + for _, a := range args { + if a == "--remove-source-files" { + foundRemove = true + } + } + if !foundRemove { + t.Errorf("--remove-source-files not found in args %v", args) + } +} + +func TestBuildMoverArgsExtraFlags(t *testing.T) { + cfg := MoverConfig{Source: "/src", Dest: "/dst", ExtraFlags: []string{"--exclude=*.tmp", "--max-size=2G"}} + args, err := BuildMoverArgs(cfg) + if err != nil { + t.Fatalf("BuildMoverArgs with extra flags: %v", err) + } + if args[0] != "rsync" { + t.Errorf("args[0] = %q, want rsync", args[0]) + } + idx := -1 + for i, a := range args { + if a == "--exclude=*.tmp" { + idx = i + } + } + if idx == -1 { + t.Errorf("--exclude=*.tmp not found in args %v", args) + } +} + +func TestParseExtraRsyncFlags(t *testing.T) { + tests := []struct { + name string + input string + want []string + wantErr bool + }{ + {"empty", "", nil, false}, + {"single flag", "--exclude=*.tmp", []string{"--exclude=*.tmp"}, false}, + {"multiple flags", "--exclude=*.tmp\n--max-size=2G", []string{"--exclude=*.tmp", "--max-size=2G"}, false}, + {"comment and empty", "# comment\n\n--exclude=*.tmp\n", []string{"--exclude=*.tmp"}, false}, + {"missing dash", "exclude=*.tmp", nil, true}, + {"forbidden char semicolon", "--flag;echo", nil, true}, + {"forbidden char pipe", "--flag|grep", nil, true}, + {"forbidden char dollar", "--flag$var", nil, true}, + {"forbidden char backtick", "--flag`cmd`", nil, true}, + {"newline in flag", "--flag\nline", nil, true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := ParseExtraRsyncFlags(tt.input) + if tt.wantErr { + if err == nil { + t.Errorf("ParseExtraRsyncFlags(%q) = %v, want error", tt.input, got) + } + return + } + if err != nil { + t.Errorf("unexpected error: %v", err) + return + } + if len(got) != len(tt.want) { + t.Errorf("ParseExtraRsyncFlags(%q) = %v, want %v", tt.input, got, tt.want) + return + } + for i := range got { + if got[i] != tt.want[i] { + t.Errorf("ParseExtraRsyncFlags(%q)[%d] = %q, want %q", tt.input, i, got[i], tt.want[i]) + } + } + }) + } +} diff --git a/internal/storage/snapraid.go b/internal/storage/snapraid.go new file mode 100644 index 0000000..29928df --- /dev/null +++ b/internal/storage/snapraid.go @@ -0,0 +1,70 @@ +package storage + +import ( + "fmt" + "os" + "strconv" + "strings" +) + +type SnapraidConfig struct { + Content string + DataDirs []string + ParityDir string + ScrubPlan int +} + +func BuildSnapraidArgs(kind string, cfg SnapraidConfig) ([]string, error) { + if cfg.Content == "" { + return nil, fmt.Errorf("snapraid content file is required") + } + args := []string{"snapraid", "-c", cfg.Content} + switch kind { + case "snapraid_diff": + args = append(args, "diff") + case "snapraid_sync": + args = append(args, "sync") + case "snapraid_scrub": + args = append(args, "scrub", "-p", strconv.Itoa(cfg.ScrubPlan)) + case "snapraid_check": + args = append(args, "check") + default: + return nil, fmt.Errorf("unknown snapraid kind: %s", kind) + } + return args, nil +} + +func ValidateSnapraidContent(content string) error { + if content == "" { + return fmt.Errorf("snapraid content file is required") + } + info, err := os.Stat(content) + if err != nil { + return fmt.Errorf("snapraid content %s: %w", content, err) + } + if info.IsDir() { + return fmt.Errorf("snapraid content %s: is a directory, not a file", content) + } + return nil +} + +func ValidateScrubPlan(plan int) error { + if plan < 1 || plan > 99 { + return fmt.Errorf("scrub plan must be between 1 and 99, got %d", plan) + } + return nil +} + +func ParseSnapraidDataDirs(raw string) []string { + if raw == "" { + return nil + } + var dirs []string + for _, d := range strings.Split(raw, ",") { + d = strings.TrimSpace(d) + if d != "" { + dirs = append(dirs, d) + } + } + return dirs +} diff --git a/internal/storage/snapraid_test.go b/internal/storage/snapraid_test.go new file mode 100644 index 0000000..8afcbc2 --- /dev/null +++ b/internal/storage/snapraid_test.go @@ -0,0 +1,104 @@ +package storage + +import "testing" + +func TestBuildSnapraidArgs(t *testing.T) { + tests := []struct { + kind string + cfg SnapraidConfig + want []string + }{ + { + "snapraid_diff", + SnapraidConfig{Content: "/pool/snapraid.content"}, + []string{"snapraid", "-c", "/pool/snapraid.content", "diff"}, + }, + { + "snapraid_sync", + SnapraidConfig{Content: "/pool/snapraid.content"}, + []string{"snapraid", "-c", "/pool/snapraid.content", "sync"}, + }, + { + "snapraid_check", + SnapraidConfig{Content: "/pool/snapraid.content"}, + []string{"snapraid", "-c", "/pool/snapraid.content", "check"}, + }, + { + "snapraid_scrub", + SnapraidConfig{Content: "/pool/snapraid.content", ScrubPlan: 8}, + []string{"snapraid", "-c", "/pool/snapraid.content", "scrub", "-p", "8"}, + }, + } + for _, tt := range tests { + t.Run(tt.kind, func(t *testing.T) { + got, err := BuildSnapraidArgs(tt.kind, tt.cfg) + if err != nil { + t.Fatalf("BuildSnapraidArgs: %v", err) + } + if len(got) != len(tt.want) { + t.Errorf("BuildSnapraidArgs = %v, want %v", got, tt.want) + return + } + for i := range got { + if got[i] != tt.want[i] { + t.Errorf("BuildSnapraidArgs[%d] = %q, want %q", i, got[i], tt.want[i]) + } + } + }) + } +} + +func TestBuildSnapraidArgsNoContent(t *testing.T) { + _, err := BuildSnapraidArgs("snapraid_sync", SnapraidConfig{Content: ""}) + if err == nil { + t.Error("expected error for empty content") + } +} + +func TestValidateScrubPlan(t *testing.T) { + tests := []struct { + plan int + wantErr bool + }{ + {0, true}, + {-1, true}, + {100, true}, + {1, false}, + {50, false}, + {99, false}, + } + for _, tt := range tests { + err := ValidateScrubPlan(tt.plan) + if tt.wantErr && err == nil { + t.Errorf("ValidateScrubPlan(%d) = nil, want error", tt.plan) + } + if !tt.wantErr && err != nil { + t.Errorf("ValidateScrubPlan(%d) = %v, want nil", tt.plan, err) + } + } +} + +func TestParseSnapraidDataDirs(t *testing.T) { + tests := []struct { + input string + want []string + }{ + {"", nil}, + {"/disk1", []string{"/disk1"}}, + {"/disk1,/disk2", []string{"/disk1", "/disk2"}}, + {"/disk1, /disk2 , /disk3", []string{"/disk1", "/disk2", "/disk3"}}, + {" /disk1 , /disk2 ", []string{"/disk1", "/disk2"}}, + } + for _, tt := range tests { + got := ParseSnapraidDataDirs(tt.input) + if len(got) != len(tt.want) { + t.Errorf("ParseSnapraidDataDirs(%q) = %v, want %v", tt.input, got, tt.want) + continue + } + for i := range got { + if got[i] != tt.want[i] { + t.Errorf("ParseSnapraidDataDirs(%q)[%d] = %q, want %q", tt.input, i, got[i], tt.want[i]) + } + } + } +} diff --git a/internal/web/auth.go b/internal/web/auth.go index dc7de80..2c63060 100644 --- a/internal/web/auth.go +++ b/internal/web/auth.go @@ -17,6 +17,7 @@ import ( "github.com/darroyo/nasctl/internal/db" "github.com/darroyo/nasctl/internal/engine" + "github.com/darroyo/nasctl/internal/storage" ) const ( @@ -199,6 +200,7 @@ type Server struct { AdminUsername string UploadMaxBytes int64 PreviewMaxBytes int64 + JM *storage.JobManager } type Options struct { @@ -209,6 +211,7 @@ type Options struct { AdminUsername string UploadMaxBytes int64 PreviewMaxBytes int64 + JobManager *storage.JobManager } func NewServer(database *db.DB, eng *engine.Engine, opts Options) *Server { @@ -228,6 +231,7 @@ func NewServer(database *db.DB, eng *engine.Engine, opts Options) *Server { AdminUsername: opts.AdminUsername, UploadMaxBytes: opts.UploadMaxBytes, PreviewMaxBytes: opts.PreviewMaxBytes, + JM: opts.JobManager, } } diff --git a/internal/web/handlers_storage.go b/internal/web/handlers_storage.go new file mode 100644 index 0000000..02fc8ac --- /dev/null +++ b/internal/web/handlers_storage.go @@ -0,0 +1,233 @@ +package web + +import ( + "encoding/json" + "fmt" + "io" + "net/http" + "strconv" + "strings" + "time" + + "github.com/go-chi/chi/v5" + "github.com/darroyo/nasctl/internal/db" + "github.com/darroyo/nasctl/internal/storage" +) + +const jobStreamKeepalive = 15 * time.Second + +func (s *Server) handleStorageCapabilities(w http.ResponseWriter, r *http.Request) { + caps, err := storage.Probe(r.Context()) + if err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + writeJSON(w, http.StatusOK, caps) +} + +func (s *Server) handleStorageGetConfig(w http.ResponseWriter, r *http.Request) { + cfg, err := s.DB.GetStorageConfig() + if err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + writeJSON(w, http.StatusOK, cfg) +} + +func (s *Server) handleStorageUpdateConfig(w http.ResponseWriter, r *http.Request) { + var cfg db.StorageConfig + defer r.Body.Close() + if err := json.NewDecoder(r.Body).Decode(&cfg); err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + if err := storage.ValidateScrubPlan(cfg.SnapraidScrubPlan); err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + flags, err := storage.ParseExtraRsyncFlags(cfg.MoverRsyncOptions) + if err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + _ = flags + if cfg.MoverSource != "" { + if err := validateAbsPath(cfg.MoverSource); err != nil { + writeError(w, http.StatusBadRequest, fmt.Sprintf("mover_source: %v", err)) + return + } + } + if cfg.MoverDest != "" { + if err := validateAbsPath(cfg.MoverDest); err != nil { + writeError(w, http.StatusBadRequest, fmt.Sprintf("mover_dest: %v", err)) + return + } + } + if cfg.SnapraidContent != "" { + if err := validateAbsPath(cfg.SnapraidContent); err != nil { + writeError(w, http.StatusBadRequest, fmt.Sprintf("snapraid_content: %v", err)) + return + } + } + if err := s.DB.UpdateStorageConfig(cfg); err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + writeJSON(w, http.StatusOK, cfg) +} + +func (s *Server) handleStorageListJobs(w http.ResponseWriter, r *http.Request) { + limit, _ := strconv.Atoi(r.URL.Query().Get("limit")) + if limit <= 0 { + limit = 50 + } + jobs, err := s.JM.List(limit) + if err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + writeJSON(w, http.StatusOK, map[string]any{"jobs": jobs}) +} + +func (s *Server) handleStorageGetJob(w http.ResponseWriter, r *http.Request) { + id, err := strconv.ParseInt(chi.URLParam(r, "id"), 10, 64) + if err != nil { + writeError(w, http.StatusBadRequest, "invalid job id") + return + } + job, err := s.JM.Get(id) + if err != nil { + writeError(w, http.StatusNotFound, err.Error()) + return + } + writeJSON(w, http.StatusOK, job) +} + +func (s *Server) handleStorageStartJob(w http.ResponseWriter, r *http.Request) { + if !s.JM.Ok() { + writeError(w, http.StatusServiceUnavailable, "storage operations disabled (NASCTL_EXEC_SYSTEM=false)") + return + } + var req struct { + Kind string `json:"kind"` + Args map[string]any `json:"args"` + } + defer r.Body.Close() + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + validKinds := map[string]bool{ + "mergerfs_preview": true, + "mergerfs_move": true, + "snapraid_diff": true, + "snapraid_sync": true, + "snapraid_scrub": true, + "snapraid_check": true, + } + if !validKinds[req.Kind] { + writeError(w, http.StatusBadRequest, fmt.Sprintf("unknown kind: %s", req.Kind)) + return + } + job, err := s.JM.Start(req.Kind, req.Args) + if err != nil { + if strings.Contains(err.Error(), "already running") { + writeError(w, http.StatusConflict, err.Error()) + return + } + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + writeJSON(w, http.StatusCreated, job) +} + +func (s *Server) handleStorageCancelJob(w http.ResponseWriter, r *http.Request) { + id, err := strconv.ParseInt(chi.URLParam(r, "id"), 10, 64) + if err != nil { + writeError(w, http.StatusBadRequest, "invalid job id") + return + } + if err := s.JM.Cancel(id); err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + writeJSON(w, http.StatusOK, map[string]bool{"ok": true}) +} + +func (s *Server) handleStorageJobStream(w http.ResponseWriter, r *http.Request) { + id, err := strconv.ParseInt(chi.URLParam(r, "id"), 10, 64) + if err != nil { + writeError(w, http.StatusBadRequest, "invalid job id") + return + } + + flusher, ok := w.(http.Flusher) + if !ok { + writeError(w, http.StatusInternalServerError, "streaming not supported") + return + } + + setSSEHeaders(w) + job, err := s.JM.Get(id) + if err != nil { + sseEmit(w, "error", map[string]string{"message": err.Error()}) + flusher.Flush() + return + } + if job.Status != "running" && job.Status != "queued" { + sseEmit(w, "end", map[string]any{"status": job.Status, "exit_code": job.ExitCode, "output": job.Output}) + flusher.Flush() + return + } + + events, unsub := s.JM.Subscribe(id) + defer unsub() + + ctx := r.Context() + keepalive := time.NewTicker(jobStreamKeepalive) + defer keepalive.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-keepalive.C: + fmt.Fprintf(w, ": keepalive\n\n") + flusher.Flush() + case ev, ok := <-events: + if !ok { + return + } + sseEmit(w, ev.Type, ev) + flusher.Flush() + if ev.Type == "end" || ev.Type == "error" { + return + } + } + } +} + +func setSSEHeaders(w http.ResponseWriter) { + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-cache") + w.Header().Set("Connection", "keep-alive") + w.Header().Set("X-Accel-Buffering", "no") +} + +func sseEmit(w io.Writer, event string, data any) { + body, _ := json.Marshal(data) + fmt.Fprintf(w, "event: %s\ndata: %s\n\n", event, string(body)) +} + +func validateAbsPath(path string) error { + if path == "" { + return nil + } + if !strings.HasPrefix(path, "/") { + return fmt.Errorf("must be absolute") + } + if strings.Contains(path, "..") { + return fmt.Errorf("must not contain ..") + } + return nil +} diff --git a/internal/web/router.go b/internal/web/router.go index db871b6..efd83db 100644 --- a/internal/web/router.go +++ b/internal/web/router.go @@ -86,6 +86,21 @@ func NewRouter(s *Server) chi.Router { files.Get("/preview", s.handlePreview) files.Get("/search", s.handleSearch) }) + + protected.Route("/storage", func(storageRouter chi.Router) { + storageRouter.Get("/capabilities", s.handleStorageCapabilities) + storageRouter.Get("/config", s.handleStorageGetConfig) + storageRouter.Put("/config", s.handleStorageUpdateConfig) + storageRouter.Route("/jobs", func(j chi.Router) { + j.Get("/", s.handleStorageListJobs) + j.Post("/", s.handleStorageStartJob) + j.Route("/{id}", func(item chi.Router) { + item.Get("/", s.handleStorageGetJob) + item.Post("/cancel", s.handleStorageCancelJob) + item.Get("/stream", s.handleStorageJobStream) + }) + }) + }) }) }) diff --git a/web/src/App.tsx b/web/src/App.tsx index 602ba3f..c1ed771 100644 --- a/web/src/App.tsx +++ b/web/src/App.tsx @@ -11,6 +11,7 @@ import Nfs from "./pages/Nfs"; import Log from "./pages/Log"; import Settings from "./pages/Settings"; import Files from "./pages/Files"; +import Storage from "./pages/Storage"; type AuthState = { loading: boolean; authenticated: boolean; username: string }; @@ -56,6 +57,7 @@ export default function App() { } /> } /> } /> + } /> } /> } /> } /> diff --git a/web/src/api.ts b/web/src/api.ts index 2d909c1..8c2fde6 100644 --- a/web/src/api.ts +++ b/web/src/api.ts @@ -146,6 +146,58 @@ export interface FilePreview { size: number; } +export type JobKind = + | "mergerfs_preview" + | "mergerfs_move" + | "snapraid_diff" + | "snapraid_sync" + | "snapraid_scrub" + | "snapraid_check"; + +export interface StorageCapabilities { + rsync: boolean; + mergerfs: boolean; + mergerfs_mounted: boolean; + snapraid: boolean; +} + +export interface StorageConfig { + id: number; + mover_source: string; + mover_dest: string; + mover_clean_macos: boolean; + mover_remove_source: boolean; + mover_inplace: boolean; + mover_rsync_options: string; + snapraid_content: string; + snapraid_data_dirs: string; + snapraid_parity_dir: string; + snapraid_scrub_plan: number; + updated_at: string; +} + +export interface StorageJob { + id: number; + kind: JobKind; + status: "queued" | "running" | "success" | "failed" | "cancelled" | "interrupted"; + args_json: string; + pid: number; + started_at: string | null; + finished_at: string | null; + exit_code: number; + output: string; + error: string; + created_at: string; +} + +export interface StorageEvent { + type: "line" | "status" | "end" | "error"; + line?: string; + status?: string; + exit_code?: number; + message?: string; +} + export class ApiError extends Error { status: number; constructor(status: number, message: string) { @@ -243,6 +295,17 @@ export const api = { searchFiles: (path: string, q: string, limit = 100) => request<{ results: SearchHit[] }>(`GET`, `/files/search?path=${encodeURIComponent(path)}&q=${encodeURIComponent(q)}&limit=${limit}`), fileDownloadUrl: (path: string) => `/api/files/download?path=${encodeURIComponent(path)}`, + + // storage + storageCapabilities: () => request("GET", "/storage/capabilities"), + getStorageConfig: () => request("GET", "/storage/config"), + updateStorageConfig: (c: Partial) => request("PUT", "/storage/config", c), + listStorageJobs: (limit = 50) => request<{ jobs: StorageJob[] }>("GET", `/storage/jobs?limit=${limit}`), + getStorageJob: (id: number) => request("GET", `/storage/jobs/${id}`), + startStorageJob: (kind: JobKind, args?: Record) => + request("POST", "/storage/jobs", { kind, args }), + cancelStorageJob: (id: number) => request<{ ok: boolean }>("POST", `/storage/jobs/${id}/cancel`), + storageJobStreamUrl: (id: number) => `/api/storage/jobs/${id}/stream`, }; export function formatBytes(bytes: number): string { diff --git a/web/src/components/Layout.tsx b/web/src/components/Layout.tsx index b101283..3c44d1d 100644 --- a/web/src/components/Layout.tsx +++ b/web/src/components/Layout.tsx @@ -9,6 +9,7 @@ const navItems = [ { to: "/files", label: "Archivos" }, { to: "/samba", label: "SMB / Samba" }, { to: "/nfs", label: "NFS" }, + { to: "/storage", label: "Almacenamiento" }, { to: "/log", label: "Historial" }, { to: "/settings", label: "Ajustes" }, ]; diff --git a/web/src/pages/Storage.tsx b/web/src/pages/Storage.tsx new file mode 100644 index 0000000..f376c8a --- /dev/null +++ b/web/src/pages/Storage.tsx @@ -0,0 +1,563 @@ +import React, { useEffect, useRef, useState } from "react"; +import { + api, + StorageCapabilities, + StorageConfig, + StorageJob, + StorageEvent, + JobKind, +} from "../api"; + +const KIND_LABELS: Record = { + mergerfs_preview: "Mergerfs — Vista previa", + mergerfs_move: "Mergerfs — Mover", + snapraid_diff: "SnapRAID — Diff", + snapraid_sync: "SnapRAID — Sync", + snapraid_scrub: "SnapRAID — Scrub", + snapraid_check: "SnapRAID — Check", +}; + +const STATUS_COLORS: Record = { + queued: "bg-slate-600 text-slate-300", + running: "bg-blue-600 text-blue-100 animate-pulse", + success: "bg-emerald-600 text-emerald-100", + failed: "bg-red-600 text-red-100", + cancelled: "bg-amber-600 text-amber-100", + interrupted: "bg-orange-600 text-orange-100", +}; + +export default function Storage() { + const [caps, setCaps] = useState(null); + const [config, setConfig] = useState(null); + const [jobs, setJobs] = useState([]); + const [activeJob, setActiveJob] = useState(null); + const [outputLines, setOutputLines] = useState([]); + const [saving, setSaving] = useState(false); + const [scrubPlan, setScrubPlan] = useState(8); + const [scrubPlanInput, setScrubPlanInput] = useState("8"); + const [dirtyConfig, setDirtyConfig] = useState(false); + const [pendingConfig, setPendingConfig] = useState>({}); + const outputRef = useRef(null); + const esRef = useRef(null); + + useEffect(() => { + loadCaps(); + loadConfig(); + loadJobs(); + }, []); + + async function loadCaps() { + try { + const c = await api.storageCapabilities(); + setCaps(c); + } catch (e) { + console.error(e); + } + } + + async function loadConfig() { + try { + const c = await api.getStorageConfig(); + setConfig(c); + setPendingConfig({}); + setDirtyConfig(false); + setScrubPlan(c.snapraid_scrub_plan); + setScrubPlanInput(String(c.snapraid_scrub_plan)); + } catch (e) { + console.error(e); + } + } + + async function loadJobs() { + try { + const res = await api.listStorageJobs(50); + setJobs(res.jobs); + } catch (e) { + console.error(e); + } + } + + function handleConfigChange(field: keyof StorageConfig, value: unknown) { + setPendingConfig((prev) => ({ ...prev, [field]: value })); + setDirtyConfig(true); + } + + async function saveConfig() { + if (!config) return; + setSaving(true); + try { + const merged = { ...config, ...pendingConfig }; + const updated = await api.updateStorageConfig(merged); + setConfig(updated); + setPendingConfig({}); + setDirtyConfig(false); + setScrubPlan(updated.snapraid_scrub_plan); + setScrubPlanInput(String(updated.snapraid_scrub_plan)); + } catch (e) { + alert(`Error saving: ${e}`); + } finally { + setSaving(false); + } + } + + async function startJob(kind: JobKind) { + try { + const job = await api.startStorageJob(kind); + setActiveJob(job); + setOutputLines([]); + subscribeStream(job.id); + await loadJobs(); + } catch (e: unknown) { + if (e && typeof e === "object" && "status" in e && (e as { status: number }).status === 409) { + alert("Ya hay una operación en curso. Espera a que termine."); + } else { + alert(`Error: ${e}`); + } + } + } + + function subscribeStream(jobId: number) { + if (esRef.current) { + esRef.current.close(); + } + const es = new EventSource(api.storageJobStreamUrl(jobId)); + esRef.current = es; + + es.addEventListener("line", (e) => { + const ev: StorageEvent = JSON.parse(e.data); + if (ev.line !== undefined) { + setOutputLines((prev) => [...prev, ev.line as string].slice(-500)); + } + }); + + es.addEventListener("end", async (e) => { + const ev: StorageEvent = JSON.parse(e.data); + setActiveJob((prev) => + prev ? { ...prev, status: ev.status as StorageJob["status"], exit_code: ev.exit_code ?? prev.exit_code } : prev + ); + es.close(); + esRef.current = null; + await loadJobs(); + }); + + es.addEventListener("error", (e) => { + console.error("SSE error", e); + es.close(); + }); + } + + async function cancelJob() { + if (!activeJob) return; + try { + await api.cancelStorageJob(activeJob.id); + } catch (e) { + console.error(e); + } + } + + useEffect(() => { + if (outputRef.current) { + outputRef.current.scrollTop = outputRef.current.scrollHeight; + } + }, [outputLines]); + + useEffect(() => { + return () => { + if (esRef.current) esRef.current.close(); + }; + }, []); + + const mergedConfig: StorageConfig = config + ? { ...config, ...pendingConfig } + : ({} as StorageConfig); + + return ( +
+

Almacenamiento

+ + {/* Capabilities */} +
+

+ Capacidades detectadas +

+
+ + + + +
+ {caps && !caps.rsync && ( +

+ Instala rsync para usar el mover de archivos. +

+ )} +
+ + {/* Config */} +
+
+

+ Configuración del pool +

+ +
+ +
+
+ Mergerfs Mover + handleConfigChange("mover_source", v)} + placeholder="/mnt/disks/ssd1" + /> + handleConfigChange("mover_dest", v)} + placeholder="/mnt/pool" + /> + + + +
+ +