archivdms/internal/storage/processing_jobs.go

package storage

import (
	"context"
	"errors"
	"fmt"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgconn"
)

// --- Mandanten-Job-Queue (asynchrone Dokumentverarbeitung) -----------------
//
// processing_jobs ist die Postgres-basierte Arbeitswarteschlange für die
// NACHGELAGERTE Dokumentverarbeitung (OCR-Extraktion, Taxonomie-
// Autozuordnung, on_upload-Workflows). Bewusst KEIN Redis: die Queue ist
// Teil derselben transaktionalen Einheit wie der documents-INSERT, damit ein
// Dokument niemals ohne Job (verwaist, ewig "queued") und ein Job niemals
// ohne Dokument existieren kann — siehe CreateDocumentWithJob.
//
// WORM bleibt synchron: Datei-Hashing, Move nach store/ und chmod 0440
// passieren VOR dem Insert im Request-Pfad (internal/api stageDocument). Der
// Job fasst die archivierte Datei nur lesend an und schreibt ausschließlich
// abgeleitete Metadaten (ocr_text, document_date, Titel, Tags/Typ, Workflow-
// Ergebnisse).
//
// Fairness: der Dispatcher (internal/jobqueue) zieht pro Tick EINEN Job je
// Mandant (ClaimNextJobForTenant) statt global FIFO, damit ein Massen-Upload
// eines Mandanten die Verarbeitung aller anderen Mandanten nicht blockiert.
// Das Locking läuft über FOR UPDATE SKIP LOCKED, dadurch sind beliebig viele
// Worker-Goroutinen (und theoretisch auch mehrere Prozesse) kollisionsfrei.

// Job-Status-Werte (processing_jobs.status und documents.processing_status).
const (
	JobStatusQueued     = "queued"
	JobStatusProcessing = "processing"
	JobStatusDone       = "done"
	JobStatusFailed     = "failed"
)

// ErrNoJob wird von ClaimNextJobForTenant zurückgegeben, wenn für den
// Mandanten aktuell kein fälliger Job in der Queue liegt. Kein Fehlerfall —
// der Dispatcher behandelt das als "Mandant hat nichts zu tun".
var ErrNoJob = errors.New("storage: no queued job available")

// ProcessingJob ist ein Eintrag der Verarbeitungswarteschlange.
type ProcessingJob struct {
	ID         int64      `json:"id"`
	TenantID   int64      `json:"tenant_id"`
	DocumentID int64      `json:"document_id"`
	Status     string     `json:"status"`
	RetryCount int        `json:"retry_count"`
	// DeriveTitle merkt sich, ob der Titel beim Staging nur ein Platzhalter
	// war ("Scan 30.07.2026") und daher aus dem OCR-Text neu abgeleitet werden
	// darf. Bei einem vom Benutzer/SFTP-Dateinamen vorgegebenen Titel ist das
	// false, damit die asynchrone Verarbeitung eine bewusste Vorgabe niemals
	// überschreibt (entspricht dem alten synchronen Verhalten: Titel != "" ->
	// OCR-Ableitung wurde übersprungen).
	DeriveTitle   bool       `json:"derive_title"`
	ErrorMessage  string     `json:"error_message,omitempty"`
	NextAttemptAt time.Time  `json:"next_attempt_at"`
	StartedAt     *time.Time `json:"started_at,omitempty"`
	CreatedAt     time.Time  `json:"created_at"`
	UpdatedAt     time.Time  `json:"updated_at"`
}

