nexarch/workflow/internal/processdef/processdef.go

// Package processdef ist das deklarative, versionierbare
// Prozessdefinitions-Format (ENG-03): Prozessdefinitionen kommen als JSON
// statt als Go-Wert (Akzeptanzkriterium 1, "ohne Codeaenderung aenderbar"),
// werden beim Laden gegen ein Schema geprueft (Akzeptanzkriterium 2) und
// jede geladene Version bleibt dauerhaft erhalten.
//
// Die Versionierung setzt bewusst NICHT in statemachine.Engine an (ENG-01,
// unveraendert): jede Version wird der Engine unter einem eindeutigen,
// versionierten internen Namen ("<Name>@<Version>") bekanntgegeben.
// Instanzen speichern diesen versionierten Namen als ihren Prozesstyp
// (statemachine.Instance.ProcessName) und bleiben damit dauerhaft an die
// Definition gebunden, mit der sie gestartet wurden — ein spaeterer
// Versionswechsel oder Rollback aendert nur, welche Version NEUE Instanzen
// bekommen (Akzeptanzkriterium 3, Pruefung 2 und 3).
package processdef

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"

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

	"gitea.perlbach24.de/scripte/nexarch/workflow/internal/statemachine"
)

// ErrInvalidDefinition wird zurueckgegeben, wenn eine Definition beim Laden
// die Schema-Pruefung nicht besteht (Akzeptanzkriterium 2, Pruefung 1).
var ErrInvalidDefinition = errors.New("processdef: ungueltige prozessdefinition")

// ErrUnknownVersion wird zurueckgegeben, wenn ein Rollback auf eine nicht
// existierende Version versucht wird.
var ErrUnknownVersion = errors.New("processdef: unbekannte version")

// ErrNoActiveVersion wird zurueckgegeben, wenn fuer einen Prozessnamen noch
// keine Version aktiv ist.
var ErrNoActiveVersion = errors.New("processdef: keine aktive version")

// Transition ist ein erlaubter Uebergang im deklarativen Format.
type Transition struct {
	From string `json:"from"`
	To   string `json:"to"`
}

// Definition ist eine Prozessdefinition im deklarativen JSON-Format: lesbar,
// ohne Codeaenderung anpassbar (Akzeptanzkriterium 1).
type Definition struct {
	Name        string       `json:"name"`
	States      []string     `json:"states"`
	Initial     string       `json:"initial"`
	Transitions []Transition `json:"transitions"`
}

// Parse liest eine Definition aus JSON und prueft sie strukturell (Schema:
// Pflichtfelder, Startzustand und alle Uebergangs-Endpunkte muessen zu den
// deklarierten Zustaenden gehoeren). Eine fehlerhafte Definition wird mit
// ErrInvalidDefinition abgewiesen statt sie unvalidiert zu uebernehmen
// (Akzeptanzkriterium 2, Pruefung 1).
func Parse(data []byte) (Definition, error) {
	var d Definition
	if err := json.Unmarshal(data, &d); err != nil {
		return Definition{}, fmt.Errorf("%w: json: %v", ErrInvalidDefinition, err)
	}
	if err := d.validate(); err != nil {
		return Definition{}, err
	}
	return d, nil
}

func (d Definition) hasState(s string) bool {
	for _, st := range d.States {
		if st == s {
			return true
		}
	}
	return false
}

func (d Definition) validate() error {
	if d.Name == "" {
		return fmt.Errorf("%w: name fehlt", ErrInvalidDefinition)
	}
	if len(d.States) == 0 {
		return fmt.Errorf("%w: prozess %q: keine zustaende deklariert", ErrInvalidDefinition, d.Name)
	}
	if !d.hasState(d.Initial) {
		return fmt.Errorf("%w: prozess %q: startzustand %q ist kein deklarierter zustand", ErrInvalidDefinition, d.Name, d.Initial)
	}
	for _, t := range d.Transitions {
		if !d.hasState(t.From) || !d.hasState(t.To) {
			return fmt.Errorf("%w: prozess %q: uebergang %s->%s referenziert unbekannten zustand", ErrInvalidDefinition, d.Name, t.From, t.To)
		}
	}
	return nil
}

