hyperguild/ingestion/internal/watcher/watcher.go

// ingestion/internal/watcher/watcher.go
package watcher

import (
	"context"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"
	"unicode"

	"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
	"github.com/mathiasbq/hyperguild/ingestion/internal/extract"
	"github.com/mathiasbq/hyperguild/ingestion/internal/pipeline"
)

// Config holds watcher configuration.
type Config struct {
	BrainDir string
	Interval time.Duration
	Pipeline pipeline.Config
}

// 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) {
	go func() {
		ticker := time.NewTicker(cfg.Interval)
		defer ticker.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
				date := time.Now().UTC().Format("2006-01-02")
				errs := processDir(ctx, cfg, date)
				for _, err := range errs {
					slog.Error("watcher: error processing file", "error", err)
				}
			}
		}
	}()
}

// processDir walks brain/raw/, processes each eligible file, returns any errors encountered.
func processDir(ctx context.Context, cfg Config, date string) []error {
	rawDir := filepath.Join(cfg.BrainDir, "raw")

	// Every read goes through an os.Root on raw/ (hyperguild#153): raw/ is
	// writable via Syncthing, and a file swapped for a symlink between the walk
	// and the read must not lead out of it. A symlinked raw/ is refused.
	if fi, lerr := os.Lstat(rawDir); lerr != nil {
		if os.IsNotExist(lerr) {
			return nil
		}
		return []error{fmt.Errorf("stat raw dir: %w", lerr)}
	} else if fi.Mode()&os.ModeSymlink != 0 || !fi.IsDir() {
		return []error{fmt.Errorf("raw dir %s is not a real directory; refusing to ingest from it", rawDir)}
	}
	root, err := os.OpenRoot(rawDir)
	if err != nil {
		return []error{fmt.Errorf("open raw root: %w", err)}
	}
	defer func() { _ = root.Close() }()

	var errs []error
	err = filepath.WalkDir(rawDir, func(path string, d os.DirEntry, err error) error {
		if err != nil {
			return err
		}

		// Skip the root itself.
		if path == rawDir {
			return nil
		}

		// Skip processed/, failed/, and rejected/ subdirectories entirely.
		// rejected/ (hyperguild#81, brain_reject) holds notes a human already
		// looked at and declined — walking into it would run the extraction
		// pipeline over a note and publish it to the wiki, the exact opposite
		// of the human's decision. The whole reason RejectNote moves a note
		// there instead of leaving it in raw/ was to keep it out of every
		// path that treats raw/ as "awaiting processing" — this is that path.
		if d.IsDir() {
			name := d.Name()
			if name == "processed" || name == "failed" || name == "rejected" {
				return filepath.SkipDir
			}
			return nil
		}

		// Only process supported extensions.
		ext := strings.ToLower(filepath.Ext(path))
		if ext != ".md" && ext != ".txt" && ext != ".pdf" {
			return nil
		}

		// AutoTunnel's own fuzzy-match human-review queue, not source
		// material to extract (hyperguild#88) — same exclusion as
		// api.ListPending already applies when listing raw/ for promotion.
		if strings.HasPrefix(d.Name(), "tunnel-candidates-") {
			return nil
		}
		// Only regular files are notes; a symlink's target may be anywhere.
		if !d.Type().IsRegular() {
			return nil
		}

		// Skip files that have already been processed or permanently failed.
		// Any entry at the marker path counts (Lstat, not Stat): a dangling
		// marker symlink made os.Stat report nothing and the note was
		// reprocessed every interval (hyperguild#158).
		if _, err := os.Lstat(path + ".processed"); err == nil {
			return nil
		}
		if _, err := os.Lstat(path + ".failed"); err == nil {
			return nil
		}
		// A note whose archive copy or marker was refused is not retried
		// until it changes: a refusal has no marker to stop the next round,
		// and each round can cost an LLM call (hyperguild#158).
		info, ierr := d.Info()
		if ierr == nil && refusedUnchanged(path, info) {
			return nil
		}

		if err := processFile(ctx, cfg, root, rawDir, path, date); err != nil {
			if ierr == nil && errors.Is(err, brainfs.ErrNotRegular) {
				rememberRefused(path, info)
				slog.Error("watcher: note refused; not retried until it changes", "file", filepath.Base(path), "error", err)
			}
			errs = append(errs, fmt.Errorf("process %s: %w", filepath.Base(path), err))
		}
		return nil
	})
	if err != nil {
		errs = append(errs, fmt.Errorf("walk raw dir: %w", err))
	}
	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.
func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, date string) error {
	filename := filepath.Base(path)
	source := deriveSource(filename)
	rel, err := filepath.Rel(rawDir, path)
	if err != nil {
		return fmt.Errorf("relative path: %w", err)
	}

	content, err := extract.TextIn(root, rel)
	if err != nil {
		return fmt.Errorf("extract text: %w", err)
	}

	_, runErr := pipeline.Run(ctx, cfg.Pipeline, cfg.BrainDir, content, source, false)
	if runErr != nil {
		// Copy to failed/ and leave a .failed marker so we don't retry. Both
		// written through the brain root (hyperguild#158).
		if cpErr := copyFromRoot(root, rel, cfg.BrainDir, "raw/failed/"+filename); cpErr != nil {
			return fmt.Errorf("copy to failed: %w", cpErr)
		}
		if mkErr := putBrainFile(cfg.BrainDir, "raw/"+filepath.ToSlash(rel)+".failed", []byte(runErr.Error())); mkErr != nil {
			if errors.Is(mkErr, brainfs.ErrNotRegular) {
				return fmt.Errorf("write .failed marker: %w", mkErr)
			}
			slog.Error("watcher: failed to write .failed marker", "error", mkErr)
		}

		slog.Warn("watcher: file failed", "file", filename, "error", runErr)

		if logErr := appendWatcherLog(cfg.BrainDir, filename, runErr, date); logErr != nil {
			slog.Error("watcher: failed to write log entry", "error", logErr)
		}
		// Return nil: quarantine succeeded; error already logged.
		return nil
	}

	// Copy to processed/YYYY-MM-DD/ and leave a .processed marker so we don't re-ingest.
	// Written through the brain root (hyperguild#158).
	dest := "raw/processed/" + date + "/" + filename
	if _, err := os.Lstat(filepath.Join(cfg.BrainDir, filepath.FromSlash(dest))); err == nil {
		// Archive copy already exists; append timestamp to avoid overwriting.
		ext := filepath.Ext(filename)
		base := strings.TrimSuffix(filename, ext)
		dest = "raw/processed/" + date + "/" + base + "-" + time.Now().UTC().Format("150405") + ext
	}
	if err := copyFromRoot(root, rel, cfg.BrainDir, dest); err != nil {
		return fmt.Errorf("copy to processed: %w", err)
	}
	if err := putBrainFile(cfg.BrainDir, "raw/"+filepath.ToSlash(rel)+".processed", []byte(date)); err != nil {
		if errors.Is(err, brainfs.ErrNotRegular) {
			return fmt.Errorf("write .processed marker: %w", err)
		}
		slog.Error("watcher: failed to write .processed marker", "error", err)
	}

	slog.Info("watcher: file processed", "file", filename, "source", source)
	return nil
}

