nexarch/dms/internal/workerruntime/workerruntime.go

// Package workerruntime verdrahtet die asynchronen Jobtypen des DMS (FDN-17):
// Vorschau, PDF-Konvertierung, Virenscan, Massenexport, E-Rechnungen und Texterkennung laufen
// ueber EINEN Dispatcher auf der Job-Queue (FDN-04). Jobtypen, deren Werkzeug
// oder Konfiguration fehlt, werden nicht registriert und damit nicht abgeholt;
// der Grund steht in Runtime.Notices.
package workerruntime

import (
	"context"
	"database/sql"
	"fmt"
	"log"
	"os"
	"os/exec"
	"sort"
	"sync"
	"time"

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

	dmscrypto "gitea.perlbach24.de/scripte/nexarch/dms/internal/crypto"
	"gitea.perlbach24.de/scripte/nexarch/dms/internal/downloadapi"
	"gitea.perlbach24.de/scripte/nexarch/dms/internal/einvoice"
	"gitea.perlbach24.de/scripte/nexarch/dms/internal/export"
	"gitea.perlbach24.de/scripte/nexarch/dms/internal/integrity"
	"gitea.perlbach24.de/scripte/nexarch/dms/internal/jobqueue"
	"gitea.perlbach24.de/scripte/nexarch/dms/internal/mailimport"
	"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/pdfconvert"
	"gitea.perlbach24.de/scripte/nexarch/dms/internal/plaintext"
	"gitea.perlbach24.de/scripte/nexarch/dms/internal/searchsync"
	"gitea.perlbach24.de/scripte/nexarch/dms/internal/storage"
	"gitea.perlbach24.de/scripte/nexarch/dms/internal/thumbnail"
	"gitea.perlbach24.de/scripte/nexarch/dms/internal/trigger"
	"gitea.perlbach24.de/scripte/nexarch/dms/internal/upload"
	"gitea.perlbach24.de/scripte/nexarch/dms/internal/workflowevent"
)

// Deps sind die gemeinsamen Abhaengigkeiten aller Handler.
type Deps struct {
	Pool       *pgxpool.Pool
	Storage    *storage.Service
	Crypto     *dmscrypto.Service
	TenantSlug string
	Queue      *jobqueue.Queue

	// Scanner ist der Virenscanner (clamd); nil = malwarescan.scan wird nicht abgeholt.
	Scanner malwarescan.Scanner
	// LookPath ist ueberschreibbar fuer Tests (Default exec.LookPath).
	LookPath func(string) (string, error)
	// OCR passt die Engine an (Standard: DefaultEngine).
	OCR *ocr.Processor
	// SearchIndex ist die Manticore-Verbindung; nil = searchindex.sync wird nicht abgeholt.
	SearchIndex *sql.DB
	// PlaintextMaxBytes/PlaintextMaxChars begrenzen den Klartext-Volltext (0 = Standard).
	PlaintextMaxBytes int64
	PlaintextMaxChars int
	// MailImport ist der IMAP-Importer (INT-03); nil = mailimport.poll wird nicht abgeholt.
	MailImport *mailimport.Importer
	// EInvoiceValidator ist der KoSIT-Sidecar; nil = Ergebnis not_checked.
	EInvoiceValidator *einvoice.Validator
	// PDFAValidator ist der veraPDF-Sidecar (SRC-14); nil = Ergebnis not_checked.
	PDFAValidator *pdfaconform.Validator
	// Integrity ist die Objektpruefung (FDN-23); nil = integrity.verify wird nicht abgeholt.
	Integrity *integrity.Config
	// WorkflowEvent ist die INT-13-Anbindung (leere BaseURL = abgeschaltet).
	WorkflowEvent workflowevent.Config
}

func (d Deps) lookPath(name string) error {
	lp := d.LookPath
	if lp == nil {
		lp = exec.LookPath
	}
	_, err := lp(name)
	return err
}

// Runtime haelt die registrierten Handler und die Meldungen zu nicht registrierten Jobtypen.
type Runtime struct {
	Handlers map[string]jobqueue.Handler
	Notices  []string
	// Limits sind jobtypspezifische Ressourcengrenzen (FDN-11), die Run an die Worker weitergibt.
	Limits map[string]jobqueue.Limits
	// Middleware umhuellt den Dispatcher (z. B. Fehlerverfolgung); optional.
	Middleware func(jobqueue.Handler) jobqueue.Handler
	queue      *jobqueue.Queue
}