// initProcessingJobsSchema legt die Queue-Tabelle an und ergänzt
// documents.processing_status. Idempotent, aus (*Store).initSchema()
// aufgerufen — siehe migrations/023_processing_jobs.sql (Doku).
func (s *Store) initProcessingJobsSchema(ctx context.Context) error {
	_, err := s.db.Exec(ctx, `
		CREATE TABLE IF NOT EXISTS processing_jobs (
			id              BIGSERIAL PRIMARY KEY,
			tenant_id       BIGINT NOT NULL,
			document_id     BIGINT NOT NULL REFERENCES documents(id) ON DELETE CASCADE,
			status          TEXT NOT NULL DEFAULT 'queued',
			retry_count     INT NOT NULL DEFAULT 0,
			derive_title    BOOLEAN NOT NULL DEFAULT false,
			error_message   TEXT,
			next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT now(),
			started_at      TIMESTAMPTZ,
			created_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
			updated_at      TIMESTAMPTZ NOT NULL DEFAULT now()
		);
	`)
	if err != nil {
		return fmt.Errorf("storage: create processing_jobs table: %w", err)
	}

	// Nachträglich ergänzte Spalten (idempotent, für Bestands-Installationen
	// die eine frühere Variante der Tabelle haben).
	_, err = s.db.Exec(ctx, `
		ALTER TABLE processing_jobs ADD COLUMN IF NOT EXISTS derive_title BOOLEAN NOT NULL DEFAULT false;
		ALTER TABLE processing_jobs ADD COLUMN IF NOT EXISTS next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT now();
		ALTER TABLE processing_jobs ADD COLUMN IF NOT EXISTS started_at TIMESTAMPTZ;
		ALTER TABLE processing_jobs ADD COLUMN IF NOT EXISTS error_message TEXT;
	`)
	if err != nil {
		return fmt.Errorf("storage: alter processing_jobs table: %w", err)
	}

	// Dispatch-Index: der Claim-Pfad filtert immer auf
	// (tenant_id, status, next_attempt_at) und sortiert nach created_at.
	_, err = s.db.Exec(ctx, `
		CREATE INDEX IF NOT EXISTS idx_processing_jobs_dispatch
			ON processing_jobs(tenant_id, status, next_attempt_at, created_at);
		CREATE INDEX IF NOT EXISTS idx_processing_jobs_document
			ON processing_jobs(document_id);
		CREATE INDEX IF NOT EXISTS idx_processing_jobs_status
			ON processing_jobs(status);
	`)
	if err != nil {
		return fmt.Errorf("storage: create processing_jobs indexes: %w", err)
	}

	// processing_status auf documents: Default 'done' — der komplette
	// Altbestand wurde noch synchron im Request verarbeitet und ist damit
	// per Definition fertig. Bewusst KEIN Backfill-Migrationscode, der
	// Spalten-Default deckt alle Bestandszeilen ab.
	_, err = s.db.Exec(ctx, `
		ALTER TABLE documents ADD COLUMN IF NOT EXISTS processing_status TEXT NOT NULL DEFAULT 'done';
	`)
	if err != nil {
		return fmt.Errorf("storage: alter documents add processing_status: %w", err)
	}
	return nil
}

// CreateDocumentWithJob legt Dokument UND Verarbeitungsjob in EINER
// Transaktion an. Entweder beides oder nichts — ein Dokument ohne Job würde
// nie OCR/Taxonomie bekommen, ein Job ohne Dokument liefe ins Leere.
//
// Das Dokument wird mit processing_status='queued' angelegt; der Worker setzt
// es später auf 'done' bzw. 'failed'.
func (s *Store) CreateDocumentWithJob(ctx context.Context, req CreateDocumentRequest, deriveTitle bool) (*Document, *ProcessingJob, error) {
	tx, err := s.db.Begin(ctx)
	if err != nil {
		return nil, nil, fmt.Errorf("storage: begin create document tx: %w", err)
	}
	defer tx.Rollback(ctx)

	var d Document
	err = tx.QueryRow(ctx, `
		INSERT INTO documents (tenant_id, title, doc_type, correspondent, storage_path, content_hash, ocr_text, retain_until, document_date, source, source_ref, created_by, processing_status)
		VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, 'queued')
		RETURNING id, tenant_id, title, COALESCE(doc_type, ''), COALESCE(correspondent, ''), doc_type_id, correspondent_id, storage_path, content_hash,
			COALESCE(ocr_text, ''), retain_until, document_date, COALESCE(source, ''), COALESCE(source_ref, ''), created_by, title_manually_set, has_thumbnail, processing_status, created_at, updated_at
	`, req.TenantID, req.Title, nullIfEmpty(req.DocType), nullIfEmpty(req.Correspondent), req.StoragePath, req.ContentHash,
		nullIfEmpty(req.OCRText), req.RetainUntil, req.DocumentDate, nullIfEmpty(req.Source), nullIfEmpty(req.SourceRef), req.CreatedBy,
	).Scan(&d.ID, &d.TenantID, &d.Title, &d.DocType, &d.Correspondent, &d.DocTypeID, &d.CorrespondentID, &d.StoragePath, &d.ContentHash,
		&d.OCRText, &d.RetainUntil, &d.DocumentDate, &d.Source, &d.SourceRef, &d.CreatedBy, &d.TitleManuallySet, &d.HasThumbnail, &d.ProcessingStatus, &d.CreatedAt, &d.UpdatedAt)
	if err != nil {
		var pgErr *pgconn.PgError
		if errors.As(err, &pgErr) && pgErr.Code == "23505" {
			return nil, nil, ErrDuplicateContentHash
		}
		return nil, nil, fmt.Errorf("storage: create document: %w", err)
	}

	var j ProcessingJob
	err = tx.QueryRow(ctx, `
		INSERT INTO processing_jobs (tenant_id, document_id, status, derive_title)
		VALUES ($1, $2, 'queued', $3)
		RETURNING id, tenant_id, document_id, status, retry_count, derive_title, COALESCE(error_message, ''), next_attempt_at, started_at, created_at, updated_at
	`, req.TenantID, d.ID, deriveTitle).Scan(&j.ID, &j.TenantID, &j.DocumentID, &j.Status, &j.RetryCount, &j.DeriveTitle,
		&j.ErrorMessage, &j.NextAttemptAt, &j.StartedAt, &j.CreatedAt, &j.UpdatedAt)
	if err != nil {
		return nil, nil, fmt.Errorf("storage: create processing job: %w", err)
	}

	if err := tx.Commit(ctx); err != nil {
		return nil, nil, fmt.Errorf("storage: commit create document tx: %w", err)
	}
	return &d, &j, nil
}

