Merge remote-tracking branch 'origin/feature/int-08-watch-folder-import-consume-ordner' into feature/int-14-tenant-zuordnung-zugriffsschutz-hot-folder

# Conflicts:
#	dms/cmd/app/main.go

dms/cmd/app/main.go

@@ -9,10 +9,13 @@ import (
 	"log"
 	"net/http"
 	"os"
+	"strconv"
+	"strings"
 	"time"
 
 	"github.com/jackc/pgx/v5/pgxpool"
 
+	dmscrypto "gitea.perlbach24.de/scripte/nexarch/dms/internal/crypto"
 	"gitea.perlbach24.de/scripte/nexarch/dms/internal/errortracking"
 	"gitea.perlbach24.de/scripte/nexarch/dms/internal/jobqueue"
 	"gitea.perlbach24.de/scripte/nexarch/dms/internal/logging"
@@ -21,6 +24,8 @@ import (
 	"gitea.perlbach24.de/scripte/nexarch/dms/internal/retentiondestroy"
 	"gitea.perlbach24.de/scripte/nexarch/dms/internal/shared"
 	"gitea.perlbach24.de/scripte/nexarch/dms/internal/storage"
+	"gitea.perlbach24.de/scripte/nexarch/dms/internal/upload"
+	"gitea.perlbach24.de/scripte/nexarch/dms/internal/watchfolder"
 )
 
 func main() {
@@ -49,6 +54,14 @@ func main() {
 		log.Print("dms-app: NEXARCH_DMS_APP_TENANT_DSN nicht gesetzt, RET-05-Anbindung (DOC-16) deaktiviert")
 	}
 
+	watchCtx, stopWatch := context.WithCancel(context.Background())
+	defer stopWatch()
+	if dir := os.Getenv("NEXARCH_DMS_APP_WATCHFOLDER_DIR"); dir != "" {
+		setupWatchFolder(watchCtx, dir)
+	} else {
+		log.Print("dms-app: NEXARCH_DMS_APP_WATCHFOLDER_DIR nicht gesetzt, INT-08-Consume-Ordner deaktiviert")
+	}
+
 	handler := logging.Middleware(logger)(metricsReg.Middleware(recoverMiddleware(tracker, mux)))
 
 	log.Printf("dms-app hoert auf %s (version %s)", addr, shared.Version)
@@ -121,3 +134,79 @@ func setupRetention(mux *http.ServeMux, dsn string) {
 		log.Print("dms-app: bei archive RET-05 registriert")
 	}
 }
+
+// setupWatchFolder startet INT-08s Consume-Ordner-Beobachtung im
+// Hintergrund und mündet in dieselbe upload.Service-Pipeline wie DOC-01
+// (Hash/Verschlüsselung/Dokument+Revision) — es wird kein eigener Pfad
+// dupliziert. Läuft nur, wenn Tenant-DSN, Staging-Verzeichnis und ein
+// bereits angelegter System-Benutzer (created_by) konfiguriert sind.
+func setupWatchFolder(ctx context.Context, dir string) {
+	dsn := os.Getenv("NEXARCH_DMS_APP_TENANT_DSN")
+	systemUserID := os.Getenv("NEXARCH_DMS_APP_WATCHFOLDER_SYSTEM_USER_ID")
+	stagingDir := os.Getenv("NEXARCH_DMS_APP_WATCHFOLDER_STAGING_DIR")
+	if dsn == "" || systemUserID == "" || stagingDir == "" {
+		log.Print("dms-app: INT-08-Consume-Ordner uebersprungen (TENANT_DSN/WATCHFOLDER_SYSTEM_USER_ID/WATCHFOLDER_STAGING_DIR unvollstaendig)")
+		return
+	}
+
+	pool, err := pgxpool.New(ctx, dsn)
+	if err != nil {
+		log.Fatalf("dms-app: watchfolder-datenbankverbindung: %v", err)
+	}
+
+	storageDir := os.Getenv("NEXARCH_DMS_APP_STORAGE_DIR")
+	signingSecret := os.Getenv("NEXARCH_DMS_APP_STORAGE_SIGNING_SECRET")
+	publicBaseURL := os.Getenv("NEXARCH_DMS_APP_STORAGE_PUBLIC_BASE_URL")
+	driver := storage.NewLocalDriver(storageDir, []byte(signingSecret), publicBaseURL)
+	usageReporter := storage.NewHTTPUsageReporter(
+		os.Getenv("NEXARCH_DMS_APP_USAGE_ENDPOINT_URL"),
+		os.Getenv("NEXARCH_DMS_APP_USAGE_CLIENT_ID"),
+		os.Getenv("NEXARCH_DMS_APP_USAGE_CLIENT_SECRET"),
+		nil,
+	)
+	tenantSlug := os.Getenv("NEXARCH_DMS_APP_TENANT_SLUG")
+	storageSvc := storage.NewService(driver, usageReporter, tenantSlug)
+
+	kekProvider := dmscrypto.NewHTTPKEKProvider(
+		os.Getenv("NEXARCH_DMS_APP_KEK_ENDPOINT_URL"),
+		os.Getenv("NEXARCH_DMS_APP_KEK_CLIENT_ID"),
+		os.Getenv("NEXARCH_DMS_APP_KEK_CLIENT_SECRET"),
+		nil,
+	)
+	cryptoSvc := dmscrypto.NewService(kekProvider)
+
+	maxSize := int64(0)
+	if raw := os.Getenv("NEXARCH_DMS_APP_UPLOAD_MAX_SIZE_BYTES"); raw != "" {
+		if v, err := strconv.ParseInt(raw, 10, 64); err == nil {
+			maxSize = v
+		}
+	}
+	var allowedMimeTypes []string
+	if raw := os.Getenv("NEXARCH_DMS_APP_UPLOAD_ALLOWED_MIME_TYPES"); raw != "" {
+		allowedMimeTypes = strings.Split(raw, ",")
+	}
+	validator := upload.NewValidator(maxSize, allowedMimeTypes)
+
+	sessions := upload.NewSessionStore(pool)
+	staging := upload.NewStaging(stagingDir)
+	svc := upload.NewService(pool, sessions, staging, storageSvc, cryptoSvc, validator, tenantSlug)
+
+	pollInterval := 5 * time.Second
+	if raw := os.Getenv("NEXARCH_DMS_APP_WATCHFOLDER_POLL_INTERVAL_SECONDS"); raw != "" {
+		if v, err := strconv.Atoi(raw); err == nil && v > 0 {
+			pollInterval = time.Duration(v) * time.Second
+		}
+	}
+
+	watcher := watchfolder.New(svc, watchfolder.Config{
+		Dir:          dir,
+		PollInterval: pollInterval,
+		CreatedBy:    systemUserID,
+	})
+	go func() {
+		log.Printf("dms-app: INT-08-Consume-Ordner beobachtet %q", dir)
+		if err := watcher.Run(ctx); err != nil {
+			log.Printf("dms-app: watchfolder beendet: %v", err)
+		}
+	}()
+}

dms/docs/INT-08-PRUEFPROTOKOLL.md

@@ -0,0 +1,92 @@
+# INT-08 – Prüfprotokoll: Watch-Folder-Import (Consume-Ordner)
+
+Welle 4. Voraussetzung: DOC-01 (Status "Fertig").
+
+## Umsetzung
+
+`internal/watchfolder` ergänzt DOC-01s API-Upload um eine dateisystembasierte
+Einlieferung, ohne dessen Pipeline zu duplizieren:
+
+- `Watcher.Run`/`ScanOnce` pollt ein konfiguriertes Verzeichnis (Standard
+  5 s, `PollInterval`). Kein `fsnotify`-Einsatz — `grep -r fsnotify dms/go.sum`
+  ergab keinen Treffer, Polling ist hier die kleinste Lösung.
+- `isSettled` erkennt eine noch laufende Schreiboperation über zwei
+  `os.Stat`-Aufrufe im Abstand von `SettleDelay` (Standard 2 s): Größe und
+  Änderungszeit müssen zwischen beiden Aufrufen unverändert sein, sonst wird
+  die Datei im nächsten Durchlauf erneut geprüft.
+- `importFile` ruft ausschließlich `upload.Service.StartUpload` /
+  `UploadChunk` / `Complete` auf — denselben Pfad wie ein API-Upload (Hash,
+  Verschlüsselung, transaktionale Dokument-/Revisionsanlage). Kein
+  paralleler Code für diese Schritte.
+- Erfolg: Quelldatei wird entfernt (`os.Remove`).
+- Fehler (Validierung, defekte/nicht lesbare Datei): Datei wird
+  unverändert nach `<Dir>/error/` verschoben (`os.Rename`, bei
+  Namenskollision mit Zeitstempel-Präfix), zusätzlich eine
+  `<Datei>.error.txt`-Begleitdatei mit der Fehlermeldung für die manuelle
+  Prüfung. Ein Fehler bei einer Datei bricht `ScanOnce` nicht ab, die
+  Schleife über die restlichen Ordnereinträge läuft weiter.
+- `cmd/app/main.go`: optionale Hintergrund-Aktivierung über
+  `NEXARCH_DMS_APP_WATCHFOLDER_DIR` (+ `_SYSTEM_USER_ID`, `_STAGING_DIR`),
+  analog zum bestehenden `setupRetention`-Muster. Der Ordnerpfad ist
+  Konfiguration, keine Zugangsdaten liegen im Code.
+
+## Reichweite der Pipeline auf diesem Branch (wichtig für Prüfung 2)
+
+Geprüft mit `git log --all --oneline | grep -iE 'SRC-01|SRC-02|SRC-05|DOC-02'`
+und `git merge-base --is-ancestor <commit> HEAD`:
+
+| Baustein | Commit vorhanden im Repo | Erreichbar von diesem Branch (`--is-ancestor`) |
+|---|---|---|
+| DOC-02 Dublettenerkennung | ja (`182a8ac`, `304368e`) | **nein** |
+| SRC-01 Suchindex-Infrastruktur | ja | **nein** |
+| SRC-02 Indexierungs-Worker | ja (`c5bdde0`) | **nein** |
+| SRC-05 OCR/Facetten | ja (`db73aab`) | **nein** |
+
+Diese vier Kacheln sind eigene, noch nicht in die INT-08-Vorgängerkette
+gemergte Zweige. `internal/upload/service.go` (Stand dieses Branches, wie
+tatsächlich gelesen) enthält **keinen** Aufruf eines Dedup-Gates und
+**keine** OCR-/Index-Anbindung — nur Hash → Verschlüsselung → Storage →
+Dokument/Revision-Transaktion, optional Thumbnail- und Malware-Scan-
+Enqueue. Der Watch-Folder-Import mündet exakt in diese (auf diesem Branch
+tatsächlich existierende) Pipeline. Dedup-Gate/OCR/Index wurden daher
+**nicht** getestet, weil der Code dafür von hier aus nicht aufrufbar ist —
+keine Fabrikation einer Integration mit ungemergtem Code. Sobald DOC-02/
+SRC-01/SRC-02/SRC-05 in die Kette gemergt sind, durchläuft ein
+Watch-Folder-Import sie automatisch, da `watchfolder` ausschließlich über
+`upload.Service` läuft und diesen nicht dupliziert.
+
+## Prüfungen vor der Abnahme
+
+| # | Prüfung | Zielwert | Ergebnis |
+|---|---|---|---|
+| 1 | Testdatei im Ordner abgelegt, erscheint automatisch als Dokument | Datei wird ohne manuellen Trigger als `documents`-Zeile sichtbar | **bestanden** — `TestScanOnce_DetectsAndImportsDroppedFile`: Datei per `os.WriteFile` real in ein `t.TempDir()` geschrieben, `ScanOnce` (kein manueller Import-Aufruf) findet sie, Dokument mit passendem Titel existiert danach in der DB, Quelldatei ist aus dem Ordner verschwunden |
+| 2 | Pipeline-Schritte (Hash, Dedup-Gate, Verschlüsselung) im Test nachweislich durchlaufen | Hash und Verschlüsselung nachweisbar; Dedup-Gate nur, sofern erreichbar | **eingeschränkt bestanden** — `TestScanOnce_TraversesHashDedupGateAndEncryption`: `checksum_sha256` stimmt exakt mit unabhängig berechneter SHA-256 des Originalinhalts überein (Hash-Stufe), `wrapped_dek` ist befüllt (Verschlüsselungs-Stufe), `storage_key` ist gesetzt (Storage-Stufe). Dedup-Gate **nicht testbar** — DOC-02 auf diesem Branch nicht erreichbar (siehe Tabelle oben), offen bis Merge |
+| 3 | Absichtlich beschädigte Datei landet im Fehler-Unterordner, blockiert restlichen Ordner nicht | Fehlerdatei in `error/`, gültige Datei im selben Lauf trotzdem importiert | **bestanden** — `TestScanOnce_UnreadableFileMovedToErrorSubdirWithoutBlockingOthers`: Binärdatei mit unzulässigem/erkanntem MIME-Typ landet real in `error/` (Original-Bytes unverändert, plus `.error.txt`), eine im selben `ScanOnce`-Durchlauf abgelegte gültige Datei wird trotzdem erfolgreich zu einem Dokument |
+
+## Build/Test-Ergebnis (192.168.1.131)
+
+```
+cd /root/nexarch-code-int08/dms
+go build ./...   -> clean
+go vet ./...     -> clean
+go test ./internal/watchfolder/... -v -p 1   -> 3/3 PASS
+go test ./internal/upload/... -p 1           -> ok (unveraendert, DOC-01 weiterhin gruen)
+```
+
+`go test ./... -p 1` zeigt zusätzlich Fehlschläge in `metadatafield`,
+`migrate`, `thumbnail`, `trash`, `trashapi`, `versioning`, `versioningapi` —
+diese sind auf denselben, parallel von anderen Agenten genutzten
+`TEST_TENANT_DSN` zurückzuführen (Schema-/Datenreste aus gleichzeitigen
+Testläufen anderer Kacheln, z. B. `column "path" does not exist`,
+`relation "metadata_field_versions" already exists`) und bestehen
+unabhängig von dieser Änderung — außerhalb des Umfangs dieser Kachel, nicht
+durch `internal/watchfolder` oder die `cmd/app`-Änderung verursacht.
+
+## Gesamtergebnis
+
+**Bestanden für den auf diesem Branch tatsächlich erreichbaren
+Pipeline-Umfang.** Alle drei Akzeptanzkriterien und alle drei
+Pflichtprüfungen real durchgeführt. Dedup-Gate/OCR/Index sind wie im
+Ticket vorgesehen aus anderen, hier nicht gemergten Kacheln ausgeklammert
+und bleiben offen bis zu deren Merge — `watchfolder` selbst braucht dafür
+keine Änderung, da es ausschließlich `upload.Service` aufruft.

dms/internal/watchfolder/watcher.go

@@ -0,0 +1,200 @@
+// Package watchfolder implementiert INT-08: ein Dateisystem-Ordner, der
+// periodisch auf neue Dateien geprueft wird und diese ohne API-Aufruf ueber
+// denselben Pipeline-Weg wie internal/upload (DOC-01) einliefert — Hash,
+// Verschluesselung und Dokument/Revision-Anlage laufen unveraendert ueber
+// upload.Service, dieses Paket dupliziert die Pipeline nicht.
+package watchfolder
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"log"
+	"mime"
+	"net/http"
+	"os"
+	"path/filepath"
+	"strings"
+	"time"
+
+	"gitea.perlbach24.de/scripte/nexarch/dms/internal/upload"
+)
+
+const errorSubdir = "error"
+
+// Config steuert Ort, Takt und Zuordnung des Consume-Ordners. Kein Feld
+// enthaelt Zugangsdaten — der Ordnerpfad selbst ist Konfiguration, kein
+// Geheimnis.
+type Config struct {
+	Dir          string
+	PollInterval time.Duration
+	SettleDelay  time.Duration
+	FolderID     *string
+	CreatedBy    string
+}
+
+// Watcher pollt Config.Dir und uebergibt jede settled Datei an svc.
+type Watcher struct {
+	svc *upload.Service
+	cfg Config
+}
+
+func New(svc *upload.Service, cfg Config) *Watcher {
+	if cfg.PollInterval <= 0 {
+		cfg.PollInterval = 5 * time.Second
+	}
+	if cfg.SettleDelay <= 0 {
+		cfg.SettleDelay = 2 * time.Second
+	}
+	return &Watcher{svc: svc, cfg: cfg}
+}
+
+func (w *Watcher) errorDir() string {
+	return filepath.Join(w.cfg.Dir, errorSubdir)
+}
+
+// Run blockiert bis ctx endet und scannt Config.Dir im Takt von
+// PollInterval (Akzeptanzkriterium 1: automatische Erkennung ohne
+// manuellen Trigger).
+func (w *Watcher) Run(ctx context.Context) error {
+	if err := os.MkdirAll(w.cfg.Dir, 0o755); err != nil {
+		return fmt.Errorf("watchfolder: ordner anlegen: %w", err)
+	}
+	if err := os.MkdirAll(w.errorDir(), 0o755); err != nil {
+		return fmt.Errorf("watchfolder: fehler-unterordner anlegen: %w", err)
+	}
+
+	ticker := time.NewTicker(w.cfg.PollInterval)
+	defer ticker.Stop()
+
+	w.ScanOnce(ctx)
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case <-ticker.C:
+			w.ScanOnce(ctx)
+		}
+	}
+}
+
+// ScanOnce prueft Config.Dir einmalig auf importbereite Dateien — ein
+// Fehler bei einer Datei stoppt die uebrigen nicht (Akzeptanzkriterium 3).
+func (w *Watcher) ScanOnce(ctx context.Context) {
+	entries, err := os.ReadDir(w.cfg.Dir)
+	if err != nil {
+		log.Printf("watchfolder: ordner lesen: %v", err)
+		return
+	}
+
+	for _, entry := range entries {
+		if entry.IsDir() {
+			continue
+		}
+		path := filepath.Join(w.cfg.Dir, entry.Name())
+
+		if !w.isSettled(path) {
+			continue
+		}
+
+		if err := w.importFile(ctx, path); err != nil {
+			log.Printf("watchfolder: import von %q fehlgeschlagen, verschiebe nach %q: %v", path, errorSubdir, err)
+			if moveErr := w.moveToError(path, err); moveErr != nil {
+				log.Printf("watchfolder: verschieben nach fehler-unterordner fehlgeschlagen fuer %q: %v", path, moveErr)
+			}
+		}
+	}
+}
+
+// isSettled erkennt eine noch laufende Schreiboperation durch zwei
+// stat()-Aufrufe im Abstand von SettleDelay — eine unveraenderte Groesse
+// gilt als abgeschlossen geschrieben.
+func (w *Watcher) isSettled(path string) bool {
+	first, err := os.Stat(path)
+	if err != nil {
+		return false
+	}
+	time.Sleep(w.cfg.SettleDelay)
+	second, err := os.Stat(path)
+	if err != nil {
+		return false
+	}
+	return first.Size() == second.Size() && first.ModTime().Equal(second.ModTime())
+}
+
+// importFile fuehrt die Datei durch dieselbe StartUpload/UploadChunk/
+// Complete-Sequenz wie ein API-Upload (Akzeptanzkriterium 2) und entfernt
+// die Quelldatei danach.
+func (w *Watcher) importFile(ctx context.Context, path string) error {
+	f, err := os.Open(path)
+	if err != nil {
+		return fmt.Errorf("datei oeffnen: %w", err)
+	}
+	defer func() { _ = f.Close() }()
+
+	info, err := f.Stat()
+	if err != nil {
+		return fmt.Errorf("dateigroesse ermitteln: %w", err)
+	}
+
+	mimeType, err := detectMimeType(f, filepath.Base(path))
+	if err != nil {
+		return fmt.Errorf("mime-typ erkennen: %w", err)
+	}
+
+	sessionID, err := w.svc.StartUpload(ctx, filepath.Base(path), mimeType, info.Size(), w.cfg.FolderID, w.cfg.CreatedBy)
+	if err != nil {
+		return fmt.Errorf("upload starten: %w", err)
+	}
+	if _, err := w.svc.UploadChunk(ctx, sessionID, 0, f); err != nil {
+		return fmt.Errorf("inhalt uebertragen: %w", err)
+	}
+	if _, _, err := w.svc.Complete(ctx, sessionID); err != nil {
+		return fmt.Errorf("upload abschliessen: %w", err)
+	}
+
+	if err := os.Remove(path); err != nil {
+		return fmt.Errorf("quelldatei nach erfolgreichem import entfernen: %w", err)
+	}
+	return nil
+}
+
+// detectMimeType nutzt zuerst die Dateiendung, da sie fuer Bueroformate
+// (docx, xlsx, ...) zuverlaessiger ist als eine Signaturpruefung der ersten
+// Bytes, und faellt sonst auf http.DetectContentType zurueck.
+func detectMimeType(f *os.File, filename string) (string, error) {
+	if byExt := mime.TypeByExtension(filepath.Ext(filename)); byExt != "" {
+		return strings.SplitN(byExt, ";", 2)[0], nil
+	}
+
+	buf := make([]byte, 512)
+	n, err := f.Read(buf)
+	if err != nil && err != io.EOF {
+		return "", err
+	}
+	if _, err := f.Seek(0, io.SeekStart); err != nil {
+		return "", err
+	}
+	return http.DetectContentType(buf[:n]), nil
+}
+
+// moveToError verschiebt eine fehlerhafte/nicht lesbare Datei unveraendert
+// in den Fehler-Unterordner (Akzeptanzkriterium 3) und legt eine
+// Begleitdatei mit der Fehlermeldung fuer die manuelle Pruefung ab.
+func (w *Watcher) moveToError(path string, importErr error) error {
+	if err := os.MkdirAll(w.errorDir(), 0o755); err != nil {
+		return err
+	}
+
+	name := filepath.Base(path)
+	dest := filepath.Join(w.errorDir(), name)
+	if _, err := os.Stat(dest); err == nil {
+		dest = filepath.Join(w.errorDir(), fmt.Sprintf("%d-%s", time.Now().UnixNano(), name))
+	}
+
+	if err := os.Rename(path, dest); err != nil {
+		return err
+	}
+	_ = os.WriteFile(dest+".error.txt", []byte(importErr.Error()+"\n"), 0o644)
+	return nil
+}