// versionedName ist der Name, unter dem die Engine (ENG-01) diese konkrete
// Version fuehrt.
func versionedName(name string, version int) string {
	return fmt.Sprintf("%s@%d", name, version)
}

func (d Definition) toStateMachine(version int) statemachine.ProcessDefinition {
	transitions := make([]statemachine.Transition, len(d.Transitions))
	for i, t := range d.Transitions {
		transitions[i] = statemachine.Transition{From: t.From, To: t.To}
	}
	return statemachine.ProcessDefinition{
		Name:        versionedName(d.Name, version),
		States:      d.States,
		Initial:     d.Initial,
		Transitions: transitions,
	}
}

// Registry verwaltet geladene Prozessdefinitionen und die je Prozessname
// aktive Version in Postgres.
type Registry struct {
	pool *pgxpool.Pool
}

// NewRegistry bindet die Registry an den Mandanten-Pool.
func NewRegistry(pool *pgxpool.Pool) *Registry {
	return &Registry{pool: pool}
}

// Load validiert eine Definition, legt sie als neue Version ab und macht
// sie zur aktiven Version fuer neue Instanzen. Bereits laufende Instanzen
// bleiben unberuehrt, weil sie unter dem alten versionierten Namen laufen
// (Akzeptanzkriterium 3).
func (r *Registry) Load(ctx context.Context, raw []byte) (Definition, int, error) {
	def, err := Parse(raw)
	if err != nil {
		return Definition{}, 0, err
	}
	tx, err := r.pool.Begin(ctx)
	if err != nil {
		return Definition{}, 0, fmt.Errorf("processdef: transaktion starten: %w", err)
	}
	defer func() { _ = tx.Rollback(ctx) }()

	var nextVersion int
	if err := tx.QueryRow(ctx, `
		SELECT COALESCE(MAX(version), 0) + 1 FROM process_definitions WHERE name = $1
	`, def.Name).Scan(&nextVersion); err != nil {
		return Definition{}, 0, fmt.Errorf("processdef: naechste version ermitteln: %w", err)
	}

	sourceJSON, err := json.Marshal(def)
	if err != nil {
		return Definition{}, 0, fmt.Errorf("processdef: definition serialisieren: %w", err)
	}
	if _, err := tx.Exec(ctx, `
		INSERT INTO process_definitions (name, version, source_json) VALUES ($1, $2, $3)
	`, def.Name, nextVersion, sourceJSON); err != nil {
		return Definition{}, 0, fmt.Errorf("processdef: version speichern: %w", err)
	}
	if _, err := tx.Exec(ctx, `
		INSERT INTO process_definition_active (name, version) VALUES ($1, $2)
		ON CONFLICT (name) DO UPDATE SET version = EXCLUDED.version
	`, def.Name, nextVersion); err != nil {
		return Definition{}, 0, fmt.Errorf("processdef: aktive version setzen: %w", err)
	}
	if err := tx.Commit(ctx); err != nil {
		return Definition{}, 0, fmt.Errorf("processdef: laden committen: %w", err)
	}
	return def, nextVersion, nil
}

// ActiveVersion liefert die derzeit fuer neue Instanzen aktive Version.
func (r *Registry) ActiveVersion(ctx context.Context, name string) (int, error) {
	var v int
	err := r.pool.QueryRow(ctx, `
		SELECT version FROM process_definition_active WHERE name = $1
	`, name).Scan(&v)
	if errors.Is(err, pgx.ErrNoRows) {
		return 0, ErrNoActiveVersion
	}
	if err != nil {
		return 0, fmt.Errorf("processdef: aktive version lesen: %w", err)
	}
	return v, nil
}

