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/brain"
	"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
	"github.com/mathiasbq/hyperguild/ingestion/internal/extract"
	"github.com/mathiasbq/hyperguild/ingestion/internal/pipeline"
	"github.com/mathiasbq/hyperguild/ingestion/internal/wing"
)

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

	// MaxNoteBytes refuses a note larger than this before anything is read into
	// memory (hyperguild#198). 0 means defaultMaxNoteBytes.
	MaxNoteBytes int64

	// Resolver, when set, decides each note's wing (hyperguild#205, #208): a
	// deliberate subfolder/frontmatter wing, else the default wing. A note it
	// cannot place is parked in raw/needs-review/ for a human. Nil keeps the
	// retired layout. Set it only to a non-nil value: a nil pointer inside the
	// interface is not nil.
	Resolver wing.Placer
}

// defaultMaxNoteBytes bounds a single note: extraction and the archive copy both
// hold it in memory.
const defaultMaxNoteBytes = 64 << 20

// errTooLarge marks a note refused for its size; like a refused write it is
// reported once and not retried until the note changes.
var errTooLarge = errors.New("note too large")

// 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" || name == "needs-review" {
				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) || errors.Is(err, errTooLarge)) {
				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, 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)
	rel, err := filepath.Rel(rawDir, path)
	if err != nil {
		return fmt.Errorf("relative path: %w", err)
	}

	before, err := root.Lstat(rel)
	if err != nil {
		return fmt.Errorf("stat before ingest: %w", err)
	}
	limit := cfg.MaxNoteBytes
	if limit <= 0 {
		limit = defaultMaxNoteBytes
	}
	if before.Size() > limit {
		return fmt.Errorf("%s is %d bytes, over the %d byte cap: %w", filename, before.Size(), limit, errTooLarge)
	}
	content, err := extract.TextIn(root, rel)
	if err != nil {
		return fmt.Errorf("extract text: %w", err)
	}

	pcfg := cfg.Pipeline
	if cfg.Resolver != nil {
		d := cfg.Resolver.Resolve(ctx, wing.Note{Dir: topDir(rel), FrontmatterWing: frontmatterWing(content), Text: content})
		if d.Source == wing.Exception {
			return parkForReview(cfg, root, rel, path, filename, content, d, date, before)
		}
		pcfg.Wing = d.Wing
		slog.Info("watcher: wing resolved", "file", filename, "wing", d.Wing, "how", d.Source, "reason", d.Reason)
	}
	_, runErr := pipeline.Run(ctx, pcfg, 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 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(), before)

		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 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, before)

	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 written through the brain root (#158), and if
// even that fails the note is remembered, with the size and mtime it had, so it
// is not ingested again until it changes.
func retire(cfg Config, root *os.Root, rel, path, markerSuffix, markerBody string, before os.FileInfo) {
	rmErr := root.Remove(rel)
	if rmErr == nil {
		return
	}
	slog.Error("watcher: failed to remove archived original; writing marker", "file", rel, "error", rmErr)
	if err := putBrainFile(cfg.BrainDir, "raw/"+filepath.ToSlash(rel)+markerSuffix, []byte(markerBody)); err != nil {
		slog.Error("watcher: failed to write marker; skipping this note until it changes", "file", rel, "error", err)
		rememberRefused(path, before)
	}
}

// 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)
}

// topDir is the first directory of a raw/-relative path, "" for a top-level file.
func topDir(rel string) string {
	if i := strings.IndexByte(filepath.ToSlash(rel), '/'); i > 0 {
		return filepath.ToSlash(rel)[:i]
	}
	return ""
}

// frontmatterWing reads `wing:` from a note's leading frontmatter, "" if none.
func frontmatterWing(content string) string {
	content = strings.ReplaceAll(content, "\r\n", "\n")
	if !strings.HasPrefix(content, "---\n") {
		return ""
	}
	block, _, _ := strings.Cut(content[4:], "\n---")
	for _, line := range strings.Split(block, "\n") {
		if key, val, ok := strings.Cut(line, ":"); ok && strings.TrimSpace(key) == "wing" {
			return strings.Trim(strings.TrimSpace(val), `"'`)
		}
	}
	return ""
}

