nexarch/workflow/internal/moduleadapter/handler.go

package moduleadapter

import (
	"context"
	"encoding/json"
	"errors"
	"log"
	"net/http"
	"time"

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

	"gitea.perlbach24.de/scripte/nexarch/common/svcauth"
)

// TenantVerifier prueft einen vom aufrufenden Modul mitgelieferten
// Tenant-Kontext unabhaengig gegen die Core-Tenant-Registry (TEN-01) -
// ENG-06: Workflow vertraut dabei nicht der Angabe von DMS/Mail selbst
// (Muster aus archive/internal/moduleadapter, RET-12).
type TenantVerifier interface {
	VerifyTenant(ctx context.Context, tenantSlug string) (bool, error)
}

// Server bindet die Registrierungs-API an einen Postgres-Pool und
// verlangt ein Service-Token bei jedem Aufruf (fail closed: ein leeres
// oder falsches Token wird immer abgewiesen, Lehre aus RET-14/F-08).
// ownTenantSlug bindet den Server an genau einen Mandanten (ENG-09
// Akzeptanzkriterium 1/3): Ereignisse mit abweichendem Mandanten werden
// abgelehnt statt verarbeitet.
type Server struct {
	pool          *pgxpool.Pool
	serviceToken  string
	ownTenantSlug string
	tenants       TenantVerifier
}

func NewServer(pool *pgxpool.Pool, serviceToken, ownTenantSlug string, tenants TenantVerifier) *Server {
	return &Server{pool: pool, serviceToken: serviceToken, ownTenantSlug: ownTenantSlug, tenants: tenants}
}

func (s *Server) RegisterRoutes(mux *http.ServeMux) {
	mux.HandleFunc("POST /register", s.requireAuth(s.handleRegister))
	mux.HandleFunc("POST /rollback", s.requireAuth(s.handleRollback))
	mux.HandleFunc("POST /deregister", s.requireAuth(s.handleDeregister))
	mux.HandleFunc("GET /registrations", s.requireAuth(s.handleList))
	mux.HandleFunc("POST /events", s.requireAuth(s.handlePublishEvent))
}

func (s *Server) requireAuth(next http.HandlerFunc) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		got := r.Header.Get("X-Service-Token")
		if !svcauth.Equal(got, s.serviceToken) {
			writeError(w, http.StatusUnauthorized, "service-token fehlt oder ist ungueltig")
			return
		}
		next(w, r)
	}
}

// tenantValid prueft den mitgelieferten Tenant-Kontext unabhaengig von
// der Angabe des registrierenden Moduls (ENG-06 Akzeptanzkriterium 1):
// er muss zu dieser Server-Instanz passen UND von der Core-Tenant-
// Registry als echter, gefuehrter Tenant bestaetigt werden. Ein
// fehlgeschlagener Core-Aufruf gilt als ungueltig (fail closed), kein
// stillschweigendes Durchwinken.
func (s *Server) tenantValid(ctx context.Context, tenantSlug string) bool {
	if tenantSlug == "" || tenantSlug != s.ownTenantSlug {
		return false
	}
	if s.tenants == nil {
		return true
	}
	ok, err := s.tenants.VerifyTenant(ctx, tenantSlug)
	if err != nil {
		log.Printf("moduleadapter: tenant-registry-abgleich fehlgeschlagen tenant=%q: %v", tenantSlug, err)
		return false
	}
	return ok
}

type registerRequest struct {
	ModuleName string   `json:"module_name"`
	ObjectType string   `json:"object_type"`
	Triggers   []string `json:"triggers"`
	TenantSlug string   `json:"tenant_slug"`
}

func (s *Server) handleRegister(w http.ResponseWriter, r *http.Request) {
	var req registerRequest
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		writeError(w, http.StatusBadRequest, "ungueltiger anfrage-koerper")
		return
	}
	if !s.tenantValid(r.Context(), req.TenantSlug) {
		log.Printf("moduleadapter: registrierung abgelehnt, unbekannter tenant-kontext modul=%s objekttyp=%s tenant=%q", req.ModuleName, req.ObjectType, req.TenantSlug)
		writeError(w, http.StatusForbidden, "unbekannter tenant-kontext")
		return
	}
	reg, err := Register(r.Context(), s.pool, req.ModuleName, req.ObjectType, req.Triggers)
	if errors.Is(err, ErrDuplicateTriggers) {
		writeError(w, http.StatusConflict, "registrierung unveraendert, keine neue version noetig")
		return
	}
	if err != nil {
		writeError(w, http.StatusBadRequest, err.Error())
		return
	}
	writeJSON(w, http.StatusOK, reg)
}