// RollbackTo setzt die aktive Version auf eine bereits geladene, aeltere
// (oder juengere) Version zurueck (Pruefung 3). Bereits laufende Instanzen
// sind davon unberuehrt, weil sie unter ihrem eigenen versionierten Namen
// laufen; nur NEUE Start()-Aufrufe bekommen die zurueckgesetzte Version.
func (r *Registry) RollbackTo(ctx context.Context, name string, version int) error {
	var exists bool
	if err := r.pool.QueryRow(ctx, `
		SELECT EXISTS(SELECT 1 FROM process_definitions WHERE name = $1 AND version = $2)
	`, name, version).Scan(&exists); err != nil {
		return fmt.Errorf("processdef: version pruefen: %w", err)
	}
	if !exists {
		return fmt.Errorf("%w: %s@%d", ErrUnknownVersion, name, version)
	}
	if _, err := r.pool.Exec(ctx, `
		INSERT INTO process_definition_active (name, version) VALUES ($1, $2)
		ON CONFLICT (name) DO UPDATE SET version = EXCLUDED.version
	`, name, version); err != nil {
		return fmt.Errorf("processdef: rollback schreiben: %w", err)
	}
	return nil
}

// All laedt alle jemals geladenen Versionen aller Prozesse als
// statemachine.ProcessDefinition, jeweils unter ihrem versionierten Namen.
// Damit lassen sich ALLE Versionen (nicht nur die aktiven) an
// statemachine.NewEngine uebergeben, sodass laufende Instanzen aelterer
// Versionen weiterhin gegen ihre eigene Definition arbeiten.
func (r *Registry) All(ctx context.Context) ([]statemachine.ProcessDefinition, error) {
	rows, err := r.pool.Query(ctx, `SELECT name, version, source_json FROM process_definitions`)
	if err != nil {
		return nil, fmt.Errorf("processdef: definitionen lesen: %w", err)
	}
	defer rows.Close()
	var out []statemachine.ProcessDefinition
	for rows.Next() {
		var name string
		var version int
		var raw []byte
		if err := rows.Scan(&name, &version, &raw); err != nil {
			return nil, fmt.Errorf("processdef: definition lesen: %w", err)
		}
		var def Definition
		if err := json.Unmarshal(raw, &def); err != nil {
			return nil, fmt.Errorf("processdef: gespeicherte definition %s@%d beschaedigt: %w", name, version, err)
		}
		out = append(out, def.toStateMachine(version))
	}
	return out, rows.Err()
}

// BuildEngine liest alle jemals geladenen Definitionsversionen und baut
// eine statemachine.Engine, die alle davon kennt (jede unter ihrem
// versionierten Namen). Nach jedem Load() muss der Aufrufer die Engine ueber
// diese Methode neu aufbauen, damit die neue Version fuer Start() sichtbar
// wird — eine reine Laufzeitaktion, keine Codeaenderung
// (Akzeptanzkriterium 1).
func (r *Registry) BuildEngine(ctx context.Context) (*statemachine.Engine, error) {
	defs, err := r.All(ctx)
	if err != nil {
		return nil, err
	}
	return statemachine.NewEngine(r.pool, defs...)
}

// Start startet eine neue Instanz der derzeit aktiven Version von name.
// Laesst die Engine dabei den versionierten Namen verwenden, damit die
// Instanz dauerhaft an diese Version gebunden bleibt.
func (r *Registry) Start(ctx context.Context, engine *statemachine.Engine, name, actor string) (statemachine.Instance, error) {
	v, err := r.ActiveVersion(ctx, name)
	if err != nil {
		return statemachine.Instance{}, err
	}
	inst, err := engine.Start(ctx, versionedName(name, v), actor)
	if err != nil {
		return statemachine.Instance{}, err
	}
	return inst, nil
}