nexarch/workflow/internal/moduleadapter/moduleadapter.go

// Package moduleadapter implementiert ENG-02: die Schnittstelle, ueber
// die DMS und Mail eigene Objekttypen und Ausloeser-Ereignisse bei der
// Workflow-Engine registrieren, ohne dass die Engine modulspezifisches
// Wissen braucht (Muster wie archive/internal/moduleadapter aus RET-05,
// hier von Anfang an mit Dienstauthentifizierung statt sie wie dort
// nachtraeglich einzubauen).
package moduleadapter

import (
	"context"
	"errors"
	"fmt"

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

// ErrDuplicateTriggers wird zurueckgegeben, wenn eine Registrierung mit
// unveraendertem Trigger-Satz wiederholt wird (Pruefung 2: doppelte
// Registrierung wird abgewiesen).
var ErrDuplicateTriggers = errors.New("moduleadapter: unveraenderte registrierung, keine neue version noetig")

// ErrNotRegistered wird zurueckgegeben, wenn fuer einen Objekttyp keine
// Registrierung existiert (Akzeptanzkriterium 3: klare Fehlermeldung
// statt Absturz).
var ErrNotRegistered = errors.New("moduleadapter: objekttyp ist nicht registriert")

// Registration ist der aktuelle Stand eines registrierten Objekttyps.
type Registration struct {
	ModuleName    string
	ObjectType    string
	ActiveVersion int
	Triggers      []string
	Deregistered  bool
}

// Register registriert einen Objekttyp mit seinen Triggern
// (Akzeptanzkriterium 1: ohne Codeaenderung an der Engine). Eine erneute
// Registrierung mit geaendertem Trigger-Satz legt eine neue Version an
// (Akzeptanzkriterium 2: versionierbar); mit unveraendertem Satz liefert
// sie ErrDuplicateTriggers (Pruefung 2). Ein zuvor deregistrierter
// Objekttyp wird durch erneutes Register wieder aktiv.
func Register(ctx context.Context, pool *pgxpool.Pool, moduleName, objectType string, triggers []string) (Registration, error) {
	if moduleName == "" || objectType == "" {
		return Registration{}, fmt.Errorf("moduleadapter: modulname und objekttyp sind pflicht")
	}
	if triggers == nil {
		triggers = []string{}
	}

	tx, err := pool.Begin(ctx)
	if err != nil {
		return Registration{}, fmt.Errorf("moduleadapter: transaktion starten: %w", err)
	}
	defer func() { _ = tx.Rollback(ctx) }()

	var regID string
	var activeVersion int
	err = tx.QueryRow(ctx, `
		SELECT id, active_version FROM module_registrations
		WHERE module_name = $1 AND object_type = $2
	`, moduleName, objectType).Scan(&regID, &activeVersion)
	switch {
	case errors.Is(err, pgx.ErrNoRows):
		if err := tx.QueryRow(ctx, `
			INSERT INTO module_registrations (module_name, object_type, active_version)
			VALUES ($1, $2, 0) RETURNING id
		`, moduleName, objectType).Scan(&regID); err != nil {
			return Registration{}, fmt.Errorf("moduleadapter: registrierung anlegen: %w", err)
		}
	case err != nil:
		return Registration{}, fmt.Errorf("moduleadapter: bestehende registrierung lesen: %w", err)
	default:
		if activeVersion > 0 {
			var existing []string
			if err := tx.QueryRow(ctx, `
				SELECT triggers FROM module_registration_versions
				WHERE registration_id = $1 AND version = $2
			`, regID, activeVersion).Scan(&existing); err != nil {
				return Registration{}, fmt.Errorf("moduleadapter: aktive version lesen: %w", err)
			}
			if sameTriggers(existing, triggers) {
				return Registration{}, ErrDuplicateTriggers
			}
		}
	}

	newVersion := activeVersion + 1
	if _, err := tx.Exec(ctx, `
		INSERT INTO module_registration_versions (registration_id, version, triggers)
		VALUES ($1, $2, $3)
	`, regID, newVersion, triggers); err != nil {
		return Registration{}, fmt.Errorf("moduleadapter: version anlegen: %w", err)
	}
	if _, err := tx.Exec(ctx, `
		UPDATE module_registrations SET active_version = $1, deregistered = false, updated_at = now()
		WHERE id = $2
	`, newVersion, regID); err != nil {
		return Registration{}, fmt.Errorf("moduleadapter: aktive version setzen: %w", err)
	}
	if err := tx.Commit(ctx); err != nil {
		return Registration{}, fmt.Errorf("moduleadapter: transaktion abschliessen: %w", err)
	}

	return Registration{ModuleName: moduleName, ObjectType: objectType, ActiveVersion: newVersion, Triggers: triggers}, nil
}

// Rollback setzt eine Registrierung auf eine fruehere Version zurueck
// (Akzeptanzkriterium 2: rueckrollbar). Die Zielversion muss bereits
// existieren; ein Rollback erzeugt keine neue Versionsnummer, sondern
// setzt active_version direkt.
func Rollback(ctx context.Context, pool *pgxpool.Pool, moduleName, objectType string, toVersion int) (Registration, error) {
	var regID string
	err := pool.QueryRow(ctx, `
		SELECT id FROM module_registrations WHERE module_name = $1 AND object_type = $2
	`, moduleName, objectType).Scan(&regID)
	if errors.Is(err, pgx.ErrNoRows) {
		return Registration{}, ErrNotRegistered
	}
	if err != nil {
		return Registration{}, fmt.Errorf("moduleadapter: registrierung lesen: %w", err)
	}

	var triggers []string
	err = pool.QueryRow(ctx, `
		SELECT triggers FROM module_registration_versions WHERE registration_id = $1 AND version = $2
	`, regID, toVersion).Scan(&triggers)
	if errors.Is(err, pgx.ErrNoRows) {
		return Registration{}, fmt.Errorf("moduleadapter: version %d existiert nicht", toVersion)
	}
	if err != nil {
		return Registration{}, fmt.Errorf("moduleadapter: zielversion lesen: %w", err)
	}

	if _, err := pool.Exec(ctx, `
		UPDATE module_registrations SET active_version = $1, deregistered = false, updated_at = now() WHERE id = $2
	`, toVersion, regID); err != nil {
		return Registration{}, fmt.Errorf("moduleadapter: rollback schreiben: %w", err)
	}
	return Registration{ModuleName: moduleName, ObjectType: objectType, ActiveVersion: toVersion, Triggers: triggers}, nil
}

// Deregister entfernt die Registrierung eines Objekttyps zusammen mit
// allen zugehoerigen Triggern (Pruefung 3): der aktuelle Trigger-Satz
// wird geleert, die Registrierung als deregistriert markiert. Die
// Versionshistorie bleibt zur Nachvollziehbarkeit erhalten.
func Deregister(ctx context.Context, pool *pgxpool.Pool, moduleName, objectType string) error {
	tag, err := pool.Exec(ctx, `
		UPDATE module_registrations SET deregistered = true, updated_at = now()
		WHERE module_name = $1 AND object_type = $2
	`, moduleName, objectType)
	if err != nil {
		return fmt.Errorf("moduleadapter: deregistrierung schreiben: %w", err)
	}
	if tag.RowsAffected() == 0 {
		return ErrNotRegistered
	}
	return nil
}

// TriggersFor liefert die aktiven Trigger eines registrierten
// Objekttyps. Ist der Objekttyp nicht (mehr) registriert, liefert es
// ErrNotRegistered statt einer leeren Liste (Akzeptanzkriterium 3).
func TriggersFor(ctx context.Context, pool *pgxpool.Pool, moduleName, objectType string) ([]string, error) {
	var triggers []string
	var deregistered bool
	err := pool.QueryRow(ctx, `
		SELECT v.triggers, r.deregistered
		FROM module_registrations r
		JOIN module_registration_versions v ON v.registration_id = r.id AND v.version = r.active_version
		WHERE r.module_name = $1 AND r.object_type = $2
	`, moduleName, objectType).Scan(&triggers, &deregistered)
	if errors.Is(err, pgx.ErrNoRows) {
		return nil, ErrNotRegistered
	}
	if err != nil {
		return nil, fmt.Errorf("moduleadapter: trigger lesen: %w", err)
	}
	if deregistered {
		return nil, ErrNotRegistered
	}
	return triggers, nil
}

// ListRegistrations liefert alle aktiven Registrierungen (Pruefung 1:
// zwei Testadapter aus fiktivem DMS- und Mail-Modul beide korrekt
// registriert).
func ListRegistrations(ctx context.Context, pool *pgxpool.Pool) ([]Registration, error) {
	rows, err := pool.Query(ctx, `
		SELECT r.module_name, r.object_type, r.active_version, r.deregistered,
		       COALESCE(v.triggers, '[]'::jsonb)
		FROM module_registrations r
		LEFT JOIN module_registration_versions v ON v.registration_id = r.id AND v.version = r.active_version
		ORDER BY r.module_name, r.object_type
	`)
	if err != nil {
		return nil, fmt.Errorf("moduleadapter: registrierungen auflisten: %w", err)
	}
	defer rows.Close()

	var regs []Registration
	for rows.Next() {
		var r Registration
		if err := rows.Scan(&r.ModuleName, &r.ObjectType, &r.ActiveVersion, &r.Deregistered, &r.Triggers); err != nil {
			return nil, fmt.Errorf("moduleadapter: registrierungs-zeile lesen: %w", err)
		}
		regs = append(regs, r)
	}
	return regs, rows.Err()
}

func sameTriggers(a, b []string) bool {
	if len(a) != len(b) {
		return false
	}
	seen := make(map[string]int, len(a))
	for _, t := range a {
		seen[t]++
	}
	for _, t := range b {
		seen[t]--
	}
	for _, n := range seen {
		if n != 0 {
			return false
		}
	}
	return true
}