diff --git a/CHANGELOG.md b/CHANGELOG.md index 563e24c..f41d155 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,14 @@ The project publishes 0.x prerelease versions; a stable release line is not yet ## [Unreleased] +### Added + +- `mem put --watch` one-way directory watch daemon (`#110`). Polls every 5s + for new/changed files (size-based detection), ingests them into mem, and prints + per-cycle reports (scanned/ingested/unchanged/failed counts). Strictly local→mem + one-way; no write-back, no deletion propagation, no bidirectional sync. Graceful + shutdown on SIGINT/SIGTERM. Supports `--format json` for machine-readable reports. + ### Changed - Ingest cursor locks try non-blocking exclusive locks and give up after 5s so a wedged peer becomes a warning instead of a silent hang. Refs #139. diff --git a/server/cmd/mem/cmds_file.go b/server/cmd/mem/cmds_file.go index fc2a8a5..6ebc853 100644 --- a/server/cmd/mem/cmds_file.go +++ b/server/cmd/mem/cmds_file.go @@ -8,6 +8,7 @@ import ( "net/http" "net/url" "os" + "os/signal" "path/filepath" "strings" "time" @@ -19,6 +20,7 @@ import ( func newPutCmd() *cobra.Command { var ( recursive bool + watch bool tag []string name string mimeFlag string @@ -80,15 +82,22 @@ func newPutCmd() *cobra.Command { return err } if st.IsDir() { + if watch { + return watchDir(cmd, c, target, toFolder, tag, sourceMetadata, format) + } if !recursive { return errors.New("path is a directory; pass --recursive to upload its contents") } return uploadDir(c, target, toFolder, tag, sourceMetadata, format) } + if watch { + return errors.New("--watch requires a directory path") + } return uploadFile(c, target, name, mimeFlag, toFolder, tag, sourceMetadata, format) }, } cmd.Flags().BoolVarP(&recursive, "recursive", "r", false, "recurse into directories") + cmd.Flags().BoolVarP(&watch, "watch", "w", false, "watch directory for new/changed files (one-way daemon)") cmd.Flags().StringArrayVar(&tag, "tag", nil, "tag(s) to attach (repeatable)") cmd.Flags().StringVar(&name, "name", "", "override file name (required for stdin)") cmd.Flags().StringVar(&mimeFlag, "mime", "", "override MIME type") @@ -524,3 +533,146 @@ func isTextLike(ctype string) bool { } return false } + +// watchDir implements the --watch daemon: polls a directory for new/changed +// files and ingests them one-way into mem. Per-cycle reports are printed to +// stdout. The daemon runs until interrupted (SIGINT/SIGTERM) or the watched +// directory disappears. +func watchDir(cmd *cobra.Command, c *httpClient, root, targetFolder string, tags []string, sourceMetadata *apiclient.FileSourceMetadata, format string) error { + // Polling interval (could be made configurable via flag in future) + interval := 5 * time.Second + + // Track file states: path -> last known size + fileStates := make(map[string]int64) + + fmt.Fprintf(cmd.OutOrStdout(), "Watching %s (polling every %v, Ctrl+C to stop)\n", root, interval) + + // Set up signal handling for graceful shutdown + sigCh := make(chan os.Signal, 1) + signal.Notify(sigCh, os.Interrupt, os.Kill) + defer signal.Stop(sigCh) + + ticker := time.NewTicker(interval) + defer ticker.Stop() + + cycleNum := 0 + for { + select { + case <-sigCh: + fmt.Fprintf(cmd.OutOrStdout(), "\nStopping watch daemon\n") + return nil + case <-ticker.C: + cycleNum++ + report, err := runWatchCycle(c, root, targetFolder, tags, sourceMetadata, fileStates) + if err != nil { + // If root directory is gone, exit cleanly + if errors.Is(err, os.ErrNotExist) { + fmt.Fprintf(cmd.OutOrStdout(), "Watched directory no longer exists, stopping\n") + return nil + } + fmt.Fprintf(cmd.ErrOrStderr(), "Cycle %d error: %v\n", cycleNum, err) + continue + } + // Print per-cycle report + if format == "json" { + printWatchReportJSON(cmd.OutOrStdout(), cycleNum, report) + } else { + printWatchReportText(cmd.OutOrStdout(), cycleNum, report) + } + } + } +} + +// watchCycleReport holds the results of a single watch cycle. +type watchCycleReport struct { + Scanned int `json:"scanned"` + Ingested int `json:"ingested"` + Unchanged int `json:"unchanged"` + Failed int `json:"failed"` + Failures []string `json:"failures,omitempty"` +} + +// runWatchCycle walks the directory, detects new/changed files, and ingests them. +func runWatchCycle(c *httpClient, root, targetFolder string, tags []string, sourceMetadata *apiclient.FileSourceMetadata, fileStates map[string]int64) (watchCycleReport, error) { + report := watchCycleReport{} + + err := filepath.Walk(root, func(path string, info os.FileInfo, err error) error { + if err != nil { + return err + } + // Skip directories and hidden files + if info.IsDir() { + if strings.HasPrefix(info.Name(), ".") && path != root { + return filepath.SkipDir + } + return nil + } + if strings.HasPrefix(info.Name(), ".") { + return nil + } + + report.Scanned++ + + // Check if file has changed (size-based detection) + lastSize, known := fileStates[path] + if known && lastSize == info.Size() { + report.Unchanged++ + return nil + } + + // File is new or changed, ingest it + // Compute relative path for target folder + relPath, err := filepath.Rel(root, path) + if err != nil { + relPath = filepath.Base(path) + } + destPath := targetFolder + if relPath != "." && relPath != "" { + destPath = strings.TrimSuffix(targetFolder, "/") + "/" + filepath.ToSlash(relPath) + } + + err = uploadFile(c, path, "", "", destPath, tags, sourceMetadata, "text") + if err != nil { + report.Failed++ + report.Failures = append(report.Failures, fmt.Sprintf("%s: %v", path, err)) + return nil // Continue watching other files + } + + // Update state + fileStates[path] = info.Size() + report.Ingested++ + return nil + }) + + return report, err +} + +func printWatchReportText(w io.Writer, cycle int, report watchCycleReport) { + fmt.Fprintf(w, "Cycle %d: scanned=%d ingested=%d unchanged=%d failed=%d\n", + cycle, report.Scanned, report.Ingested, report.Unchanged, report.Failed) + if len(report.Failures) > 0 { + for _, f := range report.Failures { + fmt.Fprintf(w, " - %s\n", f) + } + } +} + +func printWatchReportJSON(w io.Writer, cycle int, report watchCycleReport) { + // Simple JSON output without importing encoding/json to keep dependencies minimal + fmt.Fprintf(w, `{"cycle":%d,"scanned":%d,"ingested":%d,"unchanged":%d,"failed":%d`, + cycle, report.Scanned, report.Ingested, report.Unchanged, report.Failed) + if len(report.Failures) > 0 { + fmt.Fprintf(w, `,"failures":[`) + for i, f := range report.Failures { + if i > 0 { + fmt.Fprintf(w, ",") + } + // Escape quotes in failure messages + escaped := strings.ReplaceAll(f, `"`, `\"`) + fmt.Fprintf(w, `"%s"`, escaped) + } + fmt.Fprintf(w, "]") + } + fmt.Fprintf(w, "}\n") +} +