// ingestion/internal/watcher/watcher.go package watcher import ( "context" "fmt" "io" "log/slog" "os" "path/filepath" "strings" "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. if _, err := os.Stat(path + ".processed"); err == nil { return nil } if _, err := os.Stat(path + ".failed"); err == nil { return nil } if err := processFile(ctx, cfg, root, rawDir, path, date); err != nil { 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 { 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 { slog.Error("watcher: failed to write .processed marker", "error", err) } slog.Info("watcher: file processed", "file", filename, "source", source) return nil } // 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) }