// refused remembers notes whose archive copy or marker write was refused,
// by path, with the size and mtime they had, for the life of the process.
var refused sync.Map

type refusedStamp struct {
	size  int64
	mtime time.Time
}

func rememberRefused(path string, info os.FileInfo) {
	refused.Store(path, refusedStamp{size: info.Size(), mtime: info.ModTime()})
}

func refusedUnchanged(path string, info os.FileInfo) bool {
	v, ok := refused.Load(path)
	if !ok {
		return false
	}
	st := v.(refusedStamp)
	if st.size == info.Size() && st.mtime.Equal(info.ModTime()) {
		return true
	}
	refused.Delete(path)
	return false
}

// deriveSource turns a filename into a human-readable source name.
// "shape-up-book.md" → "Shape Up Book"
func deriveSource(filename string) string {
	// Strip extension.
	name := strings.TrimSuffix(filename, filepath.Ext(filename))
	// Split on hyphens.
	words := strings.Split(name, "-")
	// Title-case each word.
	for i, w := range words {
		if w == "" {
			continue
		}
		runes := []rune(w)
		runes[0] = unicode.ToUpper(runes[0])
		words[i] = string(runes)
	}
	return strings.Join(words, " ")
}

// appendWatcherLog appends a watcher error entry to brain/log.md.
func appendWatcherLog(brainDir, filename string, runErr error, date string) error {
	entry := fmt.Sprintf("## %s — watcher error\n\n- **File:** %s\n- **Error:** %s\n\n",
		date, filename, runErr.Error())

	if err := brainfs.AppendFile(brainDir, "log.md", []byte(entry)); err != nil {
		return fmt.Errorf("write log: %w", err)
	}
	return nil
}

// putBrainFile creates or replaces a brain-relative file through the brain
// root, never through a symlinked directory or file (hyperguild#158).
func putBrainFile(brainDir, rel string, data []byte) error {
	return brainfs.PutFile(brainDir, rel, data)
}

// copyFromRoot archives a raw/ file read through the root, refusing anything
// that is not a regular file, so a swapped-in symlink cannot copy an outside
// file into raw/processed/ or raw/failed/ (hyperguild#153).
func copyFromRoot(root *os.Root, rel, brainDir, dstRel string) error {
	fi, err := root.Lstat(rel)
	if err != nil {
		return err
	}
	if !fi.Mode().IsRegular() {
		return fmt.Errorf("%s: not a regular file", rel)
	}
	in, _, err := extract.OpenRegular(root, rel)
	if err != nil {
		return err
	}
	defer func() { _ = in.Close() }()
	data, err := io.ReadAll(in)
	if err != nil {
		return err
	}
	return putBrainFile(brainDir, dstRel, data)
}