Aether/backend/internal/documents/store_test.go

package documents

import (
	"context"
	"errors"
	"fmt"
	"os"
	"testing"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5/pgxpool"
)

// Tests der Warteschlange für die Texterkennung.
//
// **Warum ausgerechnet hier.** Ein Fehler in diesem Paket ist *still*: Die
// Volltextsuche findet dann einfach nichts – und das sieht genauso aus wie
// „steht nicht drin". Niemand bekommt eine Fehlermeldung; man merkt es erst,
// wenn man eine Rechnung sucht, von der man weiß, dass man sie hat.
//
// Geprüft werden deshalb nicht die Abfragen, sondern die drei Zustände, in
// denen die Schlange hängenbleiben kann, ohne dass es auffällt:
//
//   1. Ein Auftrag, der nie aufgibt (Endlosschleife auf einer kaputten Datei).
//   2. Ein Auftrag, der nach einem harten Abbruch für immer auf 'running'
//      steht und nie wieder angefasst wird.
//   3. Ein Text nach alter Regel, der nie erneuert wird.
//
// Läuft nur mit TEST_DSN gegen eine echte Datenbank; ohne wird übersprungen,
// damit `go test ./...` überall grün bleibt.

func testStore(t *testing.T) (*Store, *pgxpool.Pool, context.Context) {
	t.Helper()
	dsn := os.Getenv("TEST_DSN")
	if dsn == "" {
		t.Skip("TEST_DSN nicht gesetzt")
	}
	ctx := context.Background()
	pool, err := pgxpool.New(ctx, dsn)
	if err != nil {
		t.Fatalf("verbinden: %v", err)
	}
	t.Cleanup(pool.Close)
	return NewStore(pool), pool, ctx
}

// probeDatei legt Konto und Datei an, auf die sich ein Auftrag beziehen kann.
// Claim verbindet mit `files`, ein Auftrag ohne Datei wäre also unsichtbar –
// und ein Test, der das übersieht, prüft eine leere Schlange.
func probeDatei(t *testing.T, pool *pgxpool.Pool, ctx context.Context, marke string) uuid.UUID {
	t.Helper()
	var userID uuid.UUID
	err := pool.QueryRow(ctx,
		`INSERT INTO users (username, display_name, password, is_admin)
		 VALUES ($1, 'OCR Probe', 'x', false) RETURNING id`, "ocrprobe-"+marke,
	).Scan(&userID)
	if err != nil {
		t.Fatalf("konto anlegen: %v", err)
	}
	var fileID uuid.UUID
	err = pool.QueryRow(ctx,
		`INSERT INTO files (owner_id, folder_id, name, mime_type, size, checksum, storage_key)
		 VALUES ($1, NULL, $2, 'application/pdf', 1234, 'deadbeef', $3) RETURNING id`,
		userID, "probe-"+marke+".pdf", "users/ocrprobe-"+marke+"/home/probe.pdf",
	).Scan(&fileID)
	if err != nil {
		t.Fatalf("datei anlegen: %v", err)
	}
	t.Cleanup(func() {
		_, _ = pool.Exec(ctx, `DELETE FROM users WHERE id = $1`, userID)
	})
	return fileID
}

func status(t *testing.T, pool *pgxpool.Pool, ctx context.Context, fileID uuid.UUID) (string, int) {
	t.Helper()
	var s string
	var a int
	if err := pool.QueryRow(ctx,
		`SELECT status, attempts FROM ocr_jobs WHERE file_id = $1`, fileID,
	).Scan(&s, &a); err != nil {
		t.Fatalf("stand lesen: %v", err)
	}
	return s, a
}

// ── 1. Aufgeben ─────────────────────────────────────────────────────────────

