package hotfolder import ( "context" "crypto/sha256" "encoding/hex" "fmt" "os" "path/filepath" "github.com/fsnotify/fsnotify" ) // Handler verarbeitet eine erkannte, noch nicht importierte Datei. // Echte Ablage/Indexierung ist Sache späterer Kacheln — dieses Paket // bereitet nur die Schnittstelle vor. type Handler interface { ProcessFile(ctx context.Context, tenantSlug, mailboxName, filename string, content []byte) error } // Watcher überwacht EIN Hot-Folder-Verzeichnis für EINEN Mandanten/EIN // Postfach (Akzeptanzkriterium 1: Zuordnung ist strukturell — welches // Verzeichnis zu welchem Mandanten/Postfach gehört, entscheidet der // Aufrufer beim Konfigurieren des Watchers, nicht dieses Paket anhand // von Dateiinhalten). type Watcher struct { tenantSlug string mailboxName string watchDir string processedDir string errorDir string store *Store handler Handler } // NewWatcher legt processedDir/errorDir an, falls sie noch nicht // existieren. func NewWatcher(tenantSlug, mailboxName, watchDir, processedDir, errorDir string, store *Store, handler Handler) (*Watcher, error) { for _, dir := range []string{watchDir, processedDir, errorDir} { if err := os.MkdirAll(dir, 0o755); err != nil { return nil, fmt.Errorf("hotfolder: verzeichnis %s anlegen: %w", dir, err) } } return &Watcher{ tenantSlug: tenantSlug, mailboxName: mailboxName, watchDir: watchDir, processedDir: processedDir, errorDir: errorDir, store: store, handler: handler, }, nil } // ScanResult fasst einen abgeschlossenen Scan-Durchlauf zusammen. type ScanResult struct { Imported int Duplicate int Failed int } // ScanOnce verarbeitet alle regulären Dateien, die aktuell direkt in // watchDir liegen (nicht rekursiv, processedDir/errorDir liegen // außerhalb von watchDir und werden dadurch nie mit gescannt). Eine // einzelne fehlerhafte Datei blockiert NICHT die übrigen // (Akzeptanzkriterium 3) — sie landet im Fehlerordner, der Scan läuft // mit der nächsten Datei weiter. func (w *Watcher) ScanOnce(ctx context.Context) (ScanResult, error) { entries, err := os.ReadDir(w.watchDir) if err != nil { return ScanResult{}, fmt.Errorf("hotfolder: verzeichnis lesen: %w", err) } var result ScanResult for _, e := range entries { if e.IsDir() { continue } if err := ctx.Err(); err != nil { return result, err } outcome := w.processOne(ctx, e.Name()) switch outcome { case outcomeImported: result.Imported++ case outcomeDuplicate: result.Duplicate++ case outcomeFailed: result.Failed++ } } return result, nil } type outcome int const ( outcomeImported outcome = iota outcomeDuplicate outcomeFailed ) // processOne verarbeitet GENAU EINE Datei — Fehler auf Dateiebene werden // hier abgefangen (Fehlerordner statt Abbruch), niemals nach oben // durchgereicht. func (w *Watcher) processOne(ctx context.Context, filename string) outcome { fullPath := filepath.Join(w.watchDir, filename) content, err := os.ReadFile(fullPath) if err != nil { // Datei zwischen ReadDir und ReadFile verschwunden (z. B. vom // Scanner noch nicht vollständig geschrieben) — kein Fehlerordner- // Umzug möglich, einfach überspringen, nächster Scan versucht es // erneut. return outcomeFailed } hash := sha256.Sum256(content) contentHash := hex.EncodeToString(hash[:]) alreadyDone, err := w.store.IsProcessed(ctx, w.tenantSlug, w.mailboxName, contentHash) if err != nil { w.moveTo(fullPath, w.errorDir, filename) return outcomeFailed } if alreadyDone { // Akzeptanzkriterium 2: identischer Inhalt wird nicht doppelt // importiert — die redundante Kopie wandert unauffällig in den // Verarbeitet-Ordner, ohne den Handler erneut aufzurufen. w.moveTo(fullPath, w.processedDir, filename) return outcomeDuplicate } if err := w.handler.ProcessFile(ctx, w.tenantSlug, w.mailboxName, filename, content); err != nil { w.moveTo(fullPath, w.errorDir, filename) return outcomeFailed } if err := w.store.MarkProcessed(ctx, w.tenantSlug, w.mailboxName, contentHash, filename); err != nil { w.moveTo(fullPath, w.errorDir, filename) return outcomeFailed } w.moveTo(fullPath, w.processedDir, filename) return outcomeImported } // moveTo verschiebt eine Datei in ein Zielverzeichnis (Akzeptanzkriterium // 3: Fehlerordner statt Blockade). Ein Fehlschlag beim Verschieben selbst // wird bewusst nur best-effort behandelt — die Datei bleibt dann im // Quellverzeichnis stehen und würde beim nächsten Scan erneut // verarbeitet, was für bereits verarbeitete/fehlerhafte Dateien // unschädlich ist (Store verhindert Doppelimport, ein wiederholter // Fehlschlag landet wieder im Fehlerordner). func (w *Watcher) moveTo(sourcePath, targetDir, filename string) { _ = os.Rename(sourcePath, filepath.Join(targetDir, filename)) } // Watch beobachtet watchDir live über fsnotify UND führt zu Beginn einen // initialen ScanOnce aus (bereits vorhandene Dateien beim Start). // Blockiert, bis ctx beendet wird. func (w *Watcher) Watch(ctx context.Context) error { if _, err := w.ScanOnce(ctx); err != nil { return err } fsWatcher, err := fsnotify.NewWatcher() if err != nil { return fmt.Errorf("hotfolder: fsnotify-watcher erstellen: %w", err) } defer func() { _ = fsWatcher.Close() }() if err := fsWatcher.Add(w.watchDir); err != nil { return fmt.Errorf("hotfolder: verzeichnis beobachten: %w", err) } for { select { case <-ctx.Done(): return nil case event, ok := <-fsWatcher.Events: if !ok { return nil } if event.Op&(fsnotify.Create|fsnotify.Write) == 0 { continue } if _, err := w.ScanOnce(ctx); err != nil { return err } case err, ok := <-fsWatcher.Errors: if !ok { return nil } return fmt.Errorf("hotfolder: fsnotify-fehler: %w", err) } } }