nexarch/dms/internal/export/export_test.go

package export

import (
	"archive/zip"
	"bytes"
	"context"
	"encoding/csv"
	"fmt"
	"io"
	"os"
	"runtime"
	"strings"
	"testing"
	"time"

	"github.com/jackc/pgx/v5/pgxpool"

	dmscrypto "gitea.perlbach24.de/scripte/nexarch/dms/internal/crypto"
	"gitea.perlbach24.de/scripte/nexarch/dms/internal/jobqueue"
	"gitea.perlbach24.de/scripte/nexarch/dms/internal/storage"
)

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
}

// ownerChecker erlaubt Zugriff nur auf Dokumente, deren created_by dem
// Aufrufer entspricht — reale, DB-gestuetzte Pruefung (Akzeptanzkriterium 3).
type ownerChecker struct{ pool *pgxpool.Pool }

func (c *ownerChecker) CanAccessDocument(ctx context.Context, callerUserID, documentID string) (bool, error) {
	var createdBy string
	if err := c.pool.QueryRow(ctx, `SELECT created_by FROM documents WHERE id = $1`, documentID).Scan(&createdBy); err != nil {
		return false, err
	}
	return callerUserID == createdBy, nil
}

type testEnv struct {
	pool       *pgxpool.Pool
	storageSvc *storage.Service
	storageDir string
	cryptoSvc  *dmscrypto.Service
	queue      *jobqueue.Queue
	worker     *jobqueue.Worker
}

