diff --git a/ingestion/internal/watcher/watcher.go b/ingestion/internal/watcher/watcher.go index b5c41b3..204da31 100644 --- a/ingestion/internal/watcher/watcher.go +++ b/ingestion/internal/watcher/watcher.go @@ -127,9 +127,12 @@ func processDir(ctx context.Context, cfg Config, date string) []error { return errs } -// processFile reads a file, calls pipeline.Run, copies it to processed/ or failed/, -// and writes a marker file next to the original so the watcher skips it next poll. -// The original file is never deleted, keeping Syncthing-connected vaults (e.g. Obsidian) intact. +// processFile reads a file, calls pipeline.Run, and archives it into processed/ +// or failed/. The original is then removed from raw/ so that raw/ only ever +// holds unprocessed input (inbox zero). The removal syncs to Syncthing peers; +// the archive copy under raw/processed/ or raw/failed/ is what they keep. A +// marker is written next to the original only if the removal fails, so a +// stuck file is not re-ingested every poll. func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, date string) error { filename := filepath.Base(path) source := deriveSource(filename) @@ -154,9 +157,7 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d if cpErr := copyFromRoot(root, rel, dest); cpErr != nil { return fmt.Errorf("copy to failed: %w", cpErr) } - if mkErr := os.WriteFile(path+".failed", []byte(runErr.Error()), 0o644); mkErr != nil { - slog.Error("watcher: failed to write .failed marker", "error", mkErr) - } + retire(root, rel, path+".failed", runErr.Error()) slog.Warn("watcher: file failed", "file", filename, "error", runErr) @@ -182,14 +183,25 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d if err := copyFromRoot(root, rel, dest); err != nil { return fmt.Errorf("copy to processed: %w", err) } - if err := os.WriteFile(path+".processed", []byte(date), 0o644); err != nil { - slog.Error("watcher: failed to write .processed marker", "error", err) - } + retire(root, rel, path+".processed", date) slog.Info("watcher: file processed", "file", filename, "source", source) return nil } +// retire removes an archived original from raw/ (inbox zero). If the removal +// fails it falls back to a marker file so the watcher still skips the file. +func retire(root *os.Root, rel, marker, markerBody string) { + rmErr := root.Remove(rel) + if rmErr == nil { + return + } + slog.Error("watcher: failed to remove archived original; writing marker", "file", rel, "error", rmErr) + if err := os.WriteFile(marker, []byte(markerBody), 0o644); err != nil { + slog.Error("watcher: failed to write marker", "marker", marker, "error", err) + } +} + // deriveSource turns a filename into a human-readable source name. // "shape-up-book.md" → "Shape Up Book" func deriveSource(filename string) string { diff --git a/ingestion/internal/watcher/watcher_test.go b/ingestion/internal/watcher/watcher_test.go index 899b2bf..4e2c478 100644 --- a/ingestion/internal/watcher/watcher_test.go +++ b/ingestion/internal/watcher/watcher_test.go @@ -89,13 +89,12 @@ func TestStart_ProcessesFile(t *testing.T) { } require.True(t, found, "file should be copied to processed/") - // Original file should still exist (copy, not move — keeps Obsidian vault intact). + // Inbox zero: once archived, the original leaves raw/ and no marker is + // left behind, so what remains in raw/ is exactly the unprocessed set. _, err := os.Stat(rawFile) - assert.NoError(t, err, "original file should remain in raw/") - - // A .processed marker should exist next to the original. + assert.ErrorIs(t, err, os.ErrNotExist, "original should be removed from raw/ once archived") _, err = os.Stat(rawFile + ".processed") - assert.NoError(t, err, ".processed marker should be written") + assert.ErrorIs(t, err, os.ErrNotExist, "no .processed marker: the archive copy is the record") // Wiki page should exist. wikiPath := filepath.Join(brainDir, "wiki", "sources", "shape-up-book.md") @@ -142,13 +141,11 @@ func TestStart_MovesToFailedOnError(t *testing.T) { } require.True(t, found, "file should be copied to failed/") - // Original file should still exist (copy, not move — keeps Obsidian vault intact). + // Inbox zero: a quarantined original leaves raw/ too, no marker beside it. _, err := os.Stat(rawFile) - assert.NoError(t, err, "original file should remain in raw/") - - // A .failed marker should exist next to the original. + assert.ErrorIs(t, err, os.ErrNotExist, "original should be removed from raw/ once quarantined") _, err = os.Stat(rawFile + ".failed") - assert.NoError(t, err, ".failed marker should be written") + assert.ErrorIs(t, err, os.ErrNotExist, "no .failed marker: failed/ is the record") // log.md should contain a watcher error entry. logContent, err := os.ReadFile(filepath.Join(brainDir, "log.md"))