diff --git a/dms/cmd/app/main.go b/dms/cmd/app/main.go index bd5da0ed..6cb63189 100644 --- a/dms/cmd/app/main.go +++ b/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) + } + }() +} diff --git a/dms/docs/INT-08-PRUEFPROTOKOLL.md b/dms/docs/INT-08-PRUEFPROTOKOLL.md new file mode 100644 index 00000000..9f734ed1 --- /dev/null +++ b/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 `/error/` verschoben (`os.Rename`, bei + Namenskollision mit Zeitstempel-Präfix), zusätzlich eine + `.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 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. diff --git a/dms/internal/watchfolder/watcher.go b/dms/internal/watchfolder/watcher.go new file mode 100644 index 00000000..a19a5281 --- /dev/null +++ b/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 +} diff --git a/dms/internal/watchfolder/watcher_test.go b/dms/internal/watchfolder/watcher_test.go new file mode 100644 index 00000000..8cdaa7ab --- /dev/null +++ b/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)}) +}