func setupTest(t *testing.T) (*testEnv, 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 documents (
			id UUID PRIMARY KEY DEFAULT gen_random_uuid(), folder_id UUID,
			title TEXT NOT NULL, current_revision_id UUID, created_by UUID NOT NULL REFERENCES users(id),
			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 document_exports (
			id UUID PRIMARY KEY DEFAULT gen_random_uuid(), requested_by UUID NOT NULL REFERENCES users(id),
			document_ids UUID[] NOT NULL, status TEXT NOT NULL CHECK (status IN ('pending','processing','ready','failed')),
			storage_key TEXT, skipped_ids UUID[] NOT NULL DEFAULT '{}', error TEXT,
			created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
		);
		CREATE TABLE IF NOT EXISTS processing_jobs (
			id UUID PRIMARY KEY DEFAULT gen_random_uuid(), job_type TEXT NOT NULL, payload JSONB NOT NULL DEFAULT '{}'::jsonb,
			idempotency_key TEXT UNIQUE, status TEXT NOT NULL DEFAULT 'pending'
				CHECK (status IN ('pending', 'processing', 'succeeded', 'failed', 'dead_letter')),
			attempts INT NOT NULL DEFAULT 0, max_attempts INT NOT NULL DEFAULT 5,
			available_at TIMESTAMPTZ NOT NULL DEFAULT now(), locked_at TIMESTAMPTZ, locked_by TEXT, last_error TEXT,
			created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
		);
	`); err != nil {
		t.Fatalf("schema: %v", err)
	}
	t.Cleanup(func() {
		_, _ = pool.Exec(context.Background(), `TRUNCATE processing_jobs, document_exports, file_revisions, documents, users CASCADE`)
	})

	storageDir := t.TempDir()
	storageSvc := storage.NewService(storage.NewLocalDriver(storageDir, []byte("secret"), "https://files.example.test"), noopUsageReporter{}, "acme")
	cryptoSvc := dmscrypto.NewService(&fakeKEKProvider{kek: bytes.Repeat([]byte{0x22}, dmscrypto.KEKSize)})
	queue := jobqueue.NewQueue(pool, 5*time.Minute)
	proc := &Processor{Pool: pool, Storage: storageSvc, Crypto: cryptoSvc, TenantSlug: "acme", Permissions: &ownerChecker{pool: pool}}
	worker := jobqueue.NewWorker(queue, "test-export-worker", []string{JobType}, time.Millisecond, proc.Handler)

	var userID string
	if err := pool.QueryRow(ctx, `INSERT INTO users (email, name) VALUES ($1, 'A') RETURNING id`, fmt.Sprintf("export-%d@example.test", time.Now().UnixNano())).Scan(&userID); err != nil {
		t.Fatalf("user: %v", err)
	}
	return &testEnv{pool: pool, storageSvc: storageSvc, storageDir: storageDir, cryptoSvc: cryptoSvc, queue: queue, worker: worker}, userID
}

func createDocument(t *testing.T, env *testEnv, userID, title, content string) string {
	t.Helper()
	ctx := context.Background()
	var documentID string
	if err := env.pool.QueryRow(ctx, `INSERT INTO documents (title, created_by) VALUES ($1, $2) RETURNING id`, title, userID).Scan(&documentID); err != nil {
		t.Fatalf("dokument anlegen: %v", err)
	}
	envSeal, err := env.cryptoSvc.Seal(ctx, "acme", strings.NewReader(content))
	if err != nil {
		t.Fatalf("seal: %v", err)
	}
	ciphertext, err := io.ReadAll(envSeal.Ciphertext)
	if err != nil {
		t.Fatalf("ciphertext lesen: %v", err)
	}
	storageKey := storage.ObjectKey(documentID, "1")
	if _, err := env.storageSvc.Put(ctx, storageKey, bytes.NewReader(ciphertext), int64(len(ciphertext)), "application/octet-stream"); err != nil {
		t.Fatalf("objekt ablegen: %v", err)
	}
	var revisionID string
	if err := env.pool.QueryRow(ctx, `
		INSERT INTO file_revisions (document_id, revision_number, storage_key, checksum_sha256, size_bytes, mime_type, created_by, wrapped_dek)
		VALUES ($1, 1, $2, 'x', $3, 'text/plain', $4, $5) RETURNING id
	`, documentID, storageKey, len(ciphertext), userID, envSeal.WrappedDEK).Scan(&revisionID); err != nil {
		t.Fatalf("revision anlegen: %v", err)
	}
	if _, err := env.pool.Exec(ctx, `UPDATE documents SET current_revision_id = $2 WHERE id = $1`, documentID, revisionID); err != nil {
		t.Fatalf("aktuelle revision setzen: %v", err)
	}
	return documentID
}

func openProducedZip(t *testing.T, env *testEnv, storageKey string) *zip.Reader {
	t.Helper()
	data, err := os.ReadFile(env.storageDir + "/" + storageKey)
	if err != nil {
		t.Fatalf("zip-datei lesen: %v", err)
	}
	zr, err := zip.NewReader(bytes.NewReader(data), int64(len(data)))
	if err != nil {
		t.Fatalf("zip nicht lesbar (Pflichtpruefung 2 - Entpacker-Kompatibilitaet): %v", err)
	}
	return zr
}

// Akzeptanzkriterium 1: Export läuft als Job (über die reguläre
// jobqueue.Worker-Pipeline, nicht als Direktaufruf) und blockiert die
// Oberfläche nicht — der Aufrufer bekommt sofort eine Export-ID zurück,
// die eigentliche Verarbeitung passiert getrennt über RunOnce/Run.
func TestExport_RunsAsAsyncJobAndProducesReadableZip(t *testing.T) {
	env, userID := setupTest(t)
	ctx := context.Background()
	docA := createDocument(t, env, userID, "a.txt", "Inhalt A")
	docB := createDocument(t, env, userID, "b.txt", "Inhalt B")

	exportID, err := Enqueue(ctx, env.pool, env.queue, userID, []string{docA, docB}, Limits{})
	if err != nil {
		t.Fatalf("Enqueue: %v", err)
	}

	exp, err := Get(ctx, env.pool, exportID)
	if err != nil {
		t.Fatalf("Get (vor Verarbeitung): %v", err)
	}
	if exp.Status != StatusPending {
		t.Fatalf("status vor Verarbeitung = %q, want pending (Job noch nicht gelaufen)", exp.Status)
	}

	env.worker.RunOnce(ctx)

	exp, err = Get(ctx, env.pool, exportID)
	if err != nil {
		t.Fatalf("Get (nach Verarbeitung): %v", err)
	}
	if exp.Status != StatusReady {
		t.Fatalf("status = %q, want ready, fehler=%v", exp.Status, exp.Error)
	}

	// Pflichtpruefung 2: Archiv in gaengigen Entpackern lesbar.
	zr := openProducedZip(t, env, *exp.StorageKey)
	names := map[string]bool{}
	for _, f := range zr.File {
		names[f.Name] = true
	}
	if !names["a.txt"] || !names["b.txt"] {
		t.Fatalf("erwartet a.txt und b.txt im archiv, habe: %v", names)
	}
	// Akzeptanzkriterium 2: Metadaten-Beilage vorhanden.
	if !names["manifest.csv"] {
		t.Fatalf("erwartet manifest.csv im archiv, habe: %v", names)
	}
}

// Akzeptanzkriterium 2: Archiv enthält die Dokumente UND eine korrekte
// Metadaten-Beilage (Prüfsumme/Größe je Dokument).
func TestExport_ManifestListsAllDocumentsWithChecksum(t *testing.T) {
	env, userID := setupTest(t)
	ctx := context.Background()
	docA := createDocument(t, env, userID, "vertrag.txt", "Vertragsinhalt")

	exportID, err := Enqueue(ctx, env.pool, env.queue, userID, []string{docA}, Limits{})
	if err != nil {
		t.Fatalf("Enqueue: %v", err)
	}
	env.worker.RunOnce(ctx)
	exp, err := Get(ctx, env.pool, exportID)
	if err != nil || exp.Status != StatusReady {
		t.Fatalf("export nicht bereit: %+v, err=%v", exp, err)
	}

	zr := openProducedZip(t, env, *exp.StorageKey)
	var manifestFile *zip.File
	for _, f := range zr.File {
		if f.Name == "manifest.csv" {
			manifestFile = f
		}
	}
	if manifestFile == nil {
		t.Fatal("manifest.csv nicht gefunden")
	}
	rc, err := manifestFile.Open()
	if err != nil {
		t.Fatalf("manifest oeffnen: %v", err)
	}
	defer func() { _ = rc.Close() }()
	rows, err := csv.NewReader(rc).ReadAll()
	if err != nil {
		t.Fatalf("manifest parsen: %v", err)
	}
	if len(rows) != 2 || rows[1][0] != docA || rows[1][1] != "vertrag.txt" {
		t.Fatalf("manifest-inhalt falsch: %v", rows)
	}
}

// Akzeptanzkriterium 3 / Pflichtpruefung 3: nur berechtigte Dokumente
// landen im Archiv, ein Negativtest belegt den Ausschluss.
func TestExport_SkipsUnauthorizedDocuments(t *testing.T) {
	env, userID := setupTest(t)
	ctx := context.Background()
	ownDoc := createDocument(t, env, userID, "eigenes.txt", "mein inhalt")

	var otherUserID string
	if err := env.pool.QueryRow(ctx, `INSERT INTO users (email, name) VALUES ($1, 'B') RETURNING id`, fmt.Sprintf("other-%d@example.test", time.Now().UnixNano())).Scan(&otherUserID); err != nil {
		t.Fatalf("anderen benutzer anlegen: %v", err)
	}
	foreignDoc := createDocument(t, env, otherUserID, "fremdes.txt", "fremder inhalt")

	exportID, err := Enqueue(ctx, env.pool, env.queue, userID, []string{ownDoc, foreignDoc}, Limits{})
	if err != nil {
		t.Fatalf("Enqueue: %v", err)
	}
	env.worker.RunOnce(ctx)
	exp, err := Get(ctx, env.pool, exportID)
	if err != nil || exp.Status != StatusReady {
		t.Fatalf("export nicht bereit: %+v, err=%v", exp, err)
	}
	if len(exp.SkippedIDs) != 1 || exp.SkippedIDs[0] != foreignDoc {
		t.Fatalf("erwartet fremdes.txt uebersprungen, habe skipped=%v", exp.SkippedIDs)
	}

	zr := openProducedZip(t, env, *exp.StorageKey)
	names := map[string]bool{}
	for _, f := range zr.File {
		names[f.Name] = true
	}
	if !names["eigenes.txt"] {
		t.Fatalf("erwartet eigenes.txt im archiv: %v", names)
	}
	if names["fremdes.txt"] {
		t.Fatalf("fremdes.txt haette NICHT im archiv landen duerfen: %v", names)
	}
}

// Pflichtpruefung 1: Export von 1000 Dokumenten ohne Speicherueberlauf —
// echte Verarbeitung von 1000 Dokumenten, mit grober Speicher-Stichprobe
// (Heap waechst nicht proportional zur Gesamtgroesse aller Dokumente, weil
// gestreamt statt vollstaendig im Speicher aufgebaut wird).
func TestExport_HandlesOneThousandDocumentsWithoutMemoryBlowup(t *testing.T) {
	if testing.Short() {
		t.Skip("kurzer Testlauf angefordert (-short)")
	}
	env, userID := setupTest(t)
	ctx := context.Background()

	const n = 1000
	const perDocSize = 20_000 // 20 KB je Dokument = 20 MB Gesamtinhalt
	content := strings.Repeat("A", perDocSize)
	documentIDs := make([]string, n)
	for i := 0; i < n; i++ {
		documentIDs[i] = createDocument(t, env, userID, fmt.Sprintf("dok-%d.txt", i), content)
	}

	exportID, err := Enqueue(ctx, env.pool, env.queue, userID, documentIDs, Limits{})
	if err != nil {
		t.Fatalf("Enqueue: %v", err)
	}

	var before, after runtime.MemStats
	runtime.GC()
	runtime.ReadMemStats(&before)

	start := time.Now()
	env.worker.RunOnce(ctx)
	elapsed := time.Since(start)

	runtime.GC()
	runtime.ReadMemStats(&after)

	exp, err := Get(ctx, env.pool, exportID)
	if err != nil || exp.Status != StatusReady {
		t.Fatalf("export nicht bereit nach %d dokumenten: %+v, err=%v", n, exp, err)
	}
	t.Logf("export von %d dokumenten (%d bytes gesamt): %s, heap-delta ca. %d KB", n, n*perDocSize, elapsed, (after.HeapAlloc-before.HeapAlloc)/1024)

	totalContentMB := int64(n*perDocSize) / (1024 * 1024)
	heapDeltaMB := int64(after.HeapAlloc-before.HeapAlloc) / (1024 * 1024)
	if heapDeltaMB > totalContentMB/2 {
		t.Fatalf("heap-zuwachs (%d MB) liegt in der groessenordnung des gesamtinhalts (%d MB) — deutet auf vollstaendige pufferung statt streaming hin", heapDeltaMB, totalContentMB)
	}

	zr := openProducedZip(t, env, *exp.StorageKey)
	if len(zr.File) != n+1 { // +1 fuer manifest.csv
		t.Fatalf("erwartet %d eintraege (+manifest), habe %d", n+1, len(zr.File))
	}
}