// Die Zahl selbst ist eine Entscheidung, kein Zufall – und ein Test, der nur
// gegen die Konstante läuft, kann sie nicht schützen.
//
// **Das kam aus einem Mutationsversuch:** Setzt man maxAttempts auf 9999,
// blieben alle übrigen Tests grün (sie schleifen ja über die Konstante) und
// brauchten nur plötzlich 27 statt 0,2 Sekunden. Genau so verschwindet eine
// Entscheidung unbemerkt aus einer Anwendung.
//
// Zwei Wiederholungen fangen Vorübergehendes ab – Neustart mitten im Lauf,
// kurz fehlender Plattenplatz. Was dreimal scheitert, scheitert an der Datei.
func TestDreiVersucheSindEineEntscheidung(t *testing.T) {
	if maxAttempts != 3 {
		t.Fatalf("maxAttempts steht auf %d. Wenn das Absicht ist, gehört die "+
			"Begründung an die Konstante und diese Zahl hierher.", maxAttempts)
	}
}

// Eine Datei, die den Erkenner reproduzierbar scheitern lässt – ein kaputtes
// oder passwortgeschütztes PDF –, darf die Schlange nicht dauerhaft besetzen.
// Ohne Obergrenze liefe der Arbeiter für immer auf derselben Datei und käme an
// die übrigen nie heran: still, denn nach außen „läuft" die Erkennung.
func TestGibtNachDreiVersuchenAuf(t *testing.T) {
	s, pool, ctx := testStore(t)
	fileID := probeDatei(t, pool, ctx, "aufgeben")

	if err := s.Enqueue(ctx, fileID); err != nil {
		t.Fatalf("einreihen: %v", err)
	}

	for i := 1; i <= maxAttempts; i++ {
		// Obergrenze gegen eine aus dem Ruder gelaufene Konstante: Ohne sie
		// liefe dieser Test bei maxAttempts = 9999 minutenlang, statt etwas
		// zu sagen.
		if i > 10 {
			t.Fatalf("maxAttempts steht auf %d – das ist keine Wiederholung mehr", maxAttempts)
		}
		job, err := s.Claim(ctx)
		if err != nil {
			t.Fatalf("versuch %d: kein Auftrag mehr (%v) – zu früh aufgegeben", i, err)
		}
		if job.FileID != fileID {
			t.Fatalf("versuch %d: fremder Auftrag", i)
		}
		if job.Attempts != i {
			t.Fatalf("versuch %d: Zähler steht auf %d", i, job.Attempts)
		}
		if err := s.Fail(ctx, job, fmt.Errorf("kaputtes PDF")); err != nil {
			t.Fatalf("versuch %d: Fail: %v", i, err)
		}
	}

	// Jetzt ist Schluss – und das muss *sichtbar* sein, nicht bloß still.
	if st, a := status(t, pool, ctx, fileID); st != "failed" || a != maxAttempts {
		t.Fatalf("nach %d Versuchen: status=%q attempts=%d, erwartet failed/%d",
			maxAttempts, st, a, maxAttempts)
	}
	if _, err := s.Claim(ctx); !errors.Is(err, ErrNoJob) {
		t.Fatal("der Auftrag wird weiter angefasst – Endlosschleife auf einer kaputten Datei")
	}

	// Der Grund muss drinstehen. Ein 'failed' ohne Text wäre eine Sackgasse:
	// Man sieht, dass etwas nicht ging, aber nie was.
	var fehler string
	_ = pool.QueryRow(ctx, `SELECT last_error FROM ocr_jobs WHERE file_id = $1`, fileID).Scan(&fehler)
	if fehler == "" {
		t.Fatal("kein Fehlertext gespeichert")
	}
}

