improved kv metacore

2026-04-27 10:31:11 +02:00
parent 98da535335
commit bd833b7be4
33 changed files with 3091 additions and 318 deletions
@@ -29,6 +29,7 @@ type Server struct {
 	mountsManager     *mounts.Manager
 	mountsHandlers    *mounts.Handlers
 	mountPoller       *mounts.Poller
+	mountStatsPoller  *mounts.StatsPoller
 	watcherDispatcher *watcher.Dispatcher
 	fileWatcher       *watcher.Watcher
 	watcherHandlers   *watcher.Handlers
@@ -71,6 +72,11 @@ func NewServer(
 			log.Printf("[API] Warning: failed to initialize file watcher: %v", err)
 		} else {
 			s.fileWatcher = fileWatcher
+			// Wire storage so the watcher can persist (path,size,mtime,cid)
+			// tuples to Redis as it computes midhashes.
+			if stor != nil {
+				fileWatcher.SetStorageClient(stor)
+			}
 			s.watcherHandlers = watcher.NewHandlers(fileWatcher, s.watcherDispatcher)
 			// Create scan function from watcher for mounts
 			scanFunc = fileWatcher.ScanMountPath
@@ -88,6 +94,20 @@ func NewServer(
 				s.watchersPoller = watchers.NewPoller(watchersManager, s.fileWatcher, s.watcherDispatcher)
 				// Pass republish callback to handlers (handles nil metaPublisher case)
 				s.watchersHandlers = watchers.NewHandlers(watchersManager, s.watchersPoller, s.RepublishMetadata)
+
+				// Hydrate the in-memory state registry from Redis tuples so
+				// the first scan tick after boot can skip ComputeMidHash256
+				// for files whose (size, mtime) match.
+				configs := watchersManager.List()
+				roots := make([]watcher.WatcherRoot, 0, len(configs))
+				for _, c := range configs {
+					roots = append(roots, watcher.WatcherRoot{ID: c.ID, Root: c.Path})
+				}
+				if loaded, skipped, err := s.fileWatcher.HydrateStateFromStorage(roots); err != nil {
+					log.Printf("[API] Warning: failed to hydrate watcher state: %v", err)
+				} else {
+					log.Printf("[API] Hydrated watcher state from Redis: %d files loaded, %d skipped (no matching watcher root)", loaded, skipped)
+				}
 			}
 		}
 	}
@@ -103,6 +123,12 @@ func NewServer(
 		// Create and configure mount poller
 		s.mountPoller = mounts.NewPoller(mountsManager, scanFunc)
 		mountsManager.SetPoller(s.mountPoller)
+
+		// IO stats poller — queries the local rclone RC API every few seconds
+		// and exposes live throughput per mount via the same status JSON.
+		// Stats are daemon-global (rclone limitation); see stats_poller.go.
+		s.mountStatsPoller = mounts.NewStatsPoller(mountsManager)
+		mountsManager.SetStatsPoller(s.mountStatsPoller)
 	}

 	// Initialize WebDAV handler (caching is handled by nginx proxy_cache)
@@ -126,6 +152,11 @@ func (s *Server) setupRoutes() {
 	s.router.HandleFunc("/api/metadata/search", s.handleSearchMetadata).Methods("POST")
 	s.router.HandleFunc("/api/metadata/batch", s.handleBatchUpdate).Methods("POST")
 	s.router.HandleFunc("/api/metadata/clear", s.handleClearMetadata).Methods("POST")
+
+	// Snapshot (export / import / wipe)
+	s.router.HandleFunc("/api/snapshot/export", s.handleSnapshotExport).Methods("GET")
+	s.router.HandleFunc("/api/snapshot/import", s.handleSnapshotImport).Methods("POST")
+	s.router.HandleFunc("/api/snapshot/wipe", s.handleSnapshotWipe).Methods("POST")
 	s.router.HandleFunc("/api/metadata/{hashId}/property", s.handleMetadataGetProperty).Methods("GET")
 	s.router.HandleFunc("/api/metadata/{hashId}/property", s.handleMetadataUpdateProperty).Methods("PUT")
 	s.router.HandleFunc("/api/metadata/{hashId}", s.handleGetMetadataByHashId).Methods("GET")
@@ -240,6 +271,13 @@ func (s *Server) Start() error {
 		}
 	}

