baby-nas/internal/web/handlers_storage.go

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 = 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 (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
}