nexarch/workflow/cmd/workflow-app/main.go

// Command workflow-app ist der gebuendelte Diensteinstiegspunkt fuer das
// NEXARCH-Workflow-Modul (ENG-11): bindet alle bereits fertigen
// HTTP-API-Pakete an einen einzigen net/http.ServeMux, analog
// connect/cmd/connect-app. Reines Wiring, keine neue Fachlogik.
package main

import (
	"context"
	"fmt"
	"log"
	"net/http"
	"os"
	"os/signal"
	"syscall"
	"time"

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

	"gitea.perlbach24.de/scripte/nexarch/common/bearerauth"
	"gitea.perlbach24.de/scripte/nexarch/common/moduleauth"
	"gitea.perlbach24.de/scripte/nexarch/common/policyclient"
	"gitea.perlbach24.de/scripte/nexarch/common/tenantregistryclient"
	"gitea.perlbach24.de/scripte/nexarch/workflow/internal/approval"
	"gitea.perlbach24.de/scripte/nexarch/workflow/internal/approvalapi"
	"gitea.perlbach24.de/scripte/nexarch/workflow/internal/backchannel"
	"gitea.perlbach24.de/scripte/nexarch/workflow/internal/backchannelapi"
	"gitea.perlbach24.de/scripte/nexarch/workflow/internal/caseapi"
	"gitea.perlbach24.de/scripte/nexarch/workflow/internal/caseobj"
	"gitea.perlbach24.de/scripte/nexarch/workflow/internal/caseoverview"
	"gitea.perlbach24.de/scripte/nexarch/workflow/internal/caseoverviewapi"
	"gitea.perlbach24.de/scripte/nexarch/workflow/internal/escalation"
	"gitea.perlbach24.de/scripte/nexarch/workflow/internal/escalationnotify"
	"gitea.perlbach24.de/scripte/nexarch/workflow/internal/moduleadapter"
	"gitea.perlbach24.de/scripte/nexarch/workflow/internal/notifyclient"
	"gitea.perlbach24.de/scripte/nexarch/workflow/internal/reminder"
	"gitea.perlbach24.de/scripte/nexarch/workflow/internal/routingrule"
	"gitea.perlbach24.de/scripte/nexarch/workflow/internal/routingruleapi"
	"gitea.perlbach24.de/scripte/nexarch/workflow/internal/routingsuggestclient"
	"gitea.perlbach24.de/scripte/nexarch/workflow/internal/task"
	"gitea.perlbach24.de/scripte/nexarch/workflow/internal/taskapi"
)

func requireEnv(name string) string {
	v := os.Getenv(name)
	if v == "" {
		log.Fatalf("workflow-app: %s muss gesetzt sein", name)
	}
	return v
}

func main() {
	dsn := requireEnv("NEXARCH_WORKFLOW_APP_TENANT_DSN")
	serviceToken := requireEnv("NEXARCH_WORKFLOW_APP_SERVICE_TOKEN")
	tenantSlug := requireEnv("NEXARCH_WORKFLOW_APP_TENANT_SLUG")
	tenantRegistryURL := requireEnv("NEXARCH_TENANT_REGISTRY_URL")
	tenantRegistryToken := os.Getenv("NEXARCH_TENANT_REGISTRY_SERVICE_TOKEN")

	addr := os.Getenv("NEXARCH_WORKFLOW_APP_LISTEN_ADDR")
	if addr == "" {
		addr = "127.0.0.1:9600"
	}

	ctx := context.Background()
	pool, err := pgxpool.New(ctx, dsn)
	if err != nil {
		log.Fatalf("workflow-app: datenbankverbindung: %v", err)
	}
	defer pool.Close()

	verifier, allowHeaderAuth := buildIdentity(tenantSlug)

	taskStore := task.NewStore(pool)
	mux, err := buildMux(pool, taskStore, serviceToken, tenantSlug, tenantRegistryURL, tenantRegistryToken, buildPolicyClient())
	if err != nil {
		log.Fatalf("workflow-app: %v", err)
	}

	srv := &http.Server{
		Addr:              addr,
		Handler:           bearerauth.Middleware(verifier, allowHeaderAuth, mux),
		ReadHeaderTimeout: 5 * time.Second,
		ReadTimeout:       15 * time.Second,
		WriteTimeout:      15 * time.Second,
		IdleTimeout:       60 * time.Second,
	}

	go func() {
		log.Printf("workflow-app: listening on %s", addr)
		if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
			log.Fatalf("workflow-app: %v", err)
		}
	}()

	stopEscalation := startEscalationDispatch(ctx, pool, taskStore, tenantSlug)
	defer stopEscalation()

	stop := make(chan os.Signal, 1)
	signal.Notify(stop, syscall.SIGINT, syscall.SIGTERM)
	<-stop

	shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	_ = srv.Shutdown(shutdownCtx)
}

// buildIdentity richtet die Token-Pruefung ein. Ohne NEXARCH_WORKFLOW_APP_JWKS_URL
// muss der Uebergangsschalter NEXARCH_WORKFLOW_APP_ALLOW_HEADER_AUTH=1 gesetzt
// sein; sonst startet der Dienst nicht.
func buildIdentity(tenantSlug string) (*moduleauth.Verifier, bool) {
	allowHeader := os.Getenv("NEXARCH_WORKFLOW_APP_ALLOW_HEADER_AUTH") == "1"
	jwksURL := os.Getenv("NEXARCH_WORKFLOW_APP_JWKS_URL")
	if allowHeader {
		log.Print("workflow-app: WARNUNG: X-User-Id/X-Tenant-Slug ohne Token zugelassen (NEXARCH_WORKFLOW_APP_ALLOW_HEADER_AUTH=1), nicht authentifiziert")
	}
	if jwksURL == "" {
		if !allowHeader {
			log.Fatal("workflow-app: NEXARCH_WORKFLOW_APP_JWKS_URL fehlt (oder NEXARCH_WORKFLOW_APP_ALLOW_HEADER_AUTH=1 als Uebergang setzen)")
		}
		return nil, allowHeader
	}
	audience := os.Getenv("NEXARCH_WORKFLOW_APP_JWT_AUDIENCE")
	if audience == "" {
		audience = "workflow"
	}
	issuer := os.Getenv("NEXARCH_WORKFLOW_APP_JWT_ISSUER")
	if issuer == "" {
		issuer = "nexarch-core"
	}
	enforce := os.Getenv("NEXARCH_WORKFLOW_APP_JWT_AUDIENCE_ENFORCE") != "false"
	v := moduleauth.NewVerifier(5*time.Minute, tenantSlug, moduleauth.HTTPKeyFetcher(jwksURL, nil)).Expect(audience, issuer, enforce)
	return v, allowHeader
}

// buildMux verdrahtet alle bestehenden Workflow-API-Pakete auf einen
// gemeinsamen ServeMux (Akzeptanzkriterium 1) -- ausgelagert aus main(),
// damit ein Test denselben Aufbau real gegen eine Datenbank pruefen kann,
// ohne main() zu duplizieren.
func buildMux(pool *pgxpool.Pool, taskStore *task.Store, serviceToken, tenantSlug, tenantRegistryURL, tenantRegistryToken string, policy *policyclient.Client) (*http.ServeMux, error) {
	engine, err := caseobj.NewEngineWithCase(pool, approval.ProcessDefinition)
	if err != nil {
		return nil, fmt.Errorf("prozess-engine aufbauen: %w", err)
	}
	caseSvc := caseobj.NewService(pool, engine)
	approvalSvc := approval.NewService(pool, engine)
	reminderStore := reminder.NewStore(pool, taskStore)
	caseoverviewSvc := caseoverview.NewService(caseSvc, reminderStore)
	tenantVerifier := tenantregistryclient.New(tenantRegistryURL, tenantRegistryToken)
	backchannelSvc := backchannel.NewService(caseSvc, reminderStore, tenantSlug, tenantVerifier)
	routingruleStore := routingrule.NewStore(pool)

	mux := http.NewServeMux()
	mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) })
	caseapi.NewServer(caseSvc).RegisterRoutes(mux)
	approvalapi.NewServer(approvalSvc).RegisterRoutes(mux)
	caseoverviewapi.NewServer(caseoverviewSvc).RegisterRoutes(mux)
	backchannelapi.NewServer(backchannelSvc).RegisterRoutes(mux)
	routingruleapi.NewServer(routingruleStore, buildRoutingSuggester()).RegisterRoutes(mux)
	moduleadapter.NewServer(pool, serviceToken, tenantSlug, tenantVerifier).RegisterRoutes(mux)
	taskapi.NewServer(taskStore, tenantSlug).WithPolicy(policy).RegisterRoutes(mux)
	return mux, nil
}

// buildPolicyClient liefert nil nur mit NEXARCH_WORKFLOW_APP_POLICY_DISABLED=1;
// ohne URL und ohne diesen Schalter startet der Dienst nicht.
func buildPolicyClient() *policyclient.Client {
	policyURL := os.Getenv("NEXARCH_WORKFLOW_APP_POLICY_URL")
	if policyURL == "" {
		if os.Getenv("NEXARCH_WORKFLOW_APP_POLICY_DISABLED") != "1" {
			log.Fatal("workflow-app: NEXARCH_WORKFLOW_APP_POLICY_URL fehlt (oder NEXARCH_WORKFLOW_APP_POLICY_DISABLED=1 als Uebergang setzen)")
		}
		log.Print("workflow-app: WARNUNG: Rechtepruefung workflow.task.read/write deaktiviert (NEXARCH_WORKFLOW_APP_POLICY_DISABLED=1)")
		return nil
	}
	return policyclient.New(policyURL, requireEnv("NEXARCH_WORKFLOW_APP_POLICY_SERVICE_TOKEN"))
}

// buildRoutingSuggester verdrahtet RTE-05s Kopplung an AI WFS-02.
// NEXARCH_WORKFLOW_APP_AI_ROUTING_SUGGEST_BASE_URL bewusst optional --
// fehlt sie, ist das AI-Modul nicht aktiviert und routingruleapi liefert
// keinen KI-Vorschlag (AK2, fail open statt Blockade).
// Rueckgabetyp bewusst das Interface (nicht *routingsuggestclient.Client):
// ein typisierter Nil-Pointer als Interface-Wert waere != nil, das wuerde
// den fail-open-Check in routingruleapi (s.suggester == nil) unterlaufen.
func buildRoutingSuggester() routingruleapi.Suggester {
	baseURL := os.Getenv("NEXARCH_WORKFLOW_APP_AI_ROUTING_SUGGEST_BASE_URL")
	if baseURL == "" {
		log.Print("workflow-app: NEXARCH_WORKFLOW_APP_AI_ROUTING_SUGGEST_BASE_URL nicht gesetzt, KI-Routing-Vorschlaege (RTE-05) sind deaktiviert")
		return nil
	}
	serviceToken := requireEnv("NEXARCH_WORKFLOW_APP_AI_ROUTING_SUGGEST_SERVICE_TOKEN")
	return routingsuggestclient.New(baseURL, serviceToken, &http.Client{Timeout: 3 * time.Second})
}

// startEscalationDispatch startet den periodischen TSK-05-Versand
// (escalationnotify.RunAt) als Ticker-Goroutine, da escalation.ProcessDue
// sonst nie in Produktion aufgerufen wird -- es gibt hier anders als bei
// RET-07/BAK-08 keinen systemd-Timer als externen Aufrufer. Die Dispatch-
// Konfiguration ist bewusst optional: fehlt sie, startet der Dienst
// trotzdem (Fristen/Eskalationen sind kein Kernbetriebsrisiko wie eine
// fehlende DSN), nur ohne Benachrichtigungsversand.
func startEscalationDispatch(ctx context.Context, pool *pgxpool.Pool, taskStore *task.Store, tenantSlug string) func() {
	notifyBaseURL := os.Getenv("NEXARCH_WORKFLOW_APP_NOTIFY_BASE_URL")
	if notifyBaseURL == "" {
		log.Print("workflow-app: NEXARCH_WORKFLOW_APP_NOTIFY_BASE_URL nicht gesetzt, Eskalations-Benachrichtigung (TSK-05) laeuft nicht")
		return func() {}
	}
	notifyServiceToken := requireEnv("NEXARCH_WORKFLOW_APP_NOTIFY_SERVICE_TOKEN")
	interval := 5 * time.Minute
	if raw := os.Getenv("NEXARCH_WORKFLOW_APP_ESCALATION_INTERVAL"); raw != "" {
		parsed, err := time.ParseDuration(raw)
		if err != nil {
			log.Fatalf("workflow-app: NEXARCH_WORKFLOW_APP_ESCALATION_INTERVAL: %v", err)
		}
		interval = parsed
	}

	escalationStore := escalation.NewStore(pool)
	notifyClient := notifyclient.New(notifyBaseURL, notifyServiceToken)

	loopCtx, cancel := context.WithCancel(ctx)
	ticker := time.NewTicker(interval)
	go func() {
		defer ticker.Stop()
		for {
			select {
			case <-loopCtx.Done():
				return
			case now := <-ticker.C:
				dispatched, err := escalationnotify.RunAt(loopCtx, escalationStore, taskStore, notifyClient, tenantSlug, nil, now)
				if err != nil {
					log.Printf("workflow-app: eskalations-durchlauf fehlgeschlagen: %v", err)
					continue
				}
				if dispatched > 0 {
					log.Printf("workflow-app: %d eskalation(en) benachrichtigt", dispatched)
				}
			}
		}
	}()
	log.Printf("workflow-app: eskalations-benachrichtigung aktiv, intervall %s", interval)
	return cancel
}