Merge pull request 'feat(watcher): inbox zero — archived raw/ originals are removed, not copied' (#206) from wt/inbox-zero into main

ingestion/internal/watcher/watcher.go

@@ -9,6 +9,7 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 	"time"
 	"unicode"
 
@@ -21,11 +22,19 @@ type Config struct {
 	BrainDir string
 	Interval time.Duration
 	Pipeline pipeline.Config
+
+	// stuck remembers notes that were ingested but could neither be removed
+	// from raw/ nor marked (e.g. a read-only raw/), so later polls skip them
+	// instead of ingesting them again. In memory only: a restart retries once.
+	stuck *sync.Map
 }
 
 // Start launches the watcher in a background goroutine.
 // It returns immediately. The watcher stops when ctx is cancelled.
 func Start(ctx context.Context, cfg Config) {
+	if cfg.stuck == nil {
+		cfg.stuck = &sync.Map{}
+	}
 	go func() {
 		ticker := time.NewTicker(cfg.Interval)
 		defer ticker.Stop()
@@ -108,6 +117,12 @@ func processDir(ctx context.Context, cfg Config, date string) []error {
 			return nil
 		}
 
+		if cfg.stuck != nil {
+			if _, skip := cfg.stuck.Load(path); skip {
+				return nil
+			}
+		}
+
 		// Skip files that have already been processed or permanently failed.
 		if _, err := os.Stat(path + ".processed"); err == nil {
 			return nil
@@ -127,9 +142,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)
@@ -138,6 +156,10 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d
 		return fmt.Errorf("relative path: %w", err)
 	}
 
+	before, err := root.Lstat(rel)
+	if err != nil {
+		return fmt.Errorf("stat before ingest: %w", err)
+	}
 	content, err := extract.TextIn(root, rel)
 	if err != nil {
 		return fmt.Errorf("extract text: %w", err)
@@ -154,9 +176,11 @@ 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)
+		if changedSince(root, rel, before) {
+			slog.Warn("watcher: note changed during ingest; leaving it in raw/ for the next poll", "file", filename)
+			return nil
 		}
+		retire(cfg, root, rel, path, ".failed", runErr.Error())
 
 		slog.Warn("watcher: file failed", "file", filename, "error", runErr)
 
@@ -182,14 +206,45 @@ 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)
+	if changedSince(root, rel, before) {
+		slog.Warn("watcher: note changed during ingest; leaving it in raw/ for the next poll", "file", filename)
+		return nil
 	}
+	retire(cfg, root, rel, path, ".processed", date)
 
 	slog.Info("watcher: file processed", "file", filename, "source", source)
 	return nil
 }
 
+// changedSince reports whether the note at rel differs (size or mtime) from
+// before. A note replaced mid-ingest (Syncthing) was extracted from older
+// content than it now holds, so it must not be removed. Limits: a same-size
+// rewrite with a restored mtime is not detected (the new bytes then survive
+// only in the archive copy), nor a change between this check and Remove.
+func changedSince(root *os.Root, rel string, before os.FileInfo) bool {
+	now, err := root.Lstat(rel)
+	if err != nil {
+		return true
+	}
+	return now.Size() != before.Size() || !now.ModTime().Equal(before.ModTime())
+}
+
+// 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(cfg Config, root *os.Root, rel, path, markerSuffix, 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(path+markerSuffix, []byte(markerBody), 0o644); err != nil {
+		slog.Error("watcher: failed to write marker; skipping this note until restart", "file", rel, "error", err)
+		if cfg.stuck != nil {
+			cfg.stuck.Store(path, struct{}{})
+		}
+	}
+}
+
 // deriveSource turns a filename into a human-readable source name.
 // "shape-up-book.md" → "Shape Up Book"
 func deriveSource(filename string) string {

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"))
@@ -390,3 +387,86 @@ func TestProcessDir_DoesNotHangUnderAFIFOSwapRace(t *testing.T) {
 		t.Fatalf("the watcher hung on a swapped-in FIFO after %d iterations", iters.Load())
 	}
 }
