259 lines
10 KiB
Go
259 lines
10 KiB
Go
// worker ist der Hintergrund-Dienst des DMS — verarbeitet lange laufende
|
|
// Aufgaben (Indexierung, OCR, Storage-Vorgaenge in spaeteren Kacheln),
|
|
// getrennt vom App-Prozess (siehe cmd/app). DOC-16: verarbeitet zusaetzlich
|
|
// Requeue-Jobs fuer fehlgeschlagene RET-05-Registrierungen.
|
|
package main
|
|
|
|
import (
|
|
"context"
|
|
"crypto/x509"
|
|
"errors"
|
|
"log"
|
|
"net/http"
|
|
"os"
|
|
"os/signal"
|
|
"strconv"
|
|
"syscall"
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
|
|
dmscrypto "gitea.perlbach24.de/scripte/nexarch/dms/internal/crypto"
|
|
"gitea.perlbach24.de/scripte/nexarch/dms/internal/einvoice"
|
|
"gitea.perlbach24.de/scripte/nexarch/dms/internal/errortracking"
|
|
"gitea.perlbach24.de/scripte/nexarch/dms/internal/jobqueue"
|
|
"gitea.perlbach24.de/scripte/nexarch/dms/internal/logging"
|
|
"gitea.perlbach24.de/scripte/nexarch/dms/internal/mailimport"
|
|
"gitea.perlbach24.de/scripte/nexarch/dms/internal/malwarescan"
|
|
"gitea.perlbach24.de/scripte/nexarch/dms/internal/metrics"
|
|
"gitea.perlbach24.de/scripte/nexarch/dms/internal/retentionclient"
|
|
"gitea.perlbach24.de/scripte/nexarch/dms/internal/retentiondestroy"
|
|
"gitea.perlbach24.de/scripte/nexarch/dms/internal/searchindex"
|
|
"gitea.perlbach24.de/scripte/nexarch/dms/internal/shared"
|
|
"gitea.perlbach24.de/scripte/nexarch/dms/internal/storage"
|
|
"gitea.perlbach24.de/scripte/nexarch/dms/internal/upload"
|
|
"gitea.perlbach24.de/scripte/nexarch/dms/internal/workerruntime"
|
|
)
|
|
|
|
func main() {
|
|
log.Printf("dms-worker gestartet (version %s)", shared.Version)
|
|
|
|
logger := logging.NewDefault()
|
|
metricsReg := metrics.NewRegistry()
|
|
tracker := errortracking.New(logger, 5, time.Minute)
|
|
|
|
metricsAddr := os.Getenv("NEXARCH_DMS_WORKER_METRICS_ADDR")
|
|
if metricsAddr == "" {
|
|
metricsAddr = ":8601"
|
|
}
|
|
go func() {
|
|
if err := http.ListenAndServe(metricsAddr, metricsReg.Handler()); err != nil {
|
|
log.Printf("dms-worker: /metrics-server beendet: %v", err)
|
|
}
|
|
}()
|
|
|
|
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
|
defer stop()
|
|
|
|
lease, err := jobqueue.ParseLeaseSeconds(os.Getenv("NEXARCH_DMS_WORKER_JOB_LEASE_SECONDS"))
|
|
if err != nil {
|
|
log.Fatalf("dms-worker: NEXARCH_DMS_WORKER_JOB_LEASE_SECONDS: %v", err)
|
|
}
|
|
|
|
var (
|
|
queue *jobqueue.Queue
|
|
client *retentionclient.Client
|
|
)
|
|
if dsn := os.Getenv("NEXARCH_DMS_WORKER_TENANT_DSN"); dsn != "" {
|
|
pool, err := pgxpool.New(ctx, dsn)
|
|
if err != nil {
|
|
log.Fatalf("dms-worker: datenbankverbindung: %v", err)
|
|
}
|
|
queue = jobqueue.NewQueue(pool, lease)
|
|
client = retentionclient.New(os.Getenv("NEXARCH_DMS_WORKER_RETENTION_BASE_URL"), os.Getenv("NEXARCH_DMS_WORKER_TENANT_SLUG"))
|
|
startJobWorkers(ctx, pool, queue, tracker)
|
|
} else {
|
|
log.Print("dms-worker: NEXARCH_DMS_WORKER_TENANT_DSN nicht gesetzt, DOC-16-Requeue-Verarbeitung deaktiviert")
|
|
}
|
|
|
|
ticker := time.NewTicker(30 * time.Second)
|
|
defer ticker.Stop()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
log.Println("dms-worker beendet")
|
|
return
|
|
case <-ticker.C:
|
|
if queue != nil {
|
|
if n, err := queue.QueueLength(ctx); err == nil {
|
|
metricsReg.SetQueueLength(n)
|
|
}
|
|
processRetentionRegisterJobs(ctx, queue, client, tracker)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// processRetentionRegisterJobs holt und verarbeitet ALLE aktuell
|
|
// abholbaren doc16_retention_register-Jobs (Akzeptanzkriterium 3: Requeue
|
|
// statt Absturz/Verwerfen — Fail() haengt selbst nie ab, sondern setzt
|
|
// pending mit Backoff oder dead_letter, siehe FDN-04).
|
|
func processRetentionRegisterJobs(ctx context.Context, queue *jobqueue.Queue, client *retentionclient.Client, tracker *errortracking.Tracker) {
|
|
for {
|
|
job, err := queue.Dequeue(ctx, "dms-worker", []string{retentiondestroy.JobTypeRegister})
|
|
if err != nil {
|
|
if !errors.Is(err, jobqueue.ErrNoJobAvailable) {
|
|
log.Printf("dms-worker: dequeue fehlgeschlagen: %v", err)
|
|
}
|
|
return
|
|
}
|
|
// FDN-08 Akzeptanzkriterium 1: Job-ID als Korrelations-ID, analog zu
|
|
// X-Request-Id bei HTTP-Anfragen (siehe internal/jobqueue.Worker).
|
|
jobCtx := logging.WithRequestID(ctx, job.ID)
|
|
if err := retentiondestroy.ProcessRegisterJob(jobCtx, client, job.Payload); err != nil {
|
|
log.Printf("dms-worker: retention-registrierung erneut fehlgeschlagen (job %s): %v", job.ID, err)
|
|
tracker.Capture(jobCtx, "worker.retention-register", err)
|
|
if failErr := queue.Fail(ctx, job.ID, err); failErr != nil {
|
|
log.Printf("dms-worker: job als fehlgeschlagen markieren: %v", failErr)
|
|
}
|
|
continue
|
|
}
|
|
if err := queue.Complete(ctx, job.ID); err != nil {
|
|
log.Printf("dms-worker: job abschliessen: %v", err)
|
|
} else {
|
|
log.Printf("dms-worker: retention-registrierung erfolgreich nachgeholt (job %s)", job.ID)
|
|
}
|
|
}
|
|
}
|
|
|
|
// startJobWorkers verdrahtet alle Fachjobs (FDN-17). Fehlt die Storage-/Schluessel-
|
|
// Konfiguration, werden sie nicht gestartet und das wird gemeldet.
|
|
func startJobWorkers(ctx context.Context, pool *pgxpool.Pool, queue *jobqueue.Queue, tracker *errortracking.Tracker) {
|
|
storageDir := os.Getenv("NEXARCH_DMS_WORKER_STORAGE_DIR")
|
|
tenantSlug := os.Getenv("NEXARCH_DMS_WORKER_TENANT_SLUG")
|
|
kekURL := os.Getenv("NEXARCH_DMS_WORKER_KEK_ENDPOINT_URL")
|
|
if storageDir == "" || tenantSlug == "" || kekURL == "" {
|
|
log.Print("dms-worker: NEXARCH_DMS_WORKER_STORAGE_DIR/_TENANT_SLUG/_KEK_ENDPOINT_URL unvollstaendig, Fachjobs (Vorschau, Konvertierung, Virenscan, Export, Texterkennung) werden NICHT verarbeitet")
|
|
return
|
|
}
|
|
driver := storage.NewLocalDriver(storageDir, []byte(os.Getenv("NEXARCH_DMS_WORKER_STORAGE_SIGNING_SECRET")), os.Getenv("NEXARCH_DMS_WORKER_STORAGE_PUBLIC_BASE_URL"), signedURLMaxTTL(os.Getenv("NEXARCH_DMS_WORKER_SIGNED_URL_MAX_TTL")))
|
|
usage := storage.NewHTTPUsageReporter(os.Getenv("NEXARCH_DMS_WORKER_USAGE_ENDPOINT_URL"), os.Getenv("NEXARCH_DMS_WORKER_USAGE_CLIENT_ID"), os.Getenv("NEXARCH_DMS_WORKER_USAGE_CLIENT_SECRET"), nil)
|
|
cryptoSvc := dmscrypto.NewService(dmscrypto.NewHTTPKEKProvider(kekURL, os.Getenv("NEXARCH_DMS_WORKER_KEK_CLIENT_ID"), os.Getenv("NEXARCH_DMS_WORKER_KEK_CLIENT_SECRET"), nil))
|
|
|
|
deps := workerruntime.Deps{Pool: pool, Storage: storage.NewService(driver, usage, tenantSlug), Crypto: cryptoSvc, TenantSlug: tenantSlug, Queue: queue}
|
|
if addr := os.Getenv("NEXARCH_DMS_WORKER_CLAMD_ADDRESS"); addr != "" {
|
|
network := os.Getenv("NEXARCH_DMS_WORKER_CLAMD_NETWORK")
|
|
if network == "" {
|
|
network = "unix"
|
|
}
|
|
deps.Scanner = &malwarescan.ClamdScanner{Network: network, Address: addr}
|
|
}
|
|
if addr := os.Getenv("NEXARCH_DMS_WORKER_MANTICORE_ADDR"); addr != "" {
|
|
name := os.Getenv("NEXARCH_DMS_WORKER_SEARCH_INDEX_NAME")
|
|
if name == "" {
|
|
name = searchindex.DefaultTableName
|
|
}
|
|
if err := searchindex.ValidateTableName(name); err != nil {
|
|
log.Fatalf("dms-worker: NEXARCH_DMS_WORKER_SEARCH_INDEX_NAME: %v", err)
|
|
}
|
|
idx, err := searchindex.OpenNamed(addr, name)
|
|
if err == nil {
|
|
err = searchindex.EnsureSchema(ctx, idx)
|
|
}
|
|
if err != nil {
|
|
log.Printf("dms-worker: suchindex nicht verfuegbar, searchindex.sync wird NICHT verarbeitet: %v", err)
|
|
} else {
|
|
deps.SearchIndex = idx
|
|
}
|
|
}
|
|
if v, err := strconv.ParseInt(os.Getenv("NEXARCH_DMS_WORKER_PLAINTEXT_MAX_BYTES"), 10, 64); err == nil && v > 0 {
|
|
deps.PlaintextMaxBytes = v
|
|
}
|
|
if v, err := strconv.Atoi(os.Getenv("NEXARCH_DMS_WORKER_PLAINTEXT_MAX_CHARS")); err == nil && v > 0 {
|
|
deps.PlaintextMaxChars = v
|
|
}
|
|
var mailSvc *upload.Service
|
|
if stagingDir := os.Getenv("NEXARCH_DMS_WORKER_MAILIMPORT_STAGING_DIR"); stagingDir != "" {
|
|
maxBytes := int64(mailimport.DefaultMaxMessageBytes)
|
|
if v, err := strconv.ParseInt(os.Getenv("NEXARCH_DMS_WORKER_MAILIMPORT_MAX_MESSAGE_BYTES"), 10, 64); err == nil && v > 0 {
|
|
maxBytes = v
|
|
}
|
|
mailSvc = upload.NewService(pool, upload.NewSessionStore(pool), upload.NewStaging(stagingDir), deps.Storage, cryptoSvc, upload.NewValidator(maxBytes, nil), tenantSlug)
|
|
dialer := mailimport.Dialer{}
|
|
if caFile := os.Getenv("NEXARCH_DMS_WORKER_MAILIMPORT_CA_FILE"); caFile != "" {
|
|
pem, err := os.ReadFile(caFile)
|
|
roots := x509.NewCertPool()
|
|
if err != nil || !roots.AppendCertsFromPEM(pem) {
|
|
log.Printf("dms-worker: NEXARCH_DMS_WORKER_MAILIMPORT_CA_FILE nicht nutzbar, System-Zertifikate werden verwendet: %v", err)
|
|
} else {
|
|
dialer.RootCAs = roots
|
|
}
|
|
}
|
|
deps.MailImport = &mailimport.Importer{
|
|
Pool: pool,
|
|
Store: &mailimport.Store{Pool: pool, Crypto: cryptoSvc, TenantSlug: tenantSlug},
|
|
Upload: mailSvc,
|
|
Dialer: dialer,
|
|
MaxMessageBytes: maxBytes,
|
|
}
|
|
}
|
|
if addr := os.Getenv("NEXARCH_DMS_WORKER_KOSIT_ADDR"); addr != "" {
|
|
timeout := 30 * time.Second
|
|
if v, err := strconv.Atoi(os.Getenv("NEXARCH_DMS_WORKER_KOSIT_TIMEOUT_SECONDS")); err == nil && v > 0 {
|
|
timeout = time.Duration(v) * time.Second
|
|
}
|
|
val, err := einvoice.NewValidator(addr, os.Getenv("NEXARCH_DMS_WORKER_KOSIT_RULESET"), timeout)
|
|
if err != nil {
|
|
log.Printf("dms-worker: e-rechnungs-validator ungueltig konfiguriert, Ergebnis bleibt not_checked: %v", err)
|
|
} else {
|
|
deps.EInvoiceValidator = val
|
|
go func() {
|
|
rulesets, err := val.CurrentRulesets(ctx)
|
|
if err != nil {
|
|
log.Printf("dms-worker: regelwerk des validators nicht ermittelbar, nutze Angabe %q: %v", val.Ruleset, err)
|
|
rulesets = []string{val.Ruleset}
|
|
}
|
|
n, err := einvoice.EnqueueRecheck(ctx, pool, queue, rulesets...)
|
|
if err != nil {
|
|
log.Printf("dms-worker: neupruefung von e-rechnungen nicht eingereiht: %v", err)
|
|
} else if n > 0 {
|
|
log.Printf("dms-worker: %d e-rechnungen zur neupruefung gegen regelwerk %v eingereiht", n, rulesets)
|
|
}
|
|
}()
|
|
}
|
|
} else {
|
|
log.Print("dms-worker: kein e-rechnungs-validator (NEXARCH_DMS_WORKER_KOSIT_ADDR) konfiguriert, Konformitaet bleibt not_checked")
|
|
}
|
|
rt := workerruntime.New(deps)
|
|
rt.Middleware = func(next jobqueue.Handler) jobqueue.Handler {
|
|
return func(ctx context.Context, job *jobqueue.Job) error {
|
|
err := next(ctx, job)
|
|
if err != nil {
|
|
tracker.Capture(ctx, "worker."+job.JobType, err)
|
|
}
|
|
return err
|
|
}
|
|
}
|
|
for _, n := range rt.Notices {
|
|
log.Print("dms-worker: ", n)
|
|
}
|
|
concurrency := 2
|
|
if v, err := strconv.Atoi(os.Getenv("NEXARCH_DMS_WORKER_CONCURRENCY")); err == nil && v > 0 {
|
|
concurrency = v
|
|
}
|
|
log.Printf("dms-worker: verarbeitet jobtypen %v mit %d parallelen workern", rt.JobTypes(), concurrency)
|
|
if mailSvc != nil {
|
|
workerruntime.AttachEnqueuers(mailSvc, pool, queue, rt.JobTypes())
|
|
go mailimport.RunScheduler(ctx, pool, queue, 15*time.Second)
|
|
}
|
|
go rt.Run(ctx, concurrency, workerruntime.HostName(), time.Second)
|
|
}
|
|
|
|
func signedURLMaxTTL(raw string) time.Duration {
|
|
ttl, err := storage.ParseMaxSignedURLTTL(raw)
|
|
if err != nil {
|
|
log.Fatalf("dms-worker: %v", err)
|
|
}
|
|
return ttl
|
|
}
|