// New registriert alle Jobtypen, deren Voraussetzungen erfuellt sind.
func New(d Deps) *Runtime {
	rt := &Runtime{Handlers: map[string]jobqueue.Handler{}, Limits: map[string]jobqueue.Limits{}, queue: d.Queue}
	skip := func(jobType, why string) {
		rt.Notices = append(rt.Notices, fmt.Sprintf("jobtyp %s wird nicht abgeholt: %s", jobType, why))
	}
	triggers := &trigger.Adapter{Pool: d.Pool, Vault: trigger.CryptoVault{Crypto: d.Crypto}, TenantSlug: d.TenantSlug}

	if err := d.lookPath("pdftoppm"); err != nil {
		skip(thumbnail.JobType, "pdftoppm nicht gefunden")
	} else {
		g := &thumbnail.Generator{Pool: d.Pool, Storage: d.Storage, Crypto: d.Crypto, TenantSlug: d.TenantSlug}
		rt.Handlers[thumbnail.JobType] = g.Handler
	}

	if err := d.lookPath("soffice"); err != nil {
		skip(pdfconvert.JobType, "soffice (LibreOffice) nicht gefunden")
	} else {
		g := &pdfconvert.Generator{Pool: d.Pool, Storage: d.Storage, Crypto: d.Crypto, TenantSlug: d.TenantSlug}
		rt.Handlers[pdfconvert.JobType] = g.Handler
	}

	if d.Scanner == nil {
		skip(malwarescan.JobType, "kein Virenscanner (clamd) konfiguriert; hochgeladene Dokumente bleiben unsichtbar, bis er laeuft")
	} else {
		p := &malwarescan.Processor{Pool: d.Pool, Storage: d.Storage, Crypto: d.Crypto, TenantSlug: d.TenantSlug, Scanner: d.Scanner}
		rt.Handlers[malwarescan.JobType] = p.Handler
	}

	p := &export.Processor{Pool: d.Pool, Storage: d.Storage, Crypto: d.Crypto, TenantSlug: d.TenantSlug, Permissions: &downloadapi.DocumentOwnerChecker{Pool: d.Pool}}
	rt.Handlers[export.JobType] = p.Handler

	if d.SearchIndex == nil {
		skip(searchsync.JobType, "kein Suchindex (Manticore) konfiguriert; Dokumente sind nicht ueber die Suche auffindbar")
	} else {
		sy := &searchsync.Syncer{Pool: d.Pool, Index: d.SearchIndex, Text: searchsync.MultiText{ocr.TextStore{Pool: d.Pool, Crypto: d.Crypto, TenantSlug: d.TenantSlug}, einvoice.TextStore{Pool: d.Pool, Crypto: d.Crypto, TenantSlug: d.TenantSlug}, plaintext.TextStore{Pool: d.Pool, Storage: d.Storage, Crypto: d.Crypto, TenantSlug: d.TenantSlug, MaxBytes: d.PlaintextMaxBytes, MaxChars: d.PlaintextMaxChars}}}
		rt.Handlers[searchsync.JobType] = sy.Handler
	}

	if d.MailImport == nil {
		skip(mailimport.JobType, "kein Mail-Importer (Staging-Verzeichnis) konfiguriert; Postfaecher werden nicht abgerufen")
	} else {
		rt.Handlers[mailimport.JobType] = d.MailImport.Handler
	}

	if d.Integrity == nil {
		skip(integrity.JobType, "keine Integritaetspruefung konfiguriert (NEXARCH_DMS_WORKER_INTEGRITY_INTERVAL_SECONDS)")
	} else {
		iv := &integrity.Verifier{Pool: d.Pool, Storage: d.Storage, Crypto: d.Crypto, TenantSlug: d.TenantSlug, Config: *d.Integrity}
		rt.Handlers[integrity.JobType] = iv.Handler
		rt.Limits[integrity.JobType] = d.Integrity.JobLimits()
	}

	ei := &einvoice.Processor{Validator: d.EInvoiceValidator, Pool: d.Pool, Storage: d.Storage, Crypto: d.Crypto, TenantSlug: d.TenantSlug}
	if _, ok := rt.Handlers[searchsync.JobType]; ok {
		ei.OnRead = func(ctx context.Context, documentID, _ string) error {
			return searchsync.Enqueue(ctx, d.Queue, documentID)
		}
	}
	// INT-20: document.einvoice_checked nach jeder Konformitaetspruefung (Erst- und Neupruefung).
	ei.OnConformanceChecked = func(ctx context.Context, documentID, _ string) error {
		return triggers.Fire(ctx, trigger.EventEInvoiceChecked, documentID)
	}
	rt.Handlers[einvoice.JobType] = ei.Handler
	rt.Handlers[einvoice.JobTypeValidate] = ei.HandlerValidate

	missing := ""
	for _, tool := range []string{"tesseract", "pdftoppm", "pdftotext", "pdfinfo"} {
		if err := d.lookPath(tool); err != nil {
			missing = tool
			break
		}
	}
	if missing != "" {
		for _, jt := range ocr.JobTypes {
			skip(jt, missing+" nicht gefunden")
		}
	} else {
		proc := d.OCR
		if proc == nil {
			proc = &ocr.Processor{Engine: ocr.DefaultEngine()}
		}
		proc.Pool, proc.Storage, proc.Crypto, proc.TenantSlug, proc.Queue = d.Pool, d.Storage, d.Crypto, d.TenantSlug, d.Queue
		_, syncIndex := rt.Handlers[searchsync.JobType]
		if proc.OnText == nil {
			proc.OnText = func(ctx context.Context, documentID, _ string) error {
				if syncIndex {
					if err := searchsync.Enqueue(ctx, d.Queue, documentID); err != nil {
						return err
					}
				}
				// INT-20: document.ocr_completed nach Vorliegen des Volltexts.
				return triggers.Fire(ctx, trigger.EventOCRCompleted, documentID)
			}
		}
		if proc.OnPDFACreated == nil {
			proc.OnPDFACreated = func(ctx context.Context, documentID, revisionID string) error {
				return pdfaconform.Enqueue(ctx, d.Pool, d.Queue, documentID, revisionID)
			}
		}
		for _, jt := range ocr.JobTypes {
			rt.Handlers[jt] = proc.Handle
		}
	}
	pc := &pdfaconform.Processor{Validator: d.PDFAValidator, Pool: d.Pool, Storage: d.Storage, Crypto: d.Crypto, TenantSlug: d.TenantSlug}
	rt.Handlers[pdfaconform.JobType] = pc.Handler

	if !d.WorkflowEvent.Enabled() {
		skip(workflowevent.JobType, "keine Workflow-Anbindung konfiguriert (NEXARCH_DMS_APP_WORKFLOW_URL)")
	} else {
		rt.Handlers[workflowevent.JobType] = workflowevent.Handler(d.WorkflowEvent)
	}
	return rt
}

