Aether/backend/internal/documents/store.go

package documents

import (
	"context"
	"errors"
	"fmt"
	"strings"

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

// ExtractVersion ist die Fassung der Textaufbereitung.
//
// Sie steht an jedem Eintrag in file_text. Wird die Aufbereitung geändert, so
// dass alte Einträge nicht mehr taugen, wird die Zahl erhöht – beim Start
// werden sie dann erneut erkannt, statt still nach der alten Regel liegen zu
// bleiben. Ohne so eine Marke fällt der Unterschied nie auf: Der Text ist ja
// da, er kann nur weniger.
//
//	1 – Seitenumbruch (\f) bleibt im Text stehen (23.08.2026)
const ExtractVersion = 1

// ErrNoJob heißt: gerade nichts zu tun. Kein Fehler, sondern der Normalfall.
var ErrNoJob = errors.New("no pending job")

// maxAttempts begrenzt, wie oft ein Auftrag wiederholt wird.
//
// Zwei Wiederholungen fangen das ab, was vorübergehend war – ein Neustart
// mitten im Lauf, kurzzeitig fehlender Speicherplatz. Was dreimal scheitert,
// scheitert an der Datei selbst (kaputtes PDF, passwortgeschützt), und ein
// vierter Versuch verbrennt nur Rechenzeit. Der Auftrag bleibt dann als
// 'failed' mit Fehlertext stehen, statt still zu verschwinden.
const maxAttempts = 3

type Store struct {
	db *pgxpool.Pool
}

func NewStore(db *pgxpool.Pool) *Store {
	return &Store{db: db}
}

// Job ist ein Auftrag samt allem, was der Arbeiter über die Datei braucht.
type Job struct {
	ID         uuid.UUID
	FileID     uuid.UUID
	Attempts   int
	Name       string
	MimeType   string
	StorageKey string
	Size       int64
}

// Enqueue trägt eine Datei zur Texterkennung ein.
//
// ON CONFLICT statt eines zweiten Eintrags: Wird dieselbe Datei erneut
// hochgeladen, ist der alte Text hinfällig – der Auftrag wird auf 'pending'
// zurückgesetzt, statt sich daneben zu stapeln. Der Zähler geht dabei auf 0,
// weil eine neue Fassung eine neue Chance verdient: Vielleicht war es genau
// der kaputte Scan, der ersetzt wurde.
func (s *Store) Enqueue(ctx context.Context, fileID uuid.UUID) error {
	_, err := s.db.Exec(ctx, `
		INSERT INTO ocr_jobs (file_id, status)
		VALUES ($1, 'pending')
		ON CONFLICT (file_id) DO UPDATE
		   SET status = 'pending', attempts = 0, last_error = '', updated_at = NOW()`,
		fileID)
	if err != nil {
		return fmt.Errorf("enqueue ocr job: %w", err)
	}
	return nil
}

// Claim holt den ältesten offenen Auftrag und markiert ihn als laufend.
//
// FOR UPDATE SKIP LOCKED ist der Kern: Zwei Arbeiter greifen damit nie
// denselben Auftrag – der zweite überspringt die gesperrte Zeile, statt darauf
// zu warten. Heute läuft nur einer, aber die Sperre kostet nichts und erspart
// die Suche nach doppelt erkannten Dateien, sobald es zwei werden.
//
// 'running' wird mit übernommen, nicht nur 'pending': Ein Auftrag, der beim
// Neustart mittendrin war, steht sonst für immer auf 'running' und wird nie
// wieder angefasst. Der Zähler sorgt dafür, dass daraus keine Endlosschleife
// wird, falls die Datei den Server reproduzierbar umbringt.
func (s *Store) Claim(ctx context.Context) (*Job, error) {
	var j Job
	err := s.db.QueryRow(ctx, `
		UPDATE ocr_jobs
		   SET status = 'running', attempts = attempts + 1, updated_at = NOW()
		 WHERE id = (
		       SELECT j.id
		         FROM ocr_jobs j
		         JOIN files f ON f.id = j.file_id AND f.is_deleted = false
		        WHERE j.status IN ('pending', 'running')
		          AND j.attempts < $1
		        ORDER BY j.created_at
		        FOR UPDATE SKIP LOCKED
		        LIMIT 1)
		RETURNING id, file_id, attempts,
		          (SELECT name        FROM files WHERE id = file_id),
		          (SELECT mime_type   FROM files WHERE id = file_id),
		          (SELECT storage_key FROM files WHERE id = file_id),
		          (SELECT size        FROM files WHERE id = file_id)`,
		maxAttempts,
	).Scan(&j.ID, &j.FileID, &j.Attempts, &j.Name, &j.MimeType, &j.StorageKey, &j.Size)
	if errors.Is(err, pgx.ErrNoRows) {
		return nil, ErrNoJob
	}
	if err != nil {
		return nil, fmt.Errorf("claim ocr job: %w", err)
	}
	return &j, nil
}

// Complete speichert den Text und schließt den Auftrag ab. Beides in einer
// Transaktion: Ein gespeicherter Text ohne erledigten Auftrag würde beim
// nächsten Durchgang erneut erkannt, ein erledigter Auftrag ohne Text ließe die
// Datei unauffindbar zurück.
func (s *Store) Complete(ctx context.Context, job *Job, res Result) error {
	tx, err := s.db.Begin(ctx)
	if err != nil {
		return fmt.Errorf("begin: %w", err)
	}
	defer tx.Rollback(ctx)

	if _, err := tx.Exec(ctx, `
		INSERT INTO file_text (file_id, content, source, lang, chars, pages, extract_version)
		VALUES ($1, $2, $3, $4, $5, $6, $7)
		ON CONFLICT (file_id) DO UPDATE
		   SET content = EXCLUDED.content, source = EXCLUDED.source,
		       lang = EXCLUDED.lang, chars = EXCLUDED.chars,
		       pages = EXCLUDED.pages, extract_version = EXCLUDED.extract_version,
		       created_at = NOW()`,
		job.FileID, res.Text, res.Source, res.Lang, res.Chars, res.Pages, ExtractVersion,
	); err != nil {
		return fmt.Errorf("save text: %w", err)
	}

	if _, err := tx.Exec(ctx,
		`UPDATE ocr_jobs SET status = 'done', last_error = '', updated_at = NOW() WHERE id = $1`,
		job.ID,
	); err != nil {
		return fmt.Errorf("finish job: %w", err)
	}
	return tx.Commit(ctx)
}

// Fail vermerkt den Fehler. Ist der Zähler ausgeschöpft, bleibt der Auftrag als
// 'failed' liegen; sonst geht er zurück in die Schlange.
func (s *Store) Fail(ctx context.Context, job *Job, cause error) error {
	msg := cause.Error()
	if len(msg) > 500 {
		msg = msg[:500] + "…"
	}
	status := "pending"
	if job.Attempts >= maxAttempts {
		status = "failed"
	}
	_, err := s.db.Exec(ctx,
		`UPDATE ocr_jobs SET status = $2, last_error = $3, updated_at = NOW() WHERE id = $1`,
		job.ID, status, msg)
	return err
}

// Skip schließt einen Auftrag ohne Text ab – etwa wenn die Datei inzwischen
// gelöscht wurde. 'done' statt 'failed': Es ist nichts schiefgegangen.
func (s *Store) Skip(ctx context.Context, jobID uuid.UUID, reason string) error {
	_, err := s.db.Exec(ctx,
		`UPDATE ocr_jobs SET status = 'done', last_error = $2, updated_at = NOW() WHERE id = $1`,
		jobID, reason)
	return err
}

// Stats zählt die Aufträge je Zustand – für die Anzeige im Admin-Bereich und
// für die Prüfung, ob der Arbeiter überhaupt vorankommt.
type Stats struct {
	Pending int `json:"pending"`
	Running int `json:"running"`
	Done    int `json:"done"`
	Failed  int `json:"failed"`
	// Indexed ist die Zahl der Dateien mit gespeichertem Text.
	Indexed int `json:"indexed"`
}

func (s *Store) Stats(ctx context.Context) (Stats, error) {
	var st Stats
	err := s.db.QueryRow(ctx, `
		SELECT COUNT(*) FILTER (WHERE status = 'pending'),
		       COUNT(*) FILTER (WHERE status = 'running'),
		       COUNT(*) FILTER (WHERE status = 'done'),
		       COUNT(*) FILTER (WHERE status = 'failed'),
		       (SELECT COUNT(*) FROM file_text)
		  FROM ocr_jobs`,
	).Scan(&st.Pending, &st.Running, &st.Done, &st.Failed, &st.Indexed)
	if err != nil {
		return st, fmt.Errorf("ocr stats: %w", err)
	}
	return st, nil
}

// StatsForUser zählt nur, was diesem Nutzer gehört.
//
// Warum es das getrennt von Stats gibt: Den Gesamtstand darf nur der
// Administrator sehen – die Zahl aller Dokumente auf dem Server verrät etwas
// über fremde Bestände. Wissen muss es aber genau der, der die Rechnung
// hochgeladen hat: Findet seine Suche sie nicht, ist die Frage „ist sie schon
// gelesen?" die erste, die er stellt.
//
// Gezählt wird über owner_id, also die eigenen Dateien. Dateien in
// Gruppenordnern bleiben außen vor – sie gehören der Gruppe, und ihr Stand
// wäre für den Einzelnen keine Auskunft über den eigenen Bestand.
func (s *Store) StatsForUser(ctx context.Context, userID uuid.UUID) (Stats, error) {
	var st Stats
	err := s.db.QueryRow(ctx, `
		SELECT COUNT(*) FILTER (WHERE j.status = 'pending'),
		       COUNT(*) FILTER (WHERE j.status = 'running'),
		       COUNT(*) FILTER (WHERE j.status = 'done'),
		       COUNT(*) FILTER (WHERE j.status = 'failed'),
		       COUNT(*) FILTER (WHERE t.file_id IS NOT NULL)
		  FROM files f
		  JOIN ocr_jobs j ON j.file_id = f.id
		  LEFT JOIN file_text t ON t.file_id = f.id
		 WHERE f.owner_id = $1 AND f.is_deleted = false`,
		userID,
	).Scan(&st.Pending, &st.Running, &st.Done, &st.Failed, &st.Indexed)
	if err != nil {
		return st, fmt.Errorf("ocr stats for user: %w", err)
	}
	return st, nil
}

// EnqueueMissing trägt alle Bestandsdateien nach, die noch keinen Auftrag
// haben. Ohne das bliebe alles unauffindbar, was vor dem Einbau hochgeladen
// wurde – und das ist beim ersten Start genau der gesamte Bestand.
//
// Die Auswahl der Typen passiert hier in SQL statt über Supported(): Bei
// zehntausend Dateien wäre es Unfug, jede einzelne nach Go zu holen, nur um
// ihren MIME-Typ anzusehen.
func (s *Store) EnqueueMissing(ctx context.Context, limit int) (int, error) {
	tag, err := s.db.Exec(ctx, `
		INSERT INTO ocr_jobs (file_id, status)
		SELECT f.id, 'pending'
		  FROM files f
		 WHERE f.is_deleted = false
		   AND (f.mime_type ILIKE '%pdf%'
		        OR (f.mime_type ILIKE 'image/%' AND f.mime_type NOT ILIKE '%svg%')
		        OR f.name ~* '\.(pdf|png|jpe?g|tiff?|bmp|webp)$')
		   AND NOT EXISTS (SELECT 1 FROM ocr_jobs j WHERE j.file_id = f.id)
		 ORDER BY f.created_at DESC
		 LIMIT $1
		ON CONFLICT (file_id) DO NOTHING`, limit)
	if err != nil {
		return 0, fmt.Errorf("enqueue missing: %w", err)
	}
	return int(tag.RowsAffected()), nil
}

// EnqueueOutdated legt Einträge erneut in die Warteschlange, die nach einer
// älteren Regel entstanden sind.
//
// Anlass war die Seitenmarke: Bis Fassung 1 wurde der Seitenumbruch aus dem
// Text entfernt. Wer damit indiziert wurde, ist zwar auffindbar – aber der
// Betrachter kann bei einem Scan nicht sagen, auf welcher Seite das Wort
// steht. Ein stiller Rest, der sich nur durch erneutes Erkennen auflöst.
//
// Bewusst mit derselben Obergrenze wie EnqueueMissing: Bei einem großen
// Bestand soll ein Neustart nicht die ganze Platte an OCR verheizen, sondern
// den Rückstand über mehrere Starte abtragen.
func (s *Store) EnqueueOutdated(ctx context.Context, limit int) (int, error) {
	tag, err := s.db.Exec(ctx, `
		INSERT INTO ocr_jobs (file_id, status)
		SELECT t.file_id, 'pending'
		  FROM file_text t
		  JOIN files f ON f.id = t.file_id AND f.is_deleted = false
		 WHERE t.extract_version < $2
		 LIMIT $1
		ON CONFLICT (file_id) DO UPDATE
		   SET status = 'pending', attempts = 0, last_error = '', updated_at = NOW()`,
		limit, ExtractVersion)
	if err != nil {
		return 0, fmt.Errorf("enqueue outdated: %w", err)
	}
	return int(tag.RowsAffected()), nil
}

// TextPages liefert den erkannten Text einer Datei, in Seiten zerlegt.
//
// Für den PDF-Betrachter: Bei einem Scan steht im PDF selbst kein Wort, nur
// ein Bild. Ohne diesen Weg fände die Suche im Dokument nichts – ausgerechnet
// bei dem Dateityp, für den der erkannte Text überhaupt gebaut wurde.
//
// Die Zugriffsprüfung passiert im Aufrufer; hier liegt nur die Abfrage.
func (s *Store) TextPages(ctx context.Context, fileID uuid.UUID) ([]string, string, error) {
	var content, source string
	err := s.db.QueryRow(ctx,
		`SELECT content, source FROM file_text WHERE file_id = $1`, fileID,
	).Scan(&content, &source)
	if errors.Is(err, pgx.ErrNoRows) {
		return nil, "", nil
	}
	if err != nil {
		return nil, "", fmt.Errorf("text pages: %w", err)
	}
	return strings.Split(content, "\f"), source, nil
}

// DeleteText entfernt den Text einer Datei. Wird beim endgültigen Löschen
// gebraucht; solange die Datei nur im Papierkorb liegt, bleibt er stehen –
// beim Wiederherstellen wäre er sonst weg.
func (s *Store) DeleteText(ctx context.Context, fileID uuid.UUID) error {
	_, err := s.db.Exec(ctx, `DELETE FROM file_text WHERE file_id = $1`, fileID)
	return err
}