// TenantsWithDueJobs liefert alle Mandanten-IDs, für die aktuell mindestens
// ein fälliger (queued, next_attempt_at <= now()) Job in der Queue liegt.
// Grundlage des Round-Robin-Dispatch: der Dispatcher iteriert über diese
// Liste und zieht je Mandant genau einen Job pro Runde.
func (s *Store) TenantsWithDueJobs(ctx context.Context) ([]int64, error) {
	rows, err := s.db.Query(ctx, `
		SELECT DISTINCT tenant_id FROM processing_jobs
		WHERE status = 'queued' AND next_attempt_at <= now()
		ORDER BY tenant_id
	`)
	if err != nil {
		return nil, fmt.Errorf("storage: list tenants with due jobs: %w", err)
	}
	defer rows.Close()

	out := make([]int64, 0)
	for rows.Next() {
		var id int64
		if err := rows.Scan(&id); err != nil {
			return nil, fmt.Errorf("storage: scan tenant with due jobs: %w", err)
		}
		out = append(out, id)
	}
	return out, rows.Err()
}

// ClaimNextJobForTenant markiert den ältesten fälligen Job eines Mandanten
// atomar als 'processing' und gibt ihn zurück. FOR UPDATE SKIP LOCKED sorgt
// dafür, dass parallele Worker sich nicht gegenseitig blockieren und kein Job
// doppelt gezogen wird. Liefert ErrNoJob, wenn nichts anliegt.
func (s *Store) ClaimNextJobForTenant(ctx context.Context, tenantID int64) (*ProcessingJob, error) {
	var j ProcessingJob
	err := s.db.QueryRow(ctx, `
		UPDATE processing_jobs SET status = 'processing', started_at = now(), updated_at = now()
		WHERE id = (
			SELECT id FROM processing_jobs
			WHERE tenant_id = $1 AND status = 'queued' AND next_attempt_at <= now()
			ORDER BY created_at
			LIMIT 1
			FOR UPDATE SKIP LOCKED
		)
		RETURNING id, tenant_id, document_id, status, retry_count, derive_title, COALESCE(error_message, ''), next_attempt_at, started_at, created_at, updated_at
	`, tenantID).Scan(&j.ID, &j.TenantID, &j.DocumentID, &j.Status, &j.RetryCount, &j.DeriveTitle,
		&j.ErrorMessage, &j.NextAttemptAt, &j.StartedAt, &j.CreatedAt, &j.UpdatedAt)
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return nil, ErrNoJob
		}
		return nil, fmt.Errorf("storage: claim processing job: %w", err)
	}
	// Dokument-Status mitziehen, damit die UI "wird verarbeitet" anzeigen kann.
	// Best-effort: der Job ist bereits verbindlich geclaimt, ein Fehler beim
	// reinen Anzeigestatus darf die Verarbeitung nicht verhindern.
	_ = s.SetDocumentProcessingStatus(ctx, j.DocumentID, j.TenantID, JobStatusProcessing)
	return &j, nil
}

// MarkJobDone schließt einen Job erfolgreich ab und setzt das Dokument auf
// processing_status='done'.
func (s *Store) MarkJobDone(ctx context.Context, jobID, tenantID, documentID int64) error {
	_, err := s.db.Exec(ctx, `
		UPDATE processing_jobs SET status = 'done', error_message = NULL, updated_at = now()
		WHERE id = $1 AND tenant_id = $2
	`, jobID, tenantID)
	if err != nil {
		return fmt.Errorf("storage: mark job done: %w", err)
	}
	if err := s.SetDocumentProcessingStatus(ctx, documentID, tenantID, JobStatusDone); err != nil && !errors.Is(err, ErrDocumentNotFound) {
		return err
	}
	return nil
}

