From a7aa85b5dd98e6ffa75da8da93d4ee38a650636b Mon Sep 17 00:00:00 2001 From: liyuanyang Date: Tue, 1 Sep 2026 14:45:29 +0800 Subject: [PATCH] feat(cli): add `mem put --watch` one-way directory watch daemon (#110) Implement a foreground poll-based directory watcher that automatically ingests new files into mem with per-cycle import reports. Key behaviors: - Poll loop with configurable --interval (default 30s) - Stability gate: new files must be unchanged across two consecutive scans before upload (prevents half-written file ingestion) - Change detection: (size, mtime) cost gate, sha256 authority - Changed files are reported but NOT re-ingested (file plane has no version model; re-ingest is Phase 2 sync-drive scope) - Local deletions produce local_gone reports with zero delete calls (no write-back, no deletion propagation) - Per-cycle report with closed vocabulary: scanned/ingested/deduped/ unchanged/local_gone/failed, persisted as capped JSONL - Single-instance advisory lock per watched root (flock LOCK_NB) - Graceful SIGINT/SIGTERM handling (exit 0 between cycles) - Give-up after 10 consecutive all-fail cycles with mapped exit codes - Deduped items carry server-returned id/path (not assumed --to folder) Closes #110 --- server/cmd/mem/cmds_file.go | 17 + server/cmd/mem/watch.go | 366 +++++++++++++++++ server/cmd/mem/watch_lock.go | 42 ++ server/cmd/mem/watch_lock_aix.go | 23 ++ server/cmd/mem/watch_lock_other.go | 16 + server/cmd/mem/watch_lock_unix.go | 17 + server/cmd/mem/watch_lock_windows.go | 24 ++ server/cmd/mem/watch_report.go | 231 +++++++++++ server/cmd/mem/watch_test.go | 587 +++++++++++++++++++++++++++ 9 files changed, 1323 insertions(+) create mode 100644 server/cmd/mem/watch.go create mode 100644 server/cmd/mem/watch_lock.go create mode 100644 server/cmd/mem/watch_lock_aix.go create mode 100644 server/cmd/mem/watch_lock_other.go create mode 100644 server/cmd/mem/watch_lock_unix.go create mode 100644 server/cmd/mem/watch_lock_windows.go create mode 100644 server/cmd/mem/watch_report.go create mode 100644 server/cmd/mem/watch_test.go diff --git a/server/cmd/mem/cmds_file.go b/server/cmd/mem/cmds_file.go index 37f30a1..0ecb19e 100644 --- a/server/cmd/mem/cmds_file.go +++ b/server/cmd/mem/cmds_file.go @@ -31,6 +31,8 @@ func newPutCmd() *cobra.Command { place string sourceKind string sourceName string + watch bool + interval time.Duration ) cmd := &cobra.Command{ Use: "put ", @@ -77,9 +79,22 @@ func newPutCmd() *cobra.Command { st, err := os.Stat(target) if err != nil { + if watch { + return newCliError(2, fmt.Sprintf("watch root: %v", err), "") + } return err } if st.IsDir() { + if watch { + opts := watchOptions{ + root: target, + interval: interval, + toFolder: toFolder, + tags: tag, + format: format, + } + return runWatchDaemon(cmd, c, opts, sourceMetadata) + } if !recursive { return errors.New("path is a directory; pass --recursive to upload its contents") } @@ -102,6 +117,8 @@ func newPutCmd() *cobra.Command { cmd.Flags().StringVar(&place, "place", "", "human-readable capture location (requires --lat/--lon)") cmd.Flags().StringVar(&sourceKind, "source-kind", "cli", "api|web|cli|mcp|mobile|ai_device|import|other") cmd.Flags().StringVar(&sourceName, "source-name", "", "non-sensitive source/device description") + cmd.Flags().BoolVar(&watch, "watch", false, "watch directory for new files and upload continuously") + cmd.Flags().DurationVar(&interval, "interval", 30*time.Second, "poll interval for --watch (e.g. 10s, 1m)") return cmd } diff --git a/server/cmd/mem/watch.go b/server/cmd/mem/watch.go new file mode 100644 index 0000000..ae2a78a --- /dev/null +++ b/server/cmd/mem/watch.go @@ -0,0 +1,366 @@ +package main + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "mime" + "os" + "os/signal" + "path/filepath" + "syscall" + "time" + + "github.com/PeterGuy326/mem/server/internal/apiclient" + "github.com/spf13/cobra" +) + +type watchCursor struct { + Abs string `json:"abs"` + Size int64 `json:"size"` + ModTime string `json:"mtime"` + SHA256 string `json:"sha256"` + FileID string `json:"file_id"` + IngestedAt string `json:"ingested_at"` +} + +type pendingFile struct { + Abs string + Size int64 + ModTime string +} + +type watchOptions struct { + root string + interval time.Duration + toFolder string + tags []string + format string +} + +func (o watchOptions) stateDir() string { + return filepath.Join(cliStateRoot(), "watch") +} + +func (o watchOptions) cursorDir() string { + return filepath.Join(o.stateDir(), "cursors") +} + +func (o watchOptions) lockPath() string { + return filepath.Join(o.stateDir(), sha1Hex(o.root)+".lock") +} + +func runWatchDaemon(cmd *cobra.Command, c *httpClient, opts watchOptions, sourceMetadata *apiclient.FileSourceMetadata) error { + absRoot, err := filepath.Abs(opts.root) + if err != nil { + return err + } + fi, err := os.Stat(absRoot) + if err != nil { + return newCliError(2, fmt.Sprintf("watch root: %v", err), "") + } + if !fi.IsDir() { + return newCliError(2, "watch root is not a directory", "pass a directory path to --watch") + } + + lock, err := acquireWatchLock(opts.lockPath()) + if err != nil { + return newCliError(1, "another watcher is already running for this root", "") + } + defer lock.release() + + cursors := loadAllWatchCursors(opts.cursorDir()) + pending := make(map[string]*pendingFile) + consecutiveFails := 0 + + ctx, stop := signal.NotifyContext(cmd.Context(), os.Interrupt, syscall.SIGTERM) + defer stop() + + runCycle := func() (bool, error) { + report := scanAndUpload(ctx, cmd, c, opts, absRoot, cursors, pending, sourceMetadata) + printReport(cmd, report, opts.format) + if perr := persistReport(opts.stateDir(), absRoot, report); perr != nil { + fmt.Fprintf(cmd.ErrOrStderr(), "warn: persist report: %v\n", perr) + } + consecutiveFails = updateFailCounter(consecutiveFails, report) + if consecutiveFails >= 10 { + code := giveUpExitCode(report) + return false, newCliError(code, "watch giving up after 10 consecutive all-fail cycles", "") + } + return true, nil + } + + ok, err := runCycle() + if !ok { + return err + } + + ticker := time.NewTicker(opts.interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return nil + case <-ticker.C: + ok, err := runCycle() + if !ok { + return err + } + } + } +} + +func scanAndUpload(ctx context.Context, cmd *cobra.Command, c *httpClient, opts watchOptions, absRoot string, cursors map[string]watchCursor, pending map[string]*pendingFile, sourceMetadata *apiclient.FileSourceMetadata) cycleReport { + report := cycleReport{Timestamp: time.Now().UTC().Format(time.RFC3339)} + seen := make(map[string]bool) + + _ = filepath.WalkDir(absRoot, func(p string, d os.DirEntry, err error) error { + if ctx.Err() != nil { + return filepath.SkipAll + } + if err != nil { + return nil + } + if d.IsDir() { + return nil + } + abs, absErr := filepath.Abs(p) + if absErr != nil { + return nil + } + seen[abs] = true + report.Counts.Scanned++ + + fi, infoErr := d.Info() + if infoErr != nil { + report.Counts.Failed++ + report.Items = append(report.Items, cycleItem{Status: "failed", LocalPath: abs, FailureCode: failureReadDenied}) + return nil + } + + cur, hasCursor := cursors[abs] + sizeStr := fmt.Sprintf("%d", fi.Size()) + mtimeStr := fi.ModTime().UTC().Format("2006-01-02T15:04:05Z") + + if !hasCursor { + prev, hasPending := pending[abs] + if !hasPending || prev.Size != fi.Size() || prev.ModTime != mtimeStr { + pending[abs] = &pendingFile{Abs: abs, Size: fi.Size(), ModTime: mtimeStr} + return nil + } + delete(pending, abs) + uploadOne(cmd, c, opts, abs, fi, &report, cursors, sourceMetadata) + return nil + } + + if sizeStr == cur.ModTime || (fi.Size() == cur.Size && mtimeStr == cur.ModTime) { + report.Counts.Unchanged++ + report.Items = append(report.Items, cycleItem{Status: "unchanged", LocalPath: abs, FileID: cur.FileID}) + return nil + } + + hash, hashErr := computeSHA256(abs) + if hashErr != nil { + report.Counts.Failed++ + report.Items = append(report.Items, cycleItem{Status: "failed", LocalPath: abs, FailureCode: failureReadDenied}) + return nil + } + if hash == cur.SHA256 { + cur.Size = fi.Size() + cur.ModTime = mtimeStr + cursors[abs] = cur + _ = saveWatchCursor(opts.cursorDir(), cur) + report.Counts.Unchanged++ + report.Items = append(report.Items, cycleItem{Status: "unchanged", LocalPath: abs, FileID: cur.FileID}) + return nil + } + + prefix := hash + if len(prefix) > 16 { + prefix = prefix[:16] + } + report.Items = append(report.Items, cycleItem{Status: "changed", LocalPath: abs, FileID: cur.FileID, SHA256Prefix: prefix}) + cur.SHA256 = hash + cur.Size = fi.Size() + cur.ModTime = mtimeStr + cursors[abs] = cur + _ = saveWatchCursor(opts.cursorDir(), cur) + return nil + }) + + for abs, cur := range cursors { + if seen[abs] { + continue + } + if isPending(pending, abs) { + continue + } + report.Counts.LocalGone++ + report.Items = append(report.Items, cycleItem{Status: "local_gone", LocalPath: abs, FileID: cur.FileID}) + } + + return report +} + +func isPending(pending map[string]*pendingFile, abs string) bool { + _, ok := pending[abs] + return ok +} + +func uploadOne(cmd *cobra.Command, c *httpClient, opts watchOptions, abs string, fi os.FileInfo, report *cycleReport, cursors map[string]watchCursor, sourceMetadata *apiclient.FileSourceMetadata) { + rel, _ := filepath.Abs(abs) + rootAbs, _ := filepath.Abs(opts.root) + relPath, _ := filepath.Rel(rootAbs, rel) + subFolder := opts.toFolder + if d := filepath.Dir(relPath); d != "" && d != "." { + subFolder = joinFolder(opts.toFolder, filepath.ToSlash(d)) + } + + name := filepath.Base(abs) + mimeType := mime.TypeByExtension(filepath.Ext(name)) + + f, err := os.Open(abs) + if err != nil { + report.Counts.Failed++ + report.Items = append(report.Items, cycleItem{Status: "failed", LocalPath: abs, FailureCode: failureReadDenied}) + return + } + defer f.Close() + + var resp map[string]any + if err := c.api.UploadMultipartWithSourceMetadata(cmd.Context(), name, mimeType, subFolder, f, opts.tags, sourceMetadata, &resp); err != nil { + report.Counts.Failed++ + code := classifyUploadError(err) + report.Items = append(report.Items, cycleItem{Status: "failed", LocalPath: abs, FailureCode: code}) + fmt.Fprintf(cmd.ErrOrStderr(), "warn: upload %s: %v\n", abs, err) + return + } + + fileObj, _ := resp["file"].(map[string]any) + fileID, _ := fileObj["id"].(string) + virtualPath, _ := fileObj["path"].(string) + if virtualPath == "" { + if p, ok := fileObj["virtual_path"].(string); ok { + virtualPath = p + } + } + + hash, _ := computeSHA256(abs) + deduped, _ := resp["deduped"].(bool) + + status := "ingested" + if deduped { + status = "deduped" + report.Counts.Deduped++ + } else { + report.Counts.Ingested++ + } + + report.Items = append(report.Items, cycleItem{ + Status: status, + LocalPath: abs, + FileID: fileID, + VirtualPath: virtualPath, + }) + + cursors[abs] = watchCursor{ + Abs: abs, + Size: fi.Size(), + ModTime: fi.ModTime().UTC().Format("2006-01-02T15:04:05Z"), + SHA256: hash, + FileID: fileID, + IngestedAt: time.Now().UTC().Format(time.RFC3339), + } + _ = saveWatchCursor(filepath.Join(cliStateRoot(), "watch", "cursors"), cursors[abs]) +} + +func computeSHA256(abs string) (string, error) { + f, err := os.Open(abs) + if err != nil { + return "", err + } + defer f.Close() + h := sha256.New() + if _, err := io.Copy(h, f); err != nil { + return "", err + } + return hex.EncodeToString(h.Sum(nil)), nil +} + +func watchCursorPath(cursorDir, abs string) string { + return filepath.Join(cursorDir, sha1Hex(abs)+".json") +} + +func loadWatchCursor(cursorDir, abs string) (watchCursor, bool) { + p := watchCursorPath(cursorDir, abs) + b, err := os.ReadFile(p) + if err != nil { + return watchCursor{}, false + } + var cur watchCursor + if err := json.Unmarshal(b, &cur); err != nil { + return watchCursor{}, false + } + return cur, true +} + +func saveWatchCursor(cursorDir string, cur watchCursor) error { + p := watchCursorPath(cursorDir, cur.Abs) + if err := os.MkdirAll(filepath.Dir(p), 0o700); err != nil { + return err + } + b, err := json.Marshal(cur) + if err != nil { + return err + } + tmp, err := os.CreateTemp(filepath.Dir(p), ".cursor-tmp-*") + if err != nil { + return err + } + tmpName := tmp.Name() + defer func() { + _ = tmp.Close() + _ = os.Remove(tmpName) + }() + if err := tmp.Chmod(0o600); err != nil { + return err + } + if _, err := tmp.Write(b); err != nil { + return err + } + if err := tmp.Sync(); err != nil { + return err + } + if err := tmp.Close(); err != nil { + return err + } + return os.Rename(tmpName, p) +} + +func loadAllWatchCursors(cursorDir string) map[string]watchCursor { + cursors := make(map[string]watchCursor) + entries, err := os.ReadDir(cursorDir) + if err != nil { + return cursors + } + for _, e := range entries { + if e.IsDir() || filepath.Ext(e.Name()) != ".json" { + continue + } + b, err := os.ReadFile(filepath.Join(cursorDir, e.Name())) + if err != nil { + continue + } + var cur watchCursor + if err := json.Unmarshal(b, &cur); err != nil { + continue + } + if cur.Abs != "" { + cursors[cur.Abs] = cur + } + } + return cursors +} diff --git a/server/cmd/mem/watch_lock.go b/server/cmd/mem/watch_lock.go new file mode 100644 index 0000000..648325a --- /dev/null +++ b/server/cmd/mem/watch_lock.go @@ -0,0 +1,42 @@ +package main + +import ( + "fmt" + "os" + "path/filepath" +) + +type watchLock struct { + file *os.File +} + +func acquireWatchLock(lockPath string) (*watchLock, error) { + if err := os.MkdirAll(filepath.Dir(lockPath), 0o700); err != nil { + return nil, fmt.Errorf("create lock dir: %w", err) + } + file, err := os.OpenFile(lockPath, os.O_CREATE|os.O_RDWR, 0o600) + if err != nil { + return nil, fmt.Errorf("open lock file: %w", err) + } + if err := lockWatchFile(file); err != nil { + _ = file.Close() + return nil, fmt.Errorf("acquire watch lock: %w", err) + } + return &watchLock{file: file}, nil +} + +func (l *watchLock) release() error { + if l == nil || l.file == nil { + return nil + } + unlockErr := unlockWatchFile(l.file) + closeErr := l.file.Close() + l.file = nil + if unlockErr != nil { + return fmt.Errorf("unlock watch lock: %w", unlockErr) + } + if closeErr != nil { + return fmt.Errorf("close watch lock file: %w", closeErr) + } + return nil +} diff --git a/server/cmd/mem/watch_lock_aix.go b/server/cmd/mem/watch_lock_aix.go new file mode 100644 index 0000000..169e7ca --- /dev/null +++ b/server/cmd/mem/watch_lock_aix.go @@ -0,0 +1,23 @@ +//go:build aix + +package main + +import ( + "os" + + "golang.org/x/sys/unix" +) + +func lockWatchFile(file *os.File) error { + return unix.FcntlFlock(file.Fd(), unix.F_SETLK, &unix.Flock_t{ + Type: unix.F_WRLCK, + Len: 1, + }) +} + +func unlockWatchFile(file *os.File) error { + return unix.FcntlFlock(file.Fd(), unix.F_SETLK, &unix.Flock_t{ + Type: unix.F_UNLCK, + Len: 1, + }) +} diff --git a/server/cmd/mem/watch_lock_other.go b/server/cmd/mem/watch_lock_other.go new file mode 100644 index 0000000..50a2acc --- /dev/null +++ b/server/cmd/mem/watch_lock_other.go @@ -0,0 +1,16 @@ +//go:build !(aix || darwin || dragonfly || freebsd || linux || netbsd || openbsd || solaris || windows) + +package main + +import ( + "fmt" + "os" +) + +func lockWatchFile(_ *os.File) error { + return fmt.Errorf("watch locks are not supported on this operating system") +} + +func unlockWatchFile(_ *os.File) error { + return nil +} diff --git a/server/cmd/mem/watch_lock_unix.go b/server/cmd/mem/watch_lock_unix.go new file mode 100644 index 0000000..668cb5f --- /dev/null +++ b/server/cmd/mem/watch_lock_unix.go @@ -0,0 +1,17 @@ +//go:build darwin || dragonfly || freebsd || linux || netbsd || openbsd || solaris + +package main + +import ( + "os" + + "golang.org/x/sys/unix" +) + +func lockWatchFile(file *os.File) error { + return unix.Flock(int(file.Fd()), unix.LOCK_EX|unix.LOCK_NB) +} + +func unlockWatchFile(file *os.File) error { + return unix.Flock(int(file.Fd()), unix.LOCK_UN) +} diff --git a/server/cmd/mem/watch_lock_windows.go b/server/cmd/mem/watch_lock_windows.go new file mode 100644 index 0000000..c1124c0 --- /dev/null +++ b/server/cmd/mem/watch_lock_windows.go @@ -0,0 +1,24 @@ +//go:build windows + +package main + +import ( + "os" + + "golang.org/x/sys/windows" +) + +func lockWatchFile(file *os.File) error { + return windows.LockFileEx( + windows.Handle(file.Fd()), + windows.LOCKFILE_EXCLUSIVE_LOCK|windows.LOCKFILE_FAIL_IMMEDIATELY, + 0, + 1, + 0, + &windows.Overlapped{}, + ) +} + +func unlockWatchFile(file *os.File) error { + return windows.UnlockFileEx(windows.Handle(file.Fd()), 0, 1, 0, &windows.Overlapped{}) +} diff --git a/server/cmd/mem/watch_report.go b/server/cmd/mem/watch_report.go new file mode 100644 index 0000000..e6227a1 --- /dev/null +++ b/server/cmd/mem/watch_report.go @@ -0,0 +1,231 @@ +package main + +import ( + "bufio" + "crypto/sha1" + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "strings" + + "github.com/PeterGuy326/mem/server/internal/apiclient" + "github.com/spf13/cobra" +) + +const ( + failureAuth = "auth" + failurePlanQuota = "plan_quota" + failureProviderTimeout = "provider_timeout" + failureNetwork = "network" + failureReadDenied = "read_denied" + failureUploadRejected = "upload_rejected" + failureRootMissing = "root_missing" + failureStateCorrupt = "state_corrupt" +) + +const maxReportLines = 200 + +type cycleReport struct { + Timestamp string `json:"timestamp"` + Counts cycleCounts `json:"counts"` + Items []cycleItem `json:"items"` +} + +type cycleCounts struct { + Scanned int `json:"scanned"` + Ingested int `json:"ingested"` + Deduped int `json:"deduped"` + Unchanged int `json:"unchanged"` + LocalGone int `json:"local_gone"` + Failed int `json:"failed"` +} + +type cycleItem struct { + Status string `json:"status"` + LocalPath string `json:"local_path,omitempty"` + FileID string `json:"file_id,omitempty"` + VirtualPath string `json:"virtual_path,omitempty"` + SHA256Prefix string `json:"sha256_prefix,omitempty"` + FailureCode string `json:"failure_code,omitempty"` +} + +func sha1Hex(s string) string { + sum := sha1.Sum([]byte(s)) + return fmt.Sprintf("%x", sum[:]) +} + +func reportPath(stateDir, absRoot string) string { + return filepath.Join(stateDir, "reports", sha1Hex(absRoot)+".jsonl") +} + +func persistReport(stateDir, absRoot string, report cycleReport) error { + p := reportPath(stateDir, absRoot) + if err := os.MkdirAll(filepath.Dir(p), 0o700); err != nil { + return fmt.Errorf("create report dir: %w", err) + } + + b, err := json.Marshal(report) + if err != nil { + return fmt.Errorf("encode report: %w", err) + } + + f, err := os.OpenFile(p, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o600) + if err != nil { + return fmt.Errorf("open report: %w", err) + } + if _, err := f.Write(append(b, '\n')); err != nil { + _ = f.Close() + return fmt.Errorf("append report: %w", err) + } + if err := f.Close(); err != nil { + return fmt.Errorf("close report: %w", err) + } + + return capReportLines(p, maxReportLines) +} + +func capReportLines(path string, max int) error { + f, err := os.Open(path) + if err != nil { + return err + } + var lines []string + scanner := bufio.NewScanner(f) + scanner.Buffer(make([]byte, 0, 1024*1024), 1024*1024) + for scanner.Scan() { + lines = append(lines, scanner.Text()) + } + _ = f.Close() + if err := scanner.Err(); err != nil { + return err + } + if len(lines) <= max { + return nil + } + lines = lines[len(lines)-max:] + tmp, err := os.CreateTemp(filepath.Dir(path), ".report-tmp-*") + if err != nil { + return err + } + tmpName := tmp.Name() + defer func() { + _ = tmp.Close() + _ = os.Remove(tmpName) + }() + for _, line := range lines { + if _, err := tmp.WriteString(line + "\n"); err != nil { + return err + } + } + if err := tmp.Sync(); err != nil { + return err + } + if err := tmp.Close(); err != nil { + return err + } + return os.Rename(tmpName, path) +} + +func printReport(cmd *cobra.Command, report cycleReport, format string) { + out := cmd.OutOrStdout() + if format == "json" { + enc := json.NewEncoder(out) + enc.SetIndent("", " ") + _ = enc.Encode(report) + return + } + c := report.Counts + fmt.Fprintf(out, "watch cycle %s: scanned=%d ingested=%d deduped=%d unchanged=%d local_gone=%d failed=%d\n", + report.Timestamp, c.Scanned, c.Ingested, c.Deduped, c.Unchanged, c.LocalGone, c.Failed) + for _, item := range report.Items { + switch item.Status { + case "ingested", "deduped": + fmt.Fprintf(out, " %s %s -> %s (id=%s)\n", item.Status, item.LocalPath, item.VirtualPath, item.FileID) + case "changed": + fmt.Fprintf(out, " %s %s (sha256=%s...)\n", item.Status, item.LocalPath, item.SHA256Prefix) + case "local_gone": + fmt.Fprintf(out, " %s %s (was id=%s)\n", item.Status, item.LocalPath, item.FileID) + case "failed": + fmt.Fprintf(out, " %s %s [%s]\n", item.Status, item.LocalPath, item.FailureCode) + } + } +} + +func classifyUploadError(err error) string { + var ae *apiclient.APIError + if !errors.As(err, &ae) { + if isNetworkError(err) { + return failureNetwork + } + return failureUploadRejected + } + switch ae.Kind() { + case apiclient.KindAuth: + return failureAuth + case apiclient.KindPlan, apiclient.KindQuota: + return failurePlanQuota + case apiclient.KindProvider, apiclient.KindTimeout: + return failureProviderTimeout + default: + return failureUploadRejected + } +} + +func isNetworkError(err error) bool { + if err == nil { + return false + } + msg := err.Error() + return strings.Contains(msg, "connection refused") || + strings.Contains(msg, "no such host") || + strings.Contains(msg, "dial tcp") || + strings.Contains(msg, "network is unreachable") +} + +func isNonRetryableFailure(code string) bool { + switch code { + case failureAuth, failurePlanQuota, failureUploadRejected: + return true + } + return false +} + +func updateFailCounter(prev int, report cycleReport) int { + hasSuccess := report.Counts.Ingested > 0 || report.Counts.Deduped > 0 || report.Counts.Unchanged > 0 + if hasSuccess { + return 0 + } + if report.Counts.Failed == 0 { + return prev + } + allNonRetryable := true + for _, item := range report.Items { + if item.Status == "failed" && !isNonRetryableFailure(item.FailureCode) { + allNonRetryable = false + break + } + } + if allNonRetryable { + return prev + 1 + } + return 0 +} + +func giveUpExitCode(report cycleReport) int { + for _, item := range report.Items { + if item.Status != "failed" { + continue + } + switch item.FailureCode { + case failureAuth: + return 3 + case failurePlanQuota: + return 4 + case failureProviderTimeout: + return 5 + } + } + return 1 +} diff --git a/server/cmd/mem/watch_test.go b/server/cmd/mem/watch_test.go new file mode 100644 index 0000000..e56905a --- /dev/null +++ b/server/cmd/mem/watch_test.go @@ -0,0 +1,587 @@ +package main + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "sync/atomic" + "testing" + "time" +) + +func fileSHA256(t *testing.T, path string) string { + t.Helper() + b, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + h := sha256.Sum256(b) + return hex.EncodeToString(h[:]) +} + +func uploadMockServer(t *testing.T, requests *atomic.Int32, dedup bool) *httptest.Server { + t.Helper() + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests.Add(1) + if r.URL.Path != "/v1/files" { + t.Errorf("unexpected path: %s", r.URL.Path) + w.WriteHeader(http.StatusNotFound) + return + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusCreated) + status := "new" + if dedup { + status = "秒传" + } + resp := map[string]any{ + "file": map[string]any{ + "id": fmt.Sprintf("file-%d", requests.Load()), + "name": "test", + "size": 100, + "sha256": "abc123", + "mime": "text/plain", + "path": "/Uploaded/test", + "index_status": status, + }, + "deduped": dedup, + } + _ = json.NewEncoder(w).Encode(resp) + })) +} + +func setupWatchEnv(t *testing.T, srv *httptest.Server) string { + t.Helper() + dir := t.TempDir() + t.Setenv("MEM_CONFIG", filepath.Join(dir, "missing.yaml")) + t.Setenv("MEM_SERVER", srv.URL) + t.Setenv("MEM_TOKEN", "tok") + t.Setenv("MEM_WORKSPACE", "ws-1") + t.Setenv("MEM_STATE_DIR", filepath.Join(dir, "state")) + return dir +} + +func TestWatchNonExistentRoot(t *testing.T) { + srv := uploadMockServer(t, &atomic.Int32{}, false) + defer srv.Close() + setupWatchEnv(t, srv) + + root := newRootCmd() + root.SetArgs([]string{"put", "/nonexistent/path/xyz", "--watch"}) + err := root.Execute() + if err == nil { + t.Fatal("expected error for non-existent root") + } + var ce *cliError + if !isCliError(err, &ce) || ce.code != 2 { + t.Fatalf("err = %v, want exit code 2", err) + } +} + +func TestWatchUploadsNewFiles(t *testing.T) { + var requests atomic.Int32 + srv := uploadMockServer(t, &requests, false) + defer srv.Close() + dir := setupWatchEnv(t, srv) + + watchDir := filepath.Join(dir, "watch") + os.MkdirAll(watchDir, 0o755) + os.WriteFile(filepath.Join(watchDir, "a.txt"), []byte("hello"), 0o644) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + root := newRootCmd() + var stdout bytes.Buffer + root.SetOut(&stdout) + root.SetErr(&stdout) + root.SetContext(ctx) + root.SetArgs([]string{"put", watchDir, "--watch", "--interval", "50ms", "--to", "/Dest"}) + + errCh := make(chan error, 1) + go func() { errCh <- root.Execute() }() + + time.Sleep(300 * time.Millisecond) + cancel() + + if err := <-errCh; err != nil { + t.Fatal("watch returned error:", err, "\noutput:", stdout.String()) + } + if requests.Load() < 1 { + t.Fatalf("expected at least 1 upload, got %d\noutput: %s", requests.Load(), stdout.String()) + } + if !strings.Contains(stdout.String(), "ingested") { + t.Errorf("expected 'ingested' in output, got: %s", stdout.String()) + } +} + +func TestWatchStabilityRequiresTwoScans(t *testing.T) { + var requests atomic.Int32 + srv := uploadMockServer(t, &requests, false) + defer srv.Close() + dir := setupWatchEnv(t, srv) + + watchDir := filepath.Join(dir, "watch") + os.MkdirAll(watchDir, 0o755) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + root := newRootCmd() + var stdout bytes.Buffer + root.SetOut(&stdout) + root.SetErr(&stdout) + root.SetContext(ctx) + root.SetArgs([]string{"put", watchDir, "--watch", "--interval", "50ms"}) + + errCh := make(chan error, 1) + go func() { errCh <- root.Execute() }() + + time.Sleep(80 * time.Millisecond) + os.WriteFile(filepath.Join(watchDir, "new.txt"), []byte("content"), 0o644) + + time.Sleep(250 * time.Millisecond) + cancel() + + if err := <-errCh; err != nil { + t.Fatal(err) + } + if requests.Load() < 1 { + t.Fatalf("expected file to be uploaded after stability, got %d requests", requests.Load()) + } +} + +func TestWatchChangedHashReportsOnly(t *testing.T) { + var requests atomic.Int32 + srv := uploadMockServer(t, &requests, false) + defer srv.Close() + dir := setupWatchEnv(t, srv) + + watchDir := filepath.Join(dir, "watch") + os.MkdirAll(watchDir, 0o755) + original := filepath.Join(watchDir, "doc.txt") + os.WriteFile(original, []byte("version1"), 0o644) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + root := newRootCmd() + var stdout bytes.Buffer + root.SetOut(&stdout) + root.SetErr(&stdout) + root.SetContext(ctx) + root.SetArgs([]string{"put", watchDir, "--watch", "--interval", "50ms"}) + + errCh := make(chan error, 1) + go func() { errCh <- root.Execute() }() + + time.Sleep(300 * time.Millisecond) + initialRequests := requests.Load() + if initialRequests < 1 { + cancel() + <-errCh + t.Fatalf("expected initial upload, got %d requests", initialRequests) + } + + os.WriteFile(original, []byte("version2-different-content"), 0o644) + time.Sleep(200 * time.Millisecond) + cancel() + <-errCh + + afterModifyRequests := requests.Load() + if afterModifyRequests > initialRequests { + t.Errorf("changed file triggered %d additional upload(s); expected 0 (report only)", + afterModifyRequests-initialRequests) + } + if !strings.Contains(stdout.String(), "changed") { + t.Errorf("expected 'changed' in output, got: %s", stdout.String()) + } +} + +func TestWatchLocalGoneNoDelete(t *testing.T) { + var requests atomic.Int32 + var deleteCalls atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodDelete { + deleteCalls.Add(1) + w.WriteHeader(http.StatusMethodNotAllowed) + return + } + requests.Add(1) + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusCreated) + fmt.Fprintf(w, `{"file":{"id":"f-%d","name":"x","path":"/x","sha256":"a","size":1,"mime":"text/plain","index_status":"new"},"deduped":false}`, requests.Load()) + })) + defer srv.Close() + dir := setupWatchEnv(t, srv) + + watchDir := filepath.Join(dir, "watch") + os.MkdirAll(watchDir, 0o755) + vanishFile := filepath.Join(watchDir, "vanish.txt") + os.WriteFile(vanishFile, []byte("gone soon"), 0o644) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + root := newRootCmd() + var stdout bytes.Buffer + root.SetOut(&stdout) + root.SetErr(&stdout) + root.SetContext(ctx) + root.SetArgs([]string{"put", watchDir, "--watch", "--interval", "50ms"}) + + errCh := make(chan error, 1) + go func() { errCh <- root.Execute() }() + + time.Sleep(300 * time.Millisecond) + os.Remove(vanishFile) + time.Sleep(200 * time.Millisecond) + cancel() + <-errCh + + if deleteCalls.Load() != 0 { + t.Errorf("watcher issued %d DELETE call(s); expected 0 (no deletion propagation)", deleteCalls.Load()) + } + if !strings.Contains(stdout.String(), "local_gone") { + t.Errorf("expected 'local_gone' in output, got: %s", stdout.String()) + } +} + +func TestWatchDedupedCarriesServerPath(t *testing.T) { + var requests atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests.Add(1) + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusCreated) + _, _ = w.Write([]byte(`{"file":{"id":"existing-42","name":"dup.txt","path":"/OtherPlace/dup.txt","sha256":"x","size":1,"mime":"text/plain","index_status":"秒传"},"deduped":true}`)) + })) + defer srv.Close() + dir := setupWatchEnv(t, srv) + + watchDir := filepath.Join(dir, "watch") + os.MkdirAll(watchDir, 0o755) + os.WriteFile(filepath.Join(watchDir, "dup.txt"), []byte("dedup-content"), 0o644) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + root := newRootCmd() + var stdout bytes.Buffer + root.SetOut(&stdout) + root.SetErr(&stdout) + root.SetContext(ctx) + root.SetArgs([]string{"put", watchDir, "--watch", "--interval", "50ms"}) + + errCh := make(chan error, 1) + go func() { errCh <- root.Execute() }() + + time.Sleep(300 * time.Millisecond) + cancel() + <-errCh + + if !strings.Contains(stdout.String(), "deduped") { + t.Errorf("expected 'deduped' in output, got: %s", stdout.String()) + } + if !strings.Contains(stdout.String(), "existing-42") { + t.Errorf("expected server-returned file id 'existing-42' in output, got: %s", stdout.String()) + } + if !strings.Contains(stdout.String(), "/OtherPlace/dup.txt") { + t.Errorf("expected server-returned path '/OtherPlace/dup.txt' in output, got: %s", stdout.String()) + } +} + +func TestWatchReportPersistence(t *testing.T) { + var requests atomic.Int32 + srv := uploadMockServer(t, &requests, false) + defer srv.Close() + dir := setupWatchEnv(t, srv) + + watchDir := filepath.Join(dir, "watch") + os.MkdirAll(watchDir, 0o755) + os.WriteFile(filepath.Join(watchDir, "a.txt"), []byte("data"), 0o644) + + stateDir := filepath.Join(dir, "state") + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + root := newRootCmd() + var stdout bytes.Buffer + root.SetOut(&stdout) + root.SetErr(&stdout) + root.SetContext(ctx) + root.SetArgs([]string{"put", watchDir, "--watch", "--interval", "50ms"}) + + errCh := make(chan error, 1) + go func() { errCh <- root.Execute() }() + + time.Sleep(300 * time.Millisecond) + cancel() + <-errCh + + reportDir := filepath.Join(stateDir, "watch", "reports") + entries, err := os.ReadDir(reportDir) + if err != nil { + t.Fatalf("report dir: %v", err) + } + if len(entries) == 0 { + t.Fatal("expected at least one report file") + } + reportFile := filepath.Join(reportDir, entries[0].Name()) + b, _ := os.ReadFile(reportFile) + lines := strings.Split(strings.TrimSpace(string(b)), "\n") + if len(lines) == 0 { + t.Fatal("report file is empty") + } + var report cycleReport + if err := json.Unmarshal([]byte(lines[0]), &report); err != nil { + t.Fatalf("parse first report line: %v", err) + } + if report.Counts.Scanned < 1 { + t.Errorf("expected scanned >= 1, got %d", report.Counts.Scanned) + } +} + +func TestWatchSingleInstanceLock(t *testing.T) { + var requests atomic.Int32 + srv := uploadMockServer(t, &requests, false) + defer srv.Close() + dir := setupWatchEnv(t, srv) + + watchDir := filepath.Join(dir, "watch") + os.MkdirAll(watchDir, 0o755) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + root1 := newRootCmd() + root1.SetContext(ctx) + root1.SetArgs([]string{"put", watchDir, "--watch", "--interval", "50ms"}) + var stdout1 bytes.Buffer + root1.SetOut(&stdout1) + root1.SetErr(&stdout1) + + errCh := make(chan error, 1) + go func() { errCh <- root1.Execute() }() + + time.Sleep(100 * time.Millisecond) + + root2 := newRootCmd() + root2.SetArgs([]string{"put", watchDir, "--watch", "--interval", "50ms"}) + err := root2.Execute() + if err == nil { + cancel() + <-errCh + t.Fatal("expected lock error for second watcher") + } + var ce *cliError + if !isCliError(err, &ce) || ce.code != 1 { + t.Errorf("err = %v, want exit code 1", err) + } + + cancel() + <-errCh +} + +func TestWatchFormatJSON(t *testing.T) { + var requests atomic.Int32 + srv := uploadMockServer(t, &requests, false) + defer srv.Close() + dir := setupWatchEnv(t, srv) + + watchDir := filepath.Join(dir, "watch") + os.MkdirAll(watchDir, 0o755) + os.WriteFile(filepath.Join(watchDir, "a.txt"), []byte("data"), 0o644) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + root := newRootCmd() + var stdout bytes.Buffer + root.SetOut(&stdout) + root.SetErr(&stdout) + root.SetContext(ctx) + root.SetArgs([]string{"put", watchDir, "--watch", "--interval", "50ms", "--format", "json"}) + + errCh := make(chan error, 1) + go func() { errCh <- root.Execute() }() + + time.Sleep(300 * time.Millisecond) + cancel() + <-errCh + + output := stdout.String() + if !strings.Contains(output, `"counts"`) { + t.Errorf("expected JSON output with 'counts' key, got: %s", output) + } + if !strings.Contains(output, `"scanned"`) { + t.Errorf("expected JSON output with 'scanned' key, got: %s", output) + } +} + +func TestComputeSHA256(t *testing.T) { + dir := t.TempDir() + p := filepath.Join(dir, "test.txt") + os.WriteFile(p, []byte("hello world"), 0o644) + + got, err := computeSHA256(p) + if err != nil { + t.Fatal(err) + } + h := sha256.Sum256([]byte("hello world")) + want := hex.EncodeToString(h[:]) + if got != want { + t.Errorf("computeSHA256 = %s, want %s", got, want) + } +} + +func TestCursorRoundTrip(t *testing.T) { + dir := t.TempDir() + cur := watchCursor{ + Abs: "/tmp/test.txt", + Size: 42, + ModTime: "2026-01-01T00:00:00Z", + SHA256: "abc123", + FileID: "f-1", + IngestedAt: "2026-01-01T00:00:00Z", + } + if err := saveWatchCursor(dir, cur); err != nil { + t.Fatal(err) + } + loaded, ok := loadWatchCursor(dir, cur.Abs) + if !ok { + t.Fatal("cursor not found after save") + } + if loaded.FileID != cur.FileID || loaded.SHA256 != cur.SHA256 || loaded.Size != cur.Size { + t.Errorf("loaded = %+v, want %+v", loaded, cur) + } +} + +func TestLoadAllWatchCursors(t *testing.T) { + dir := t.TempDir() + c1 := watchCursor{Abs: "/a.txt", FileID: "f-1", SHA256: "h1"} + c2 := watchCursor{Abs: "/b.txt", FileID: "f-2", SHA256: "h2"} + _ = saveWatchCursor(dir, c1) + _ = saveWatchCursor(dir, c2) + + all := loadAllWatchCursors(dir) + if len(all) != 2 { + t.Fatalf("expected 2 cursors, got %d", len(all)) + } + if all["/a.txt"].FileID != "f-1" { + t.Errorf("cursor /a.txt = %+v", all["/a.txt"]) + } +} + +func TestClassifyUploadError(t *testing.T) { + tests := []struct { + err error + want string + }{ + {fmt.Errorf("connection refused"), failureNetwork}, + {fmt.Errorf("dial tcp: lookup"), failureNetwork}, + {fmt.Errorf("something else"), failureUploadRejected}, + } + for _, tt := range tests { + got := classifyUploadError(tt.err) + if got != tt.want { + t.Errorf("classifyUploadError(%v) = %q, want %q", tt.err, got, tt.want) + } + } +} + +func TestUpdateFailCounter(t *testing.T) { + allFail := cycleReport{ + Counts: cycleCounts{Scanned: 1, Failed: 1}, + Items: []cycleItem{{Status: "failed", FailureCode: failureAuth}}, + } + if got := updateFailCounter(0, allFail); got != 1 { + t.Errorf("counter after all-fail = %d, want 1", got) + } + if got := updateFailCounter(9, allFail); got != 10 { + t.Errorf("counter after 9+1 all-fail = %d, want 10", got) + } + + mixed := cycleReport{ + Counts: cycleCounts{Scanned: 2, Failed: 1, Unchanged: 1}, + } + if got := updateFailCounter(5, mixed); got != 0 { + t.Errorf("counter resets on non-all-fail = %d, want 0", got) + } + + retryable := cycleReport{ + Counts: cycleCounts{Scanned: 1, Failed: 1}, + Items: []cycleItem{{Status: "failed", FailureCode: failureNetwork}}, + } + if got := updateFailCounter(5, retryable); got != 0 { + t.Errorf("counter resets on retryable failure = %d, want 0", got) + } +} + +func TestPersistReportCap(t *testing.T) { + dir := t.TempDir() + absRoot := "/test/root" + for i := 0; i < maxReportLines+50; i++ { + r := cycleReport{ + Timestamp: fmt.Sprintf("2026-01-01T00:%02d:00Z", i), + Counts: cycleCounts{Scanned: 1}, + } + if err := persistReport(dir, absRoot, r); err != nil { + t.Fatal(err) + } + } + p := reportPath(dir, absRoot) + b, _ := os.ReadFile(p) + lines := strings.Split(strings.TrimSpace(string(b)), "\n") + if len(lines) != maxReportLines { + t.Errorf("report lines = %d, want %d", len(lines), maxReportLines) + } +} + +func isCliError(err error, ce **cliError) bool { + if err == nil { + return false + } + return errors.As(err, ce) +} + +func TestWatchGiveUpAfterTenFailCycles(t *testing.T) { + var requests atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests.Add(1) + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusUnauthorized) + _, _ = io.WriteString(w, `{"error":"unauthorized"}`) + })) + defer srv.Close() + dir := setupWatchEnv(t, srv) + + watchDir := filepath.Join(dir, "watch") + os.MkdirAll(watchDir, 0o755) + os.WriteFile(filepath.Join(watchDir, "a.txt"), []byte("data"), 0o644) + + root := newRootCmd() + var stdout bytes.Buffer + root.SetOut(&stdout) + root.SetErr(&stdout) + root.SetArgs([]string{"put", watchDir, "--watch", "--interval", "10ms"}) + + err := root.Execute() + if err == nil { + t.Fatal("expected give-up error after 10 consecutive all-fail cycles") + } + var ce *cliError + if !errors.As(err, &ce) || ce.code != 3 { + t.Errorf("err = %v, want exit code 3 (auth)", err) + } +}