// 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 ("@") 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 }