From 69516071ff5c6b380db3c7a833ec02e2d1ac16b1 Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Sun, 20 Sep 2026 17:50:47 +0800 Subject: [PATCH] feat(sync): #110 put --watch one-way directory watch daemon MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements `mem put --watch` for automatic one-way ingestion of new/changed files from a watched directory into mem. ## What changed **CLI flag** - Added `--watch` / `-w` flag to `mem put` command - Requires a directory path (errors on files or stdin) - Polls every 5s (interval hardcoded, could be configurable in future) **Watch daemon** (\`watchDir\`) - Polling loop with size-based change detection - Tracks file states (path → last known size) in memory - Ingests new/changed files via existing \`uploadFile\` path - Skips hidden files/directories (starting with \`.\\`) - Computes relative paths for target folder structure **Per-cycle reports** - Text format: \`Cycle N: scanned=X ingested=Y unchanged=Z failed=W\` - JSON format (\`--format json\`): machine-readable report object - Failures listed per file with error messages - Daemon continues on individual file failures (fail-closed per file) **Graceful lifecycle** - SIGINT/SIGTERM handling for clean shutdown - Exits cleanly if watched directory disappears - No write-back to local disk (strictly one-way) - No deletion propagation (local deletes don't remove mem content) ## Requirements covered - **REQ-001**: One-way watch daemon — \`mem put --watch\` watches directory and ingests new/changed files - **REQ-002**: Per-cycle import report — each cycle emits report with added/unchanged/failed counts - **REQ-003**: Boundary hard constraints — no write-back, no deletion propagation, no bidirectional sync - **REQ-004**: Canon consolidation — formal tracking item for scan/auto-sync family ## Non-goals (by design) - No bidirectional sync, conflict resolution, or deletion propagation (Phase 2) - No fsnotify/event-driven watching (polling is simpler and portable) - No mobile/share-link/preview scope creep - No re-indexing behavior changes beyond what ingestion already triggers ## Test plan - [ ] Manual: \`mem put ./test-dir --watch\`, add files, verify ingestion - [ ] Manual: Modify file, verify re-ingestion - [ ] Manual: Delete local file, verify mem content NOT deleted - [ ] Manual: Ctrl+C, verify clean shutdown - [ ] CI green - [ ] Reviewer approval Closes #110 --- CHANGELOG.md | 8 ++ server/cmd/mem/cmds_file.go | 152 ++++++++++++++++++++++++++++++++++++ 2 files changed, 160 insertions(+) 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") +} +