// Wird dieselbe Datei erneut hochgeladen, verdient sie eine neue Chance –
// vielleicht war genau der kaputte Scan das Problem.
func TestNeuerUploadGibtEineNeueChance(t *testing.T) {
	s, pool, ctx := testStore(t)
	fileID := probeDatei(t, pool, ctx, "neuechance")

	_ = s.Enqueue(ctx, fileID)
	for range maxAttempts {
		job, err := s.Claim(ctx)
		if err != nil {
			t.Fatalf("claim: %v", err)
		}
		_ = s.Fail(ctx, job, fmt.Errorf("kaputt"))
	}
	if st, _ := status(t, pool, ctx, fileID); st != "failed" {
		t.Fatalf("vorbedingung: status=%q", st)
	}

	// Erneuter Upload.
	if err := s.Enqueue(ctx, fileID); err != nil {
		t.Fatalf("erneut einreihen: %v", err)
	}
	if st, a := status(t, pool, ctx, fileID); st != "pending" || a != 0 {
		t.Fatalf("nach erneutem Upload: status=%q attempts=%d, erwartet pending/0", st, a)
	}
	if _, err := s.Claim(ctx); err != nil {
		t.Fatalf("der neue Versuch wird nicht angefasst: %v", err)
	}
}

// ── 2. Nach einem harten Abbruch ────────────────────────────────────────────

// Stirbt der Prozess mitten in einem Auftrag, bleibt dessen Zeile auf
// 'running' stehen. Würde Claim nur 'pending' nehmen, wäre diese Datei für
// immer verloren – und niemand erführe es, denn die Zahl der offenen Aufträge
// sähe gesund aus.
func TestNimmtHaengengebliebeneAuftraegeWiederAuf(t *testing.T) {
	s, pool, ctx := testStore(t)
	fileID := probeDatei(t, pool, ctx, "abbruch")

	_ = s.Enqueue(ctx, fileID)
	job, err := s.Claim(ctx)
	if err != nil {
		t.Fatalf("claim: %v", err)
	}
	if st, _ := status(t, pool, ctx, fileID); st != "running" {
		t.Fatalf("vorbedingung: status=%q", st)
	}

	// Hier stirbt der Prozess: kein Complete, kein Fail. Die Zeile bleibt
	// stehen, wie sie ist.
	_ = job

	wieder, err := s.Claim(ctx)
	if err != nil {
		t.Fatalf("hängengebliebener Auftrag wird nie wieder angefasst: %v", err)
	}
	if wieder.FileID != fileID {
		t.Fatal("ein anderer Auftrag kam zurück")
	}
	if wieder.Attempts != 2 {
		t.Fatalf("Zähler steht auf %d, erwartet 2 – sonst zählt der Abbruch nicht mit",
			wieder.Attempts)
	}
}

// Die Wiederaufnahme darf keine Endlosschleife werden: Eine Datei, die den
// Arbeiter reproduzierbar umbringt, wäre sonst eine Schleife aus Absturz und
// Neustart.
func TestWiederaufnahmeIstBegrenzt(t *testing.T) {
	s, pool, ctx := testStore(t)
	fileID := probeDatei(t, pool, ctx, "begrenzt")

	_ = s.Enqueue(ctx, fileID)
	for i := 1; i <= maxAttempts; i++ {
		if _, err := s.Claim(ctx); err != nil {
			t.Fatalf("versuch %d: %v", i, err)
		}
		// Jedes Mal stirbt der Prozess – der Auftrag bleibt auf 'running'.
	}
	if _, err := s.Claim(ctx); !errors.Is(err, ErrNoJob) {
		t.Fatal("unbegrenzte Wiederaufnahme: Absturz, Neustart, Absturz …")
	}
}

// Eine gelöschte Datei darf nicht mehr erkannt werden – der Arbeiter käme
// sonst an einen Blob, den es nicht mehr gibt.
func TestGeloeschteDateienWerdenNichtGenommen(t *testing.T) {
	s, pool, ctx := testStore(t)
	fileID := probeDatei(t, pool, ctx, "geloescht")

	_ = s.Enqueue(ctx, fileID)
	if _, err := pool.Exec(ctx, `UPDATE files SET is_deleted = true WHERE id = $1`, fileID); err != nil {
		t.Fatalf("löschen: %v", err)
	}
	if _, err := s.Claim(ctx); !errors.Is(err, ErrNoJob) {
		t.Fatal("eine gelöschte Datei wurde zur Erkennung genommen")
	}
}

// ── 3. Text nach alter Regel ────────────────────────────────────────────────

