package web import ( "encoding/json" "fmt" "io" "net/http" "os" "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 } if cfg.SnapraidConf == "" { for _, path := range []string{"/etc/snapraid.conf", "/usr/local/etc/snapraid.conf"} { if info, err := os.Stat(path); err == nil && !info.IsDir() { cfg.SnapraidConf = path break } } } writeJSON(w, http.StatusOK, cfg) } func (s *Server) handleStorageGetConfFile(w http.ResponseWriter, r *http.Request) { cfg, err := s.DB.GetStorageConfig() if err != nil { writeError(w, http.StatusInternalServerError, err.Error()) return } if cfg.SnapraidConf == "" { writeError(w, http.StatusBadRequest, "snapraid conf path not configured") return } data, err := os.ReadFile(cfg.SnapraidConf) if err != nil { writeError(w, http.StatusInternalServerError, fmt.Sprintf("read %s: %v", cfg.SnapraidConf, err)) return } w.Header().Set("Content-Type", "text/plain; charset=utf-8") w.Write(data) } 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 } if cfg.MoverWarningThreshold < 1 || cfg.MoverWarningThreshold > 99 { writeError(w, http.StatusBadRequest, "mover_warning_threshold must be between 1 and 99") 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.SnapraidConf != "" { if err := validateAbsPath(cfg.SnapraidConf); err != nil { writeError(w, http.StatusBadRequest, fmt.Sprintf("snapraid_conf: %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 = 20 } offset, _ := strconv.Atoi(r.URL.Query().Get("offset")) if offset < 0 { offset = 0 } jobs, err := s.JM.List(limit, offset) if err != nil { writeError(w, http.StatusInternalServerError, err.Error()) return } total, err := s.DB.CountStorageJobs() if err != nil { writeError(w, http.StatusInternalServerError, err.Error()) return } writeJSON(w, http.StatusOK, map[string]any{"jobs": jobs, "total": total, "limit": limit, "offset": offset}) } 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 (s *Server) handleStorageDiskUsage(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, map[string]any{"source": storage.StatPath(cfg.MoverSource)}) } 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 }