// parkForReview moves a note the resolver could not place into
// raw/needs-review/, recording the proposal in its frontmatter so a human can
// approve it by moving it into raw/<wing>/. Nothing is extracted or written to
// the wiki.
//
// The parked copy is created through the brain root (real directories only,
// O_EXCL, never overwriting) and is the only copy once the original is retired,
// so the original is removed only after that write succeeds and only if it has
// not changed since it was read: a newer version delivered while the resolver
// ran stays in raw/ for the next poll.
func parkForReview(cfg Config, root *os.Root, rel, path, filename, content string, d wing.Decision, date string, before os.FileInfo) error {
	var data []byte
	if strings.ToLower(filepath.Ext(filename)) == ".pdf" {
		// Binary: park the bytes untouched; the proposal is in log.md only.
		in, _, err := extract.OpenRegular(root, rel)
		if err != nil {
			return fmt.Errorf("park pdf: %w", err)
		}
		data, err = io.ReadAll(in)
		_ = in.Close()
		if err != nil {
			return fmt.Errorf("park pdf: %w", err)
		}
	} else {
		data = []byte(withProposal(content, d))
	}
	if err := createParked(cfg.BrainDir, filename, data); err != nil {
		return fmt.Errorf("park note: %w", err)
	}
	if err := appendWatcherLog(cfg.BrainDir, filename, fmt.Errorf("needs review (proposed wing %q): %s", d.Wing, d.Reason), date); err != nil {
		slog.Error("watcher: failed to write log entry", "error", err)
	}
	slog.Info("watcher: parked for review", "file", filename, "proposed_wing", d.Wing, "reason", d.Reason)
	if changedSince(root, rel, before) {
		slog.Warn("watcher: note changed while it was being resolved; leaving the newer version in raw/", "file", filename)
		return nil
	}
	retire(cfg, root, rel, path, ".processed", date, before)
	return nil
}

// proposalKeys are the frontmatter keys the resolver owns in a parked note.
var proposalKeys = map[string]bool{"proposed_wing": true, "proposal_confidence": true, "proposal_reason": true}

// withProposal adds the resolver's proposal to a note's frontmatter, creating
// the block when the note has none. Every value is on one line: the wing only
// if it is a plain wing name (else it appears in the reason alone), strings
// Go-quoted. Any proposal keys the note already carried are dropped first, so a
// note cannot pre-empt the resolver's (frontmatter reads are first-wins).
// Line endings are normalised to LF so a CRLF note keeps one valid block.
func withProposal(content string, d wing.Decision) string {
	content = strings.ReplaceAll(content, "\r\n", "\n")
	reason := d.Reason
	var kv strings.Builder
	if d.Wing != "" && brain.Sanitise(d.Wing) == d.Wing {
		fmt.Fprintf(&kv, "proposed_wing: %s\n", d.Wing)
	} else if d.Wing != "" {
		reason = fmt.Sprintf("proposed wing %q is not a plain wing name; %s", d.Wing, reason)
	}
	fmt.Fprintf(&kv, "proposal_confidence: %.2f\nproposal_reason: %q\n", d.Confidence, reason)

	if !strings.HasPrefix(content, "---\n") {
		return "---\n" + kv.String() + "---\n" + content
	}
	block, rest, found := cutFence(content[4:])
	if !found {
		return "---\n" + kv.String() + "---\n" + content
	}
	var kept []string
	for _, line := range strings.Split(block, "\n") {
		if key, _, ok := strings.Cut(line, ":"); ok && proposalKeys[strings.TrimSpace(key)] && !strings.HasPrefix(line, " ") {
			continue
		}
		kept = append(kept, line)
	}
	return "---\n" + kv.String() + strings.Join(kept, "\n") + "\n---" + rest
}

// cutFence splits s at the first closing fence: a line that is exactly "---".
// "----" or "---x" is not a fence, matching how the frontmatter readers parse.
func cutFence(s string) (block, rest string, found bool) {
	if i := strings.Index(s, "\n---\n"); i >= 0 {
		return s[:i], s[i+4:], true
	}
	if strings.HasSuffix(s, "\n---") {
		return s[:len(s)-4], "", true
	}
	return "", "", false
}