// Ändert sich die Extraktionsregel, müssen alte Einträge erneuert werden.
// Sonst bleibt ein stiller Rest: auffindbar, aber im Betrachter nicht auf der
// richtigen Seite – und niemand sieht, dass diese Dateien anders sind.
func TestErneuertTextNachAltererRegel(t *testing.T) {
	s, pool, ctx := testStore(t)
	fileID := probeDatei(t, pool, ctx, "veraltet")

	// Ein Text, der nach einer älteren Regel entstanden ist.
	if _, err := pool.Exec(ctx, `
		INSERT INTO file_text (file_id, content, source, lang, chars, pages, extract_version)
		VALUES ($1, 'alter text', 'ocr', 'deu', 10, 1, $2)`,
		fileID, ExtractVersion-1); err != nil {
		t.Fatalf("alten text anlegen: %v", err)
	}

	n, err := s.EnqueueOutdated(ctx, 100)
	if err != nil {
		t.Fatalf("EnqueueOutdated: %v", err)
	}
	if n == 0 {
		t.Fatal("der veraltete Eintrag wurde nicht erneut eingeplant")
	}
	if st, a := status(t, pool, ctx, fileID); st != "pending" || a != 0 {
		t.Fatalf("status=%q attempts=%d, erwartet pending/0", st, a)
	}

	// Und einer auf aktuellem Stand darf **nicht** erneut eingeplant werden –
	// sonst erkennt der Server bei jedem Start den ganzen Bestand neu.
	_, _ = pool.Exec(ctx, `UPDATE file_text SET extract_version = $2 WHERE file_id = $1`,
		fileID, ExtractVersion)
	_, _ = pool.Exec(ctx, `DELETE FROM ocr_jobs WHERE file_id = $1`, fileID)
	n, err = s.EnqueueOutdated(ctx, 100)
	if err != nil {
		t.Fatalf("EnqueueOutdated (aktuell): %v", err)
	}
	var da bool
	_ = pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM ocr_jobs WHERE file_id = $1)`, fileID).Scan(&da)
	if da {
		t.Fatalf("ein aktueller Eintrag wurde erneut eingeplant (%d Zeilen) – der Bestand würde bei jedem Start neu erkannt", n)
	}
}

// ── Abschluss ───────────────────────────────────────────────────────────────

// Complete muss Text **und** Auftragsende zusammen schreiben. Ginge nur eines
// durch, wäre die Datei entweder unauffindbar oder würde bei jedem Durchgang
// erneut erkannt.
func TestCompleteSchreibtTextUndSchliesstAb(t *testing.T) {
	s, pool, ctx := testStore(t)
	fileID := probeDatei(t, pool, ctx, "complete")

	_ = s.Enqueue(ctx, fileID)
	job, err := s.Claim(ctx)
	if err != nil {
		t.Fatalf("claim: %v", err)
	}
	res := Result{Text: "Seite eins\fSeite zwei", Source: "ocr", Lang: "deu", Chars: 20, Pages: 2}
	if err := s.Complete(ctx, job, res); err != nil {
		t.Fatalf("Complete: %v", err)
	}

	if st, _ := status(t, pool, ctx, fileID); st != "done" {
		t.Fatalf("status=%q, erwartet done", st)
	}
	seiten, quelle, err := s.TextPages(ctx, fileID)
	if err != nil {
		t.Fatalf("TextPages: %v", err)
	}
	if quelle != "ocr" {
		t.Fatalf("quelle=%q", quelle)
	}
	// Die Seitenmarke ist der Grund, aus dem es ExtractVersion gibt: Ohne sie
	// kann der Betrachter eine Fundstelle nicht auf einer Seite verorten.
	if len(seiten) != 2 {
		t.Fatalf("%d Seiten, erwartet 2 – Seitenmarke ging verloren", len(seiten))
	}
	if _, err := s.Claim(ctx); !errors.Is(err, ErrNoJob) {
		t.Fatal("ein fertiger Auftrag wird erneut genommen")
	}
}