// MarkJobFailed verbucht einen fehlgeschlagenen Versuch.
//
// Solange retry_count < maxRetries geht der Job zurück auf 'queued' mit
// exponentiellem Backoff (2^retry_count Sekunden ab jetzt). Ist das Limit
// erreicht, bleibt er dauerhaft auf 'failed' — kein weiterer Automatik-Retry,
// ein manueller Retry (Phase 3, Frontend-Button) muss ihn explizit
// requeuen (RequeueJob).
//
// Rückgabewert requeued sagt dem Aufrufer, ob noch ein Versuch folgt (für
// Logging/Audit-Detail).
func (s *Store) MarkJobFailed(ctx context.Context, jobID, tenantID, documentID int64, jobErr string, maxRetries int) (requeued bool, err error) {
	var retryCount int
	err = s.db.QueryRow(ctx, `
		SELECT retry_count FROM processing_jobs WHERE id = $1 AND tenant_id = $2
	`, jobID, tenantID).Scan(&retryCount)
	if err != nil {
		return false, fmt.Errorf("storage: load job retry_count: %w", err)
	}

	next := retryCount + 1
	if next > maxRetries {
		if _, err := s.db.Exec(ctx, `
			UPDATE processing_jobs SET status = 'failed', retry_count = $3, error_message = $4, updated_at = now()
			WHERE id = $1 AND tenant_id = $2
		`, jobID, tenantID, next, jobErr); err != nil {
			return false, fmt.Errorf("storage: mark job failed: %w", err)
		}
		if err := s.SetDocumentProcessingStatus(ctx, documentID, tenantID, JobStatusFailed); err != nil && !errors.Is(err, ErrDocumentNotFound) {
			return false, err
		}
		return false, nil
	}

	// Exponentielles Backoff: 2^retry_count Sekunden (1s, 2s, 4s, 8s, 16s ...)
	// ab dem aktuellen Fehlschlag gerechnet.
	backoff := time.Duration(1<<uint(retryCount)) * time.Second
	if _, err := s.db.Exec(ctx, `
		UPDATE processing_jobs
		SET status = 'queued', retry_count = $3, error_message = $4,
		    next_attempt_at = now() + ($5::double precision * interval '1 second'), started_at = NULL, updated_at = now()
		WHERE id = $1 AND tenant_id = $2
	`, jobID, tenantID, next, jobErr, int64(backoff.Seconds())); err != nil {
		return false, fmt.Errorf("storage: requeue job: %w", err)
	}
	if err := s.SetDocumentProcessingStatus(ctx, documentID, tenantID, JobStatusQueued); err != nil && !errors.Is(err, ErrDocumentNotFound) {
		return true, err
	}
	return true, nil
}

// ReapStaleJobs setzt hängengebliebene 'processing'-Jobs zurück, deren
// started_at älter als timeout ist (Prozess-Neustart mitten in der
// Verarbeitung, gestorbener OCR-Subprozess, ...). Zurückgesetzt wird wie ein
// normaler Fehlschlag: retry_count hoch, exponentielles Backoff, ab
// maxRetries dauerhaft 'failed'. Gibt die Anzahl betroffener Jobs zurück.
func (s *Store) ReapStaleJobs(ctx context.Context, timeout time.Duration, maxRetries int) (int, error) {
	secs := int64(timeout.Seconds())
	if secs < 1 {
		secs = 1
	}
	rows, err := s.db.Query(ctx, `
		SELECT id, tenant_id, document_id FROM processing_jobs
		WHERE status = 'processing' AND started_at IS NOT NULL
		  AND started_at < now() - ($1::double precision * interval '1 second')
	`, secs)
	if err != nil {
		return 0, fmt.Errorf("storage: find stale jobs: %w", err)
	}
	type staleJob struct{ id, tenantID, documentID int64 }
	var stale []staleJob
	for rows.Next() {
		var sj staleJob
		if err := rows.Scan(&sj.id, &sj.tenantID, &sj.documentID); err != nil {
			rows.Close()
			return 0, fmt.Errorf("storage: scan stale job: %w", err)
		}
		stale = append(stale, sj)
	}
	rows.Close()
	if err := rows.Err(); err != nil {
		return 0, fmt.Errorf("storage: iterate stale jobs: %w", err)
	}

	for _, sj := range stale {
		if _, err := s.MarkJobFailed(ctx, sj.id, sj.tenantID, sj.documentID,
			fmt.Sprintf("job timed out after %s (reaper)", timeout), maxRetries); err != nil {
			return len(stale), fmt.Errorf("storage: reap job %d: %w", sj.id, err)
		}
	}
	return len(stale), nil
}