+	// Start IO stats poller alongside the mount poller. Cheap (one /proc read
+	// per CIFS/NFS mount per cycle, plus one RC call shared by all rclone
+	// mounts), so always-on rather than gated by config.
+	if s.mountStatsPoller != nil {
+		s.mountStatsPoller.Start()
+	}
+
 	// Start meta publisher (if storage is connected)
 	// Publishes keyspace notifications to meta:events stream
 	if s.storage != nil && s.storage.IsConnected() {
@@ -278,6 +316,11 @@ func (s *Server) Stop() error {
 		}
 	}

+	// Stop IO stats poller
+	if s.mountStatsPoller != nil {
+		s.mountStatsPoller.Stop()
+	}
+
 	// Stop meta publisher
 	if s.metaPublisher != nil {
 		if err := s.metaPublisher.Stop(); err != nil {
@@ -0,0 +1,167 @@
+package api
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+	"net/http"
+	"path/filepath"
+	"strings"
+	"time"
+
+	"github.com/metazla/meta-core/internal/snapshot"
+)
+
+// Bound the upload size so a malicious client can't OOM the server.
+// 2 GiB is well above any realistic export.
+const maxImportBytes = 2 << 30
+
+// pluginOutputDir is where plugin-produced artifacts live on disk
+// (subtitles, posters, etc). Mounted from the host into the meta-core
+// container at $FILES_PATH/plugin.
+func (s *Server) pluginOutputDir() string {
+	return filepath.Join(s.config.FilesPath, "plugin")
+}
+
+// handleSnapshotExport streams a ZIP snapshot of all metadata.
+// GET /api/snapshot/export?include=hash-index,plugin-output
+func (s *Server) handleSnapshotExport(w http.ResponseWriter, r *http.Request) {
+	if !s.storage.IsConnected() {
+		writeError(w, http.StatusServiceUnavailable, "storage not connected")
+		return
+	}
+
+	// The server has a 30s WriteTimeout (server.go:222). Long exports
+	// blow past that, leaving a truncated ZIP without a central directory.
+	// Clear the deadline for this handler so the export can run as long as
+	// needed. ResponseController is the standard way to do this.
+	rc := http.NewResponseController(w)
+	_ = rc.SetWriteDeadline(time.Time{})
+
+	include := splitCSVParam(r.URL.Query().Get("include"))
+	opts := snapshot.ExportOptions{
+		IncludePluginOutput: include["plugin-output"],
+		PluginOutputDir:     s.pluginOutputDir(),
+	}
+
+	filename := fmt.Sprintf("metamesh-snapshot-%s.zip", time.Now().UTC().Format("20060102-150405"))
+	w.Header().Set("Content-Type", "application/zip")
+	w.Header().Set("Content-Disposition", fmt.Sprintf(`attachment; filename="%s"`, filename))
+	// We stream — no Content-Length.
+
+	exporter := snapshot.NewExporter(s.storage)
+	if _, err := exporter.Export(w, opts); err != nil {
+		// Headers may already be sent; log and bail.
+		// Client will see a truncated zip, which is the best signal we can give.
+		fmt.Printf("[Snapshot] export failed: %v\n", err)
+		return
+	}
+}
+
+// handleSnapshotImport applies an uploaded ZIP snapshot.
+// POST /api/snapshot/import?mode=replace|merge&conflict=mine|source&dry_run=true
+// Body: application/zip (raw) OR multipart/form-data with field "snapshot"
+func (s *Server) handleSnapshotImport(w http.ResponseWriter, r *http.Request) {
+	if !s.storage.IsConnected() {
+		writeError(w, http.StatusServiceUnavailable, "storage not connected")
+		return
+	}
+
+	// Long imports can blow past the server's 30s ReadTimeout/WriteTimeout
+	// when the snapshot is large. Clear both for this handler.
+	rc := http.NewResponseController(w)
+	_ = rc.SetReadDeadline(time.Time{})
+	_ = rc.SetWriteDeadline(time.Time{})
+
+	q := r.URL.Query()
+	opts := snapshot.ImportOptions{
+		Mode:            snapshot.Mode(q.Get("mode")),
+		Conflict:        snapshot.Conflict(q.Get("conflict")),
+		DryRun:          strings.EqualFold(q.Get("dry_run"), "true"),
+		PluginOutputDir: s.pluginOutputDir(),
+	}
+	if opts.Mode == "" {
+		opts.Mode = snapshot.ModeMerge
+	}
+	if opts.Mode == snapshot.ModeMerge && opts.Conflict == "" {
+		opts.Conflict = snapshot.ConflictSource
+	}
+
+	body, err := readImportBody(r)
+	if err != nil {
+		writeError(w, http.StatusBadRequest, err.Error())
+		return
+	}
+
+	importer := snapshot.NewImporter(s.storage)
+	result, err := importer.Import(bytes.NewReader(body), int64(len(body)), opts)
+	if err != nil {
+		writeError(w, http.StatusBadRequest, err.Error())
+		return
+	}
+
+	writeJSON(w, http.StatusOK, result)
+}
+
+// handleSnapshotWipe clears Redis metadata.
+// POST /api/snapshot/wipe?scope=metadata
+func (s *Server) handleSnapshotWipe(w http.ResponseWriter, r *http.Request) {
+	if !s.storage.IsConnected() {
+		writeError(w, http.StatusServiceUnavailable, "storage not connected")
+		return
+	}
+
+	scopes := splitCSVParam(r.URL.Query().Get("scope"))
+	if len(scopes) == 0 {
+		writeError(w, http.StatusBadRequest, "scope query param is required (e.g. scope=metadata)")
+		return
+	}
+	scope := snapshot.WipeScope{
+		Metadata: scopes["metadata"],
+	}
+	if !scope.Metadata {
+		writeError(w, http.StatusBadRequest, "no supported scope requested; scope=metadata is the only accepted value")
+		return
+	}
+
+	wiper := snapshot.NewWiper(s.storage)
+	res, err := wiper.Wipe(scope)
+	if err != nil {
+		writeError(w, http.StatusInternalServerError, err.Error())
+		return
+	}
+	writeJSON(w, http.StatusOK, res)
+}
+
+// readImportBody accepts either a raw application/zip body or a multipart
+// upload with a field named "snapshot". Buffers up to maxImportBytes.
+func readImportBody(r *http.Request) ([]byte, error) {
+	r.Body = http.MaxBytesReader(nil, r.Body, maxImportBytes)
+
+	ct := r.Header.Get("Content-Type")
+	if strings.HasPrefix(ct, "multipart/form-data") {
+		if err := r.ParseMultipartForm(32 << 20); err != nil {
+			return nil, fmt.Errorf("parse multipart: %w", err)
+		}
+		f, _, err := r.FormFile("snapshot")
+		if err != nil {
+			return nil, fmt.Errorf("missing 'snapshot' file field: %w", err)
+		}
+		defer f.Close()
+		return io.ReadAll(f)
+	}
+
+	return io.ReadAll(r.Body)
+}
+
+// splitCSVParam parses "a,b,c" into a set. Empty values yield an empty set.
+func splitCSVParam(v string) map[string]bool {
+	out := map[string]bool{}
+	for _, part := range strings.Split(v, ",") {
+		p := strings.TrimSpace(part)
+		if p != "" {
+			out[p] = true
+		}
+	}
+	return out
+}