dms/internal/watchfolder/watcher_test.go

@@ -0,0 +1,241 @@
+package watchfolder
+
+import (
+	"bytes"
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"fmt"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	dmscrypto "gitea.perlbach24.de/scripte/nexarch/dms/internal/crypto"
+	"gitea.perlbach24.de/scripte/nexarch/dms/internal/storage"
+	"gitea.perlbach24.de/scripte/nexarch/dms/internal/upload"
+)
+
+type fakeKEKProvider struct{ kek []byte }
+
+func (f *fakeKEKProvider) TenantKEK(ctx context.Context, tenantSlug string) ([]byte, error) {
+	return f.kek, nil
+}
+
+type noopUsageReporter struct{}
+
+func (noopUsageReporter) Report(ctx context.Context, tenantSlug, metric string, delta int64) error {
+	return nil
+}
+
+// setupTest baut dieselbe Tabellen-Fixture wie internal/upload's eigene
+// Tests auf, da watchfolder ausschliesslich upload.Service aufruft und
+// keine eigene Datenzugriffsschicht hat.
+func setupTest(t *testing.T) (svc *upload.Service, pool *pgxpool.Pool, userID string) {
+	t.Helper()
+	dsn := os.Getenv("TEST_TENANT_DSN")
+	if dsn == "" {
+		t.Skip("TEST_TENANT_DSN nicht gesetzt, Integrationstest uebersprungen")
+	}
+	ctx := context.Background()
+
+	pool, err := pgxpool.New(ctx, dsn)
+	if err != nil {
+		t.Fatalf("pool: %v", err)
+	}
+	t.Cleanup(func() { pool.Close() })
+
+	if _, err := pool.Exec(ctx, `
+		CREATE EXTENSION IF NOT EXISTS pgcrypto;
+		CREATE TABLE IF NOT EXISTS users (
+			id UUID PRIMARY KEY DEFAULT gen_random_uuid(), email TEXT NOT NULL UNIQUE, name TEXT NOT NULL,
+			status TEXT NOT NULL DEFAULT 'active', created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+		);
+		CREATE TABLE IF NOT EXISTS document_types (
+			id UUID PRIMARY KEY DEFAULT gen_random_uuid(), name TEXT NOT NULL UNIQUE, retention_class TEXT,
+			created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+		);
+		CREATE TABLE IF NOT EXISTS folders (
+			id UUID PRIMARY KEY DEFAULT gen_random_uuid(), parent_folder_id UUID REFERENCES folders(id) ON DELETE CASCADE,
+			name TEXT NOT NULL, created_by UUID NOT NULL REFERENCES users(id),
+			default_document_type_id UUID REFERENCES document_types(id) ON DELETE SET NULL,
+			created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+		);
+		CREATE TABLE IF NOT EXISTS documents (
+			id UUID PRIMARY KEY DEFAULT gen_random_uuid(), folder_id UUID REFERENCES folders(id) ON DELETE SET NULL,
+			title TEXT NOT NULL, current_revision_id UUID, created_by UUID NOT NULL REFERENCES users(id),
+			document_type_id UUID REFERENCES document_types(id) ON DELETE SET NULL,
+			created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), deleted_at TIMESTAMPTZ
+		);
+		CREATE TABLE IF NOT EXISTS file_revisions (
+			id UUID PRIMARY KEY DEFAULT gen_random_uuid(), document_id UUID NOT NULL REFERENCES documents(id) ON DELETE CASCADE,
+			revision_number INT NOT NULL, storage_key TEXT NOT NULL, checksum_sha256 TEXT NOT NULL,
+			size_bytes BIGINT NOT NULL, mime_type TEXT NOT NULL, created_by UUID NOT NULL REFERENCES users(id),
+			created_at TIMESTAMPTZ NOT NULL DEFAULT now(), wrapped_dek BYTEA,
+			UNIQUE (document_id, revision_number)
+		);
+		CREATE TABLE IF NOT EXISTS upload_sessions (
+			id UUID PRIMARY KEY DEFAULT gen_random_uuid(), filename TEXT NOT NULL, mime_type TEXT NOT NULL,
+			total_size BIGINT NOT NULL CHECK (total_size > 0), bytes_received BIGINT NOT NULL DEFAULT 0 CHECK (bytes_received >= 0),
+			folder_id UUID REFERENCES folders(id) ON DELETE SET NULL, created_by UUID NOT NULL REFERENCES users(id),
+			status TEXT NOT NULL DEFAULT 'uploading' CHECK (status IN ('uploading','completed','aborted')),
+			document_id UUID REFERENCES documents(id),
+			scan_status TEXT NOT NULL DEFAULT 'pending' CHECK (scan_status IN ('pending','clean','rejected')),
+			created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+		);
+	`); err != nil {
+		t.Fatalf("schema: %v", err)
+	}
+	t.Cleanup(func() {
+		ctx := context.Background()
+		_, _ = pool.Exec(ctx, `TRUNCATE upload_sessions, file_revisions, documents, folders, document_types, users CASCADE`)
+	})
+
+	if err := pool.QueryRow(ctx, `
+		INSERT INTO users (email, name) VALUES ($1, 'Watchfolder-Test-Benutzer') RETURNING id
+	`, fmt.Sprintf("watchfolder-test-%d@example.test", time.Now().UnixNano())).Scan(&userID); err != nil {
+		t.Fatalf("testbenutzer anlegen: %v", err)
+	}
+
+	sessions := upload.NewSessionStore(pool)
+	staging := upload.NewStaging(t.TempDir())
+	storageSvc := storage.NewService(storage.NewLocalDriver(t.TempDir(), []byte("secret"), "https://files.example.test"), noopUsageReporter{}, "acme")
+	cryptoSvc := dmscrypto.NewService(&fakeKEKProvider{kek: bytes.Repeat([]byte{0x11}, dmscrypto.KEKSize)})
+	validator := upload.NewValidator(10*1024*1024, nil)
+
+	svc = upload.NewService(pool, sessions, staging, storageSvc, cryptoSvc, validator, "acme")
+	return svc, pool, userID
+}
+
+// TestScanOnce_DetectsAndImportsDroppedFile ist Pflichtpruefung 1 und
+// Akzeptanzkriterium 1: eine im Ordner abgelegte Datei wird ohne manuellen
+// Trigger als Dokument angelegt.
+func TestScanOnce_DetectsAndImportsDroppedFile(t *testing.T) {
+	svc, pool, userID := setupTest(t)
+	ctx := context.Background()
+	dir := t.TempDir()
+
+	content := []byte("INT-08 Testinhalt fuer Consume-Ordner")
+	if err := os.WriteFile(filepath.Join(dir, "beleg.txt"), content, 0o644); err != nil {
+		t.Fatalf("testdatei schreiben: %v", err)
+	}
+
+	w := New(svc, Config{Dir: dir, SettleDelay: 50 * time.Millisecond, CreatedBy: userID})
+	if err := os.MkdirAll(w.errorDir(), 0o755); err != nil {
+		t.Fatalf("fehlerordner anlegen: %v", err)
+	}
+	w.ScanOnce(ctx)
+
+	var count int
+	if err := pool.QueryRow(ctx, `SELECT count(*) FROM documents WHERE title = 'beleg.txt'`).Scan(&count); err != nil {
+		t.Fatalf("dokument zaehlen: %v", err)
+	}
+	if count != 1 {
+		t.Fatalf("dokumentanzahl = %d, want 1", count)
+	}
+
+	if _, err := os.Stat(filepath.Join(dir, "beleg.txt")); !os.IsNotExist(err) {
+		t.Fatalf("quelldatei haette nach erfolgreichem import entfernt sein sollen, stat-fehler: %v", err)
+	}
+}
+
+// TestScanOnce_TraversesHashDedupGateAndEncryption ist Pflichtpruefung 2 und
+// Akzeptanzkriterium 2 (fuer den in diesem Repo tatsaechlich erreichbaren
+// Teil der Pipeline — siehe INT-08-PRUEFPROTOKOLL.md fuer die Einordnung
+// von Dedup-Gate/OCR/Index, die aus separaten, hier nicht gemergten
+// Kacheln stammen).
+func TestScanOnce_TraversesHashDedupGateAndEncryption(t *testing.T) {
+	svc, pool, userID := setupTest(t)
+	ctx := context.Background()
+	dir := t.TempDir()
+
+	content := []byte("INT-08 Pipeline-Nachweis: Hash und Verschluesselung")
+	sum := sha256.Sum256(content)
+	wantChecksum := hex.EncodeToString(sum[:])
+	if err := os.WriteFile(filepath.Join(dir, "nachweis.txt"), content, 0o644); err != nil {
+		t.Fatalf("testdatei schreiben: %v", err)
+	}
+
+	w := New(svc, Config{Dir: dir, SettleDelay: 50 * time.Millisecond, CreatedBy: userID})
+	if err := os.MkdirAll(w.errorDir(), 0o755); err != nil {
+		t.Fatalf("fehlerordner anlegen: %v", err)
+	}
+	w.ScanOnce(ctx)
+
+	var checksum string
+	var wrappedDEK []byte
+	var storageKey string
+	if err := pool.QueryRow(ctx, `
+		SELECT checksum_sha256, wrapped_dek, storage_key FROM file_revisions
+		JOIN documents ON documents.current_revision_id = file_revisions.id
+		WHERE documents.title = 'nachweis.txt'
+	`).Scan(&checksum, &wrappedDEK, &storageKey); err != nil {
+		t.Fatalf("revision lesen: %v", err)
+	}
+	if checksum != wantChecksum {
+		t.Fatalf("checksum_sha256 = %q, want %q (Hash-Stufe nicht durchlaufen)", checksum, wantChecksum)
+	}
+	if len(wrappedDEK) == 0 {
+		t.Fatal("erwartet nicht-leeren wrapped_dek (Verschluesselungs-Stufe nicht durchlaufen)")
+	}
+	if storageKey == "" {
+		t.Fatal("erwartet gesetzten storage_key (Storage-Stufe nicht durchlaufen)")
+	}
+}
+
+// TestScanOnce_UnreadableFileMovedToErrorSubdirWithoutBlockingOthers ist
+// Pflichtpruefung 3 und Akzeptanzkriterium 3: eine absichtlich verworfene
+// Datei (hier: unzulaessiger MIME-Typ als Stellvertreter fuer "fehlerhaft/
+// nicht lesbar", da die Validierung an derselben Stelle greift wie bei
+// einer defekten Datei) landet im Fehler-Unterordner, eine parallel
+// abgelegte gueltige Datei wird trotzdem importiert.
+func TestScanOnce_UnreadableFileMovedToErrorSubdirWithoutBlockingOthers(t *testing.T) {
+	_, pool, userID := setupTest(t)
+	ctx := context.Background()
+	dir := t.TempDir()
+
+	// eigener Validator mit Positivliste, damit eine Datei zuverlaessig
+	// als "fehlerhaft" abgelehnt wird, unabhaengig vom erkannten MIME-Typ.
+	restricted := upload.NewValidator(10*1024*1024, []string{"text/plain", "text/plain; charset=utf-8"})
+	svcRestricted := upload.NewService(pool, upload.NewSessionStore(pool), upload.NewStaging(t.TempDir()),
+		newStorageSvc(t), newCryptoSvc(), restricted, "acme")
+
+	if err := os.WriteFile(filepath.Join(dir, "defekt.bin"), []byte{0x00, 0x01, 0x02, 0x03}, 0o644); err != nil {
+		t.Fatalf("defekte testdatei schreiben: %v", err)
+	}
+	if err := os.WriteFile(filepath.Join(dir, "gueltig.txt"), []byte("gueltiger Inhalt"), 0o644); err != nil {
+		t.Fatalf("gueltige testdatei schreiben: %v", err)
+	}
+
+	w := New(svcRestricted, Config{Dir: dir, SettleDelay: 50 * time.Millisecond, CreatedBy: userID})
+	if err := os.MkdirAll(w.errorDir(), 0o755); err != nil {
+		t.Fatalf("fehlerordner anlegen: %v", err)
+	}
+	w.ScanOnce(ctx)
+
+	if _, err := os.Stat(filepath.Join(w.errorDir(), "defekt.bin")); err != nil {
+		t.Fatalf("erwartet defekt.bin im fehler-unterordner: %v", err)
+	}
+	if _, err := os.Stat(filepath.Join(dir, "defekt.bin")); !os.IsNotExist(err) {
+		t.Fatal("defekt.bin haette aus dem hauptordner verschoben sein sollen")
+	}
+
+	var count int
+	if err := pool.QueryRow(ctx, `SELECT count(*) FROM documents WHERE title = 'gueltig.txt'`).Scan(&count); err != nil {
+		t.Fatalf("dokument zaehlen: %v", err)
+	}
+	if count != 1 {
+		t.Fatalf("dokumentanzahl fuer gueltig.txt = %d, want 1 (fehlerhafte Datei haette restlichen Ordner nicht blockieren duerfen)", count)
+	}
+}
+
+func newStorageSvc(t *testing.T) *storage.Service {
+	t.Helper()
+	return storage.NewService(storage.NewLocalDriver(t.TempDir(), []byte("secret"), "https://files.example.test"), noopUsageReporter{}, "acme")
+}
+
+func newCryptoSvc() *dmscrypto.Service {
+	return dmscrypto.NewService(&fakeKEKProvider{kek: bytes.Repeat([]byte{0x11}, dmscrypto.KEKSize)})
+}