// CountProcessingJobsByStatus liefert die Queue-Länge je Status über ALLE
// Mandanten hinweg. Bewusst ohne tenant_id-Filter: einziger Aufrufer ist der
// betriebsinterne Prometheus-Endpunkt GET /metrics (FDN-08), der nur
// aggregierte Zahlen ohne Mandantenbezug ausgibt — es verlassen keine
// mandantenbezogenen Daten das System. Für mandantenbezogene Auswertungen
// niemals diese Funktion nutzen.
func (s *Store) CountProcessingJobsByStatus(ctx context.Context) (map[string]int64, error) {
	rows, err := s.db.Query(ctx, `
		SELECT status, count(*) FROM processing_jobs GROUP BY status
	`)
	if err != nil {
		return nil, fmt.Errorf("storage: count processing jobs by status: %w", err)
	}
	defer rows.Close()

	out := map[string]int64{}
	for rows.Next() {
		var status string
		var n int64
		if err := rows.Scan(&status, &n); err != nil {
			return nil, fmt.Errorf("storage: scan processing job count: %w", err)
		}
		out[status] = n
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("storage: iterate processing job counts: %w", err)
	}
	return out, nil
}

// RequeueJob stellt einen Job (typischerweise einen dauerhaft 'failed'
// gelaufenen) wieder in die Queue und setzt retry_count zurück. Wird vom
// späteren manuellen Retry-Endpunkt (Phase 3) genutzt; tenant-scoped.
func (s *Store) RequeueJob(ctx context.Context, jobID, tenantID int64) error {
	var documentID int64
	err := s.db.QueryRow(ctx, `
		UPDATE processing_jobs
		SET status = 'queued', retry_count = 0, error_message = NULL,
		    next_attempt_at = now(), started_at = NULL, updated_at = now()
		WHERE id = $1 AND tenant_id = $2
		RETURNING document_id
	`, jobID, tenantID).Scan(&documentID)
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return ErrNoJob
		}
		return fmt.Errorf("storage: requeue job: %w", err)
	}
	if err := s.SetDocumentProcessingStatus(ctx, documentID, tenantID, JobStatusQueued); err != nil && !errors.Is(err, ErrDocumentNotFound) {
		return err
	}
	return nil
}

// GetJobForDocument liefert den jüngsten Job eines Dokuments (tenant-scoped).
// Für den späteren Frontend-Status/Retry-Endpunkt.
func (s *Store) GetJobForDocument(ctx context.Context, documentID, tenantID int64) (*ProcessingJob, error) {
	var j ProcessingJob
	err := s.db.QueryRow(ctx, `
		SELECT id, tenant_id, document_id, status, retry_count, derive_title, COALESCE(error_message, ''), next_attempt_at, started_at, created_at, updated_at
		FROM processing_jobs
		WHERE document_id = $1 AND tenant_id = $2
		ORDER BY id DESC LIMIT 1
	`, documentID, tenantID).Scan(&j.ID, &j.TenantID, &j.DocumentID, &j.Status, &j.RetryCount, &j.DeriveTitle,
		&j.ErrorMessage, &j.NextAttemptAt, &j.StartedAt, &j.CreatedAt, &j.UpdatedAt)
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return nil, ErrNoJob
		}
		return nil, fmt.Errorf("storage: get job for document: %w", err)
	}
	return &j, nil
}

// SetDocumentProcessingStatus aktualisiert documents.processing_status,
// tenant-scoped. Reiner UI-/Ablaufstatus (kein ACL-relevanter Inhalt), daher
// bewusst OHNE SyncIndex — analog SetDocumentHasThumbnail.
func (s *Store) SetDocumentProcessingStatus(ctx context.Context, documentID, tenantID int64, status string) error {
	tag, err := s.db.Exec(ctx, `
		UPDATE documents SET processing_status = $1, updated_at = now() WHERE id = $2 AND tenant_id = $3
	`, status, documentID, tenantID)
	if err != nil {
		return fmt.Errorf("storage: set document processing_status: %w", err)
	}
	if tag.RowsAffected() == 0 {
		return fmt.Errorf("storage: set document processing_status: %w", ErrDocumentNotFound)
	}
	return nil
}