// JobTypes liefert die registrierten Jobtypen, sortiert.
func (r *Runtime) JobTypes() []string {
	out := make([]string, 0, len(r.Handlers))
	for jt := range r.Handlers {
		out = append(out, jt)
	}
	sort.Strings(out)
	return out
}

// Dispatch ist der eine jobqueue.Handler: er leitet nach Jobtyp weiter und
// meldet einen unbekannten Typ als Fehler (Wiederholung, danach Dead-Letter).
func (r *Runtime) Dispatch(ctx context.Context, job *jobqueue.Job) error {
	h, ok := r.Handlers[job.JobType]
	if !ok {
		return fmt.Errorf("workerruntime: kein handler fuer jobtyp %q", job.JobType)
	}
	return h(ctx, job)
}

// Run startet n parallele Worker auf derselben Queue und blockiert bis ctx endet.
func (r *Runtime) Run(ctx context.Context, n int, workerPrefix string, poll time.Duration) {
	if n <= 0 {
		n = 1
	}
	types := r.JobTypes()
	if len(types) == 0 {
		log.Print("workerruntime: keine jobtypen registriert, nichts zu tun")
		<-ctx.Done()
		return
	}
	handler := jobqueue.Handler(r.Dispatch)
	if r.Middleware != nil {
		handler = r.Middleware(handler)
	}
	var opts []jobqueue.WorkerOption
	for jt, l := range r.Limits {
		opts = append(opts, jobqueue.WithJobTypeLimits(jt, l))
	}
	var wg sync.WaitGroup
	for i := 0; i < n; i++ {
		w := jobqueue.NewWorker(r.queue, fmt.Sprintf("%s-%d", workerPrefix, i+1), types, poll, handler, opts...)
		wg.Add(1)
		go func() { defer wg.Done(); w.Run(ctx) }()
	}
	wg.Wait()
}

// AttachEnqueuers versieht den Upload-Dienst mit den Einreihern fuer Vorschau,
// Virenscan, Texterkennung und Suchindexierung (nur fuer Jobtypen, die der
// Worker tatsaechlich abholt, sonst blieben sie ewig pending).
func AttachEnqueuers(svc *upload.Service, pool *pgxpool.Pool, queue *jobqueue.Queue, handled []string) {
	has := map[string]bool{}
	for _, jt := range handled {
		has[jt] = true
	}
	if has[thumbnail.JobType] {
		svc.SetThumbnailEnqueuer(&thumbnail.Enqueuer{Pool: pool, Queue: queue})
	}
	if has[malwarescan.JobType] {
		svc.SetMalwareScanEnqueuer(&malwarescan.Enqueuer{Pool: pool, Queue: queue})
	}
	if has[ocr.JobPlan] {
		svc.SetOCREnqueuer(&ocr.Enqueuer{Pool: pool, Queue: queue})
	}
	if has[einvoice.JobType] {
		svc.SetEInvoiceEnqueuer(&einvoice.Enqueuer{Pool: pool, Queue: queue})
	}
	if has[searchsync.JobType] {
		svc.SetSearchIndexEnqueuer(&searchsync.Enqueuer{Queue: queue})
	}
}

// HostName liefert einen Worker-Praefix aus Host und Prozess.
func HostName() string {
	h, _ := os.Hostname()
	return fmt.Sprintf("%s-%d", h, os.Getpid())
}