+
+// A note whose removal AND marker write both fail (read-only raw/) must not be
+// re-ingested on every poll: each pass would write another archive copy.
+func TestProcessDir_DoesNotReingestANoteItCannotRetire(t *testing.T) {
+	brainDir := setupBrainDir(t)
+	rawDir := filepath.Join(brainDir, "raw")
+	require.NoError(t, os.WriteFile(filepath.Join(rawDir, "stuck.md"), []byte("Content."), 0o644))
+	// processed/ must exist and stay writable; only raw/ itself is locked.
+	require.NoError(t, os.MkdirAll(filepath.Join(rawDir, "processed"), 0o755))
+	require.NoError(t, os.Chmod(rawDir, 0o555))
+	t.Cleanup(func() { _ = os.Chmod(rawDir, 0o755) })
+
+	var calls atomic.Int32
+	cfg := Config{
+		BrainDir: brainDir,
+		Pipeline: pipeline.Config{
+			Complete: func(ctx context.Context, s, u string) (string, error) {
+				calls.Add(1)
+				return successComplete(pipeline.RawPage{Title: "Stuck", Type: "source", Subtype: "article", Content: "x\n"})(ctx, s, u)
+			},
+			Schema: "# Schema",
+		},
+		stuck: &sync.Map{},
+	}
+	for i := 0; i < 3; i++ {
+		processDir(context.Background(), cfg, "2026-10-19")
+	}
+	assert.Equal(t, int32(1), calls.Load(), "a note that cannot be retired is ingested once, not once per poll")
+}
+
+// If the archive copy cannot be written, the original is the only copy left:
+// it must stay in raw/.
+func TestProcessDir_KeepsTheOriginalWhenTheArchiveCopyFails(t *testing.T) {
+	brainDir := setupBrainDir(t)
+	rawDir := filepath.Join(brainDir, "raw")
+	note := filepath.Join(rawDir, "keep-me.md")
+	require.NoError(t, os.WriteFile(note, []byte("Content."), 0o644))
+	// The dated archive dir exists but is read-only: MkdirAll succeeds, the
+	// copy itself (os.Create) fails.
+	dated := filepath.Join(rawDir, "processed", "2026-10-19")
+	require.NoError(t, os.MkdirAll(dated, 0o755))
+	require.NoError(t, os.Chmod(dated, 0o555))
+	t.Cleanup(func() { _ = os.Chmod(dated, 0o755) })
+
+	cfg := Config{
+		BrainDir: brainDir,
+		Pipeline: pipeline.Config{
+			Complete: successComplete(pipeline.RawPage{Title: "Keep Me", Type: "source", Subtype: "article", Content: "x\n"}),
+			Schema:   "# Schema",
+		},
+	}
+	errs := processDir(context.Background(), cfg, "2026-10-19")
+
+	assert.NotEmpty(t, errs, "the failed archive is reported")
+	_, err := os.Stat(note)
+	assert.NoError(t, err, "original stays in raw/ when it could not be archived")
+}
+
+// If the note is replaced (e.g. by Syncthing) while it is being ingested, the
+// wiki page was extracted from the old content; removing the original would
+// leave the new content only in the archive and never re-ingest it. Observes
+// size/mtime change only: a same-size rewrite with a restored mtime is not caught.
+func TestProcessDir_KeepsANoteThatChangedWhileItWasBeingIngested(t *testing.T) {
+	brainDir := setupBrainDir(t)
+	note := filepath.Join(brainDir, "raw", "edited.md")
+	require.NoError(t, os.WriteFile(note, []byte("old"), 0o644))
+
+	cfg := Config{
+		BrainDir: brainDir,
+		Pipeline: pipeline.Config{
+			Complete: func(ctx context.Context, s, u string) (string, error) {
+				require.NoError(t, os.WriteFile(note, []byte("newer and longer content"), 0o644))
+				return successComplete(pipeline.RawPage{Title: "Edited", Type: "source", Subtype: "article", Content: "x\n"})(ctx, s, u)
+			},
+			Schema: "# Schema",
+		},
+	}
+	processDir(context.Background(), cfg, "2026-10-19")
+
+	got, err := os.ReadFile(note)
+	require.NoError(t, err, "a note that changed during ingest must stay in raw/")
+	assert.Equal(t, "newer and longer content", string(got))
+}