type rollbackRequest struct {
	ModuleName string `json:"module_name"`
	ObjectType string `json:"object_type"`
	ToVersion  int    `json:"to_version"`
}

func (s *Server) handleRollback(w http.ResponseWriter, r *http.Request) {
	var req rollbackRequest
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		writeError(w, http.StatusBadRequest, "ungueltiger anfrage-koerper")
		return
	}
	reg, err := Rollback(r.Context(), s.pool, req.ModuleName, req.ObjectType, req.ToVersion)
	if errors.Is(err, ErrNotRegistered) {
		writeError(w, http.StatusNotFound, "objekttyp ist nicht registriert")
		return
	}
	if err != nil {
		writeError(w, http.StatusBadRequest, err.Error())
		return
	}
	writeJSON(w, http.StatusOK, reg)
}

type deregisterRequest struct {
	ModuleName string `json:"module_name"`
	ObjectType string `json:"object_type"`
}

func (s *Server) handleDeregister(w http.ResponseWriter, r *http.Request) {
	var req deregisterRequest
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		writeError(w, http.StatusBadRequest, "ungueltiger anfrage-koerper")
		return
	}
	err := Deregister(r.Context(), s.pool, req.ModuleName, req.ObjectType)
	if errors.Is(err, ErrNotRegistered) {
		writeError(w, http.StatusNotFound, "objekttyp ist nicht registriert")
		return
	}
	if err != nil {
		writeError(w, http.StatusInternalServerError, "deregistrierung fehlgeschlagen")
		return
	}
	w.WriteHeader(http.StatusNoContent)
}

func (s *Server) handleList(w http.ResponseWriter, r *http.Request) {
	regs, err := ListRegistrations(r.Context(), s.pool)
	if err != nil {
		writeError(w, http.StatusInternalServerError, "registrierungen konnten nicht gelesen werden")
		return
	}
	writeJSON(w, http.StatusOK, regs)
}

type publishEventRequest struct {
	ModuleName string          `json:"module_name"`
	ObjectType string          `json:"object_type"`
	Trigger    string          `json:"trigger"`
	ObjectID   string          `json:"object_id"`
	TenantSlug string          `json:"tenant_slug"`
	DedupeKey  string          `json:"dedupe_key"`
	Version    int             `json:"version"`
	OccurredAt time.Time       `json:"occurred_at"`
	Payload    json.RawMessage `json:"payload"`
}

func (s *Server) handlePublishEvent(w http.ResponseWriter, r *http.Request) {
	var req publishEventRequest
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		writeError(w, http.StatusBadRequest, "ungueltiger anfrage-koerper")
		return
	}
	accepted, err := PublishEvent(r.Context(), s.pool, s.ownTenantSlug, Event{
		ModuleName: req.ModuleName,
		ObjectType: req.ObjectType,
		Trigger:    req.Trigger,
		ObjectID:   req.ObjectID,
		TenantSlug: req.TenantSlug,
		DedupeKey:  req.DedupeKey,
		Version:    req.Version,
		OccurredAt: req.OccurredAt,
		Payload:    req.Payload,
	})
	if err != nil {
		writeError(w, http.StatusBadRequest, err.Error())
		return
	}
	if !accepted {
		writeError(w, http.StatusForbidden, "ereignis fuer fremden mandanten abgelehnt")
		return
	}
	w.WriteHeader(http.StatusAccepted)
}

func writeJSON(w http.ResponseWriter, status int, v any) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(status)
	_ = json.NewEncoder(w).Encode(v)
}

func writeError(w http.ResponseWriter, status int, message string) {
	writeJSON(w, status, map[string]string{"error": message})
}