nexarch/dms/internal/workerruntime/workerruntime_test.go

package workerruntime

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"os/exec"
	"reflect"
	"sort"
	"strings"
	"testing"
	"time"

	"gitea.perlbach24.de/scripte/nexarch/dms/internal/einvoice"
	"gitea.perlbach24.de/scripte/nexarch/dms/internal/export"
	"gitea.perlbach24.de/scripte/nexarch/dms/internal/jobqueue"
	"gitea.perlbach24.de/scripte/nexarch/dms/internal/malwarescan"
	"gitea.perlbach24.de/scripte/nexarch/dms/internal/ocr"
	"gitea.perlbach24.de/scripte/nexarch/dms/internal/pdfaconform"
	"gitea.perlbach24.de/scripte/nexarch/dms/internal/testcorpus"
	"gitea.perlbach24.de/scripte/nexarch/dms/internal/testenv"
	"gitea.perlbach24.de/scripte/nexarch/dms/internal/thumbnail"
)

type cleanScanner struct{}

func (cleanScanner) Scan(context.Context, io.Reader) (bool, string, error) { return false, "", nil }

func requireTools(t *testing.T) {
	t.Helper()
	for _, tool := range []string{"tesseract", "pdftoppm", "pdftotext", "pdfinfo"} {
		if _, err := exec.LookPath(tool); err != nil {
			t.Skipf("%s nicht installiert", tool)
		}
	}
}

func uploadPDF(t *testing.T, e *testenv.Env, rt *Runtime, queue *jobqueue.Queue, name string, data []byte) (docID, revID string) {
	t.Helper()
	svc := e.UploadService(t)
	AttachEnqueuers(svc, e.Pool, queue, rt.JobTypes())
	ctx := context.Background()
	sid, err := svc.StartUpload(ctx, name, "application/pdf", int64(len(data)), nil, e.UserID)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := svc.UploadChunk(ctx, sid, 0, bytes.NewReader(data)); err != nil {
		t.Fatal(err)
	}
	docID, revID, err = svc.Complete(ctx, sid, false)
	if err != nil {
		t.Fatal(err)
	}
	return docID, revID
}

func count(t *testing.T, e *testenv.Env, q string, args ...any) int {
	t.Helper()
	var n int
	if err := e.Pool.QueryRow(context.Background(), q, args...).Scan(&n); err != nil {
		t.Fatal(err)
	}
	return n
}

func dumpJobs(t *testing.T, e *testenv.Env) {
	t.Helper()
	rows, err := e.Pool.Query(context.Background(), `SELECT job_type, status, attempts, COALESCE(left(last_error, 160), '') FROM processing_jobs ORDER BY created_at`)
	if err != nil {
		return
	}
	defer rows.Close()
	for rows.Next() {
		var jt, st, le string
		var at int
		_ = rows.Scan(&jt, &st, &at, &le)
		t.Logf("job %s status=%s versuche=%d fehler=%q", jt, st, at, le)
	}
}

func waitFor(t *testing.T, e *testenv.Env, what string, timeout time.Duration, cond func() bool) {
	t.Helper()
	deadline := time.Now().Add(timeout)
	for time.Now().Before(deadline) {
		if cond() {
			return
		}
		time.Sleep(100 * time.Millisecond)
	}
	dumpJobs(t, e)
	t.Fatalf("zeitueberschreitung beim warten auf: %s", what)
}

// Pruefung 1: Ende zu Ende - Upload, alle Jobtypen laufen ueber EINEN Dispatcher bis zum Endzustand.
func TestEndToEnd_AllJobTypesReachTheirFinalState(t *testing.T) {
	requireTools(t)
	e := testenv.New(t)
	queue := jobqueue.NewQueue(e.Pool, 5*time.Minute)
	rt := New(Deps{Pool: e.Pool, Storage: e.Storage, Crypto: e.Crypto, TenantSlug: testenv.TenantSlug, Queue: queue, Scanner: cleanScanner{}})
	for _, jt := range []string{thumbnail.JobType, malwarescan.JobType, export.JobType, ocr.JobPlan, ocr.JobPages} {
		if _, ok := rt.Handlers[jt]; !ok {
			t.Fatalf("jobtyp %s nicht registriert (hinweise: %v)", jt, rt.Notices)
		}
	}

	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan struct{})
	go func() { rt.Run(ctx, 3, "e2e", 20*time.Millisecond); close(done) }()
	defer func() { cancel(); <-done }()

	docID, revID := uploadPDF(t, e, rt, queue, "durchsuchbar.pdf", testcorpus.GeneratePDF("Rechnung", []string{"Rechnung Nr. 2026-0001", "Musterhandel Nordlicht GmbH", "Gesamtbetrag 77,50 EUR"}))

	waitFor(t, e, "vorschau", 60*time.Second, func() bool {
		th, err := thumbnail.GetForRevision(ctx, e.Pool, revID)
		return err == nil && th.Status == thumbnail.StatusReady
	})
	waitFor(t, e, "virenscan sauber", 60*time.Second, func() bool {
		return count(t, e, `SELECT count(*) FROM revision_scans WHERE revision_id = $1 AND status = 'clean'`, revID) == 1
	})
	waitFor(t, e, "texterkennung", 60*time.Second, func() bool {
		st, err := ocr.GetState(ctx, e.Pool, revID)
		return err == nil && (st.Status == ocr.StatusReady || st.Status == ocr.StatusSkipped)
	})

	exportID, err := export.Enqueue(ctx, e.Pool, queue, e.UserID, []string{docID}, export.Limits{})
	if err != nil {
		t.Fatal(err)
	}
	waitFor(t, e, "export", 60*time.Second, func() bool {
		ex, err := export.Get(ctx, e.Pool, exportID)
		return err == nil && ex.Status == export.StatusReady
	})

	waitFor(t, e, "alle jobs im endzustand succeeded", 30*time.Second, func() bool {
		return count(t, e, `SELECT count(*) FROM processing_jobs WHERE status <> 'succeeded'`) == 0
	})
}

// Pruefung 2 / AK 2: fehlendes Werkzeug oder fehlende Konfiguration -> Jobtyp nicht abgeholt, mit Grund gemeldet;
// der Upload reiht ihn dann auch nicht ein.
func TestMissingToolsAndConfigurationAreReportedAndNotEnqueued(t *testing.T) {
	e := testenv.New(t)
	queue := jobqueue.NewQueue(e.Pool, 5*time.Minute)
	noTools := func(string) (string, error) { return "", errors.New("nicht gefunden") }
	rt := New(Deps{Pool: e.Pool, Storage: e.Storage, Crypto: e.Crypto, TenantSlug: testenv.TenantSlug, Queue: queue, LookPath: noTools})

	got := strings.Join(rt.Notices, "\n")
	for _, want := range []string{thumbnail.JobType, "pdfconvert.generate", malwarescan.JobType, "clamd", ocr.JobPlan, ocr.JobPages} {
		if !strings.Contains(got, want) {
			t.Errorf("hinweis auf %q fehlt:\n%s", want, got)
		}
	}
	want := []string{einvoice.JobType, einvoice.JobTypeValidate, export.JobType, pdfaconform.JobType}
	sort.Strings(want)
	if types := rt.JobTypes(); !reflect.DeepEqual(types, want) {
		t.Errorf("registrierte jobtypen: %v, erwartet nur %v", types, want)
	}

	uploadPDF(t, e, rt, queue, "x.pdf", testcorpus.GeneratePDF("x", []string{"Testinhalt Zeile eins", "Zeile zwei"}))
	if n := count(t, e, `SELECT count(*) FROM processing_jobs WHERE job_type IN ($1, $2, $3)`, thumbnail.JobType, malwarescan.JobType, ocr.JobPlan); n != 0 {
		t.Errorf("%d jobs fuer nicht abgeholte typen eingereiht (sie blieben ewig pending)", n)
	}
}

// Pruefung 3 / AK 1: ein unbekannter Jobtyp schlaegt sichtbar fehl und landet nach den Versuchen in der Dead-Letter-Queue.
func TestUnknownJobTypeFailsVisiblyAndEndsInDeadLetterQueue(t *testing.T) {
	e := testenv.New(t)
	queue := jobqueue.NewQueue(e.Pool, 5*time.Minute)
	rt := New(Deps{Pool: e.Pool, Storage: e.Storage, Crypto: e.Crypto, TenantSlug: testenv.TenantSlug, Queue: queue, LookPath: func(string) (string, error) { return "", errors.New("-") }})

	if err := rt.Dispatch(context.Background(), &jobqueue.Job{JobType: "gibt.es.nicht"}); err == nil || !strings.Contains(err.Error(), "kein handler") {
		t.Fatalf("dispatch: %v", err)
	}
	id, err := queue.Enqueue(context.Background(), "gibt.es.nicht", map[string]string{}, jobqueue.EnqueueOptions{MaxAttempts: 1})
	if err != nil {
		t.Fatal(err)
	}
	w := jobqueue.NewWorker(queue, "w", []string{"gibt.es.nicht"}, time.Millisecond, rt.Dispatch)
	w.RunOnce(context.Background())
	job, err := queue.Status(context.Background(), id)
	if err != nil {
		t.Fatal(err)
	}
	if job.Status != "dead_letter" || job.LastError == nil || !strings.Contains(*job.LastError, "kein handler") {
		t.Fatalf("job: status %s, fehler %v", job.Status, job.LastError)
	}
}

// AK 3: mehrere Worker verarbeiten dieselbe Queue, jeder Job genau einmal.
func TestMultipleWorkersProcessEachJobExactlyOnce(t *testing.T) {
	requireTools(t)
	e := testenv.New(t)
	queue := jobqueue.NewQueue(e.Pool, 5*time.Minute)
	rt := New(Deps{Pool: e.Pool, Storage: e.Storage, Crypto: e.Crypto, TenantSlug: testenv.TenantSlug, Queue: queue, Scanner: cleanScanner{}})
	const docs = 15
	revs := make([]string, docs)
	for i := range revs {
		_, revs[i] = uploadPDF(t, e, rt, queue, fmt.Sprintf("d%02d.pdf", i), testcorpus.GeneratePDF("t", []string{fmt.Sprintf("Dokument Nummer %d", i), "Musterhandel Nordlicht GmbH Zeile"}))
	}
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan struct{})
	go func() { rt.Run(ctx, 4, "par", 10*time.Millisecond); close(done) }()
	defer func() { cancel(); <-done }()

	waitFor(t, e, "alle jobs abgeschlossen", 3*time.Minute, func() bool {
		return count(t, e, `SELECT count(*) FROM processing_jobs WHERE status <> 'succeeded'`) == 0 && count(t, e, `SELECT count(*) FROM processing_jobs`) >= docs*2
	})
	if n := count(t, e, `SELECT count(*) FROM revision_thumbnails WHERE status = 'ready'`); n != docs {
		t.Errorf("%d von %d vorschauen fertig", n, docs)
	}
	if n := count(t, e, `SELECT count(*) FROM processing_jobs WHERE attempts > 1`); n != 0 {
		t.Errorf("%d jobs wurden mehrfach versucht (doppelverarbeitung/fehler)", n)
	}
}