nexarch/workflow/internal/eventtrigger/eventtrigger.go

// Package eventtrigger implementiert ENG-09: die mandantenweite
// Zuordnung von DMS-Ereignissen (aus moduleadapter, ENG-02/INT-13) zu
// genau einer Aufgabe oder einem Genehmigungslauf. Dispatch verarbeitet
// jede angenommene module_events-Zeile hoechstens einmal -- Aufgabe
// anlegen und Zeile als verarbeitet markieren laufen in derselben
// Transaktion, damit ein Abbruch dazwischen die Zeile unverarbeitet
// (und damit fuer den naechsten Lauf wieder sichtbar) laesst, statt eine
// zweite Aufgabe zu erzeugen.
package eventtrigger

import (
	"context"
	"errors"
	"fmt"

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

	"gitea.perlbach24.de/scripte/nexarch/workflow/internal/approval"
	"gitea.perlbach24.de/scripte/nexarch/workflow/internal/task"
)

const (
	KindTask     = "task"
	KindApproval = "approval"
)

var ErrRuleExists = errors.New("eventtrigger: regel fuer modul/objekttyp/ausloeser existiert bereits")

// Rule ist die mandantenweite Zuordnung eines Ereignisses zu einer
// Aufgabe oder einem Genehmigungslauf (Akzeptanzkriterium 2).
type Rule struct {
	ID           string
	ModuleName   string
	ObjectType   string
	Trigger      string
	Kind         string
	AssigneeType task.AssigneeType
	AssigneeID   string
	Title        string
	Enabled      bool
}

// RuleStore verwaltet die Zuordnungsregeln.
type RuleStore struct {
	pool *pgxpool.Pool
}

func NewRuleStore(pool *pgxpool.Pool) *RuleStore {
	return &RuleStore{pool: pool}
}

// Put legt eine Regel fuer (module, objectType, trigger) an. Eine
// bestehende Regel fuer dieselbe Kombination wird nicht stillschweigend
// ersetzt (kein versehentliches Umbiegen laufender Zuordnungen).
func (s *RuleStore) Put(ctx context.Context, r Rule) (Rule, error) {
	if r.Kind != KindTask && r.Kind != KindApproval {
		return Rule{}, fmt.Errorf("eventtrigger: kind muss %q oder %q sein", KindTask, KindApproval)
	}
	if r.ModuleName == "" || r.ObjectType == "" || r.Trigger == "" || r.AssigneeID == "" || r.Title == "" {
		return Rule{}, fmt.Errorf("eventtrigger: modulname, objekttyp, ausloeser, zustaendigkeit und titel sind pflicht")
	}
	err := s.pool.QueryRow(ctx, `
		INSERT INTO event_task_rules (module_name, object_type, trigger, kind, assignee_type, assignee_id, title, enabled)
		VALUES ($1, $2, $3, $4, $5, $6, $7, true)
		RETURNING id
	`, r.ModuleName, r.ObjectType, r.Trigger, r.Kind, string(r.AssigneeType), r.AssigneeID, r.Title).Scan(&r.ID)
	if isUniqueViolation(err) {
		return Rule{}, ErrRuleExists
	}
	if err != nil {
		return Rule{}, fmt.Errorf("eventtrigger: regel anlegen: %w", err)
	}
	r.Enabled = true
	return r, nil
}

func (s *RuleStore) ruleFor(ctx context.Context, q querier, moduleName, objectType, trigger string) (Rule, bool, error) {
	var r Rule
	var assigneeType string
	err := q.QueryRow(ctx, `
		SELECT id, module_name, object_type, trigger, kind, assignee_type, assignee_id, title, enabled
		FROM event_task_rules
		WHERE module_name = $1 AND object_type = $2 AND trigger = $3 AND enabled
	`, moduleName, objectType, trigger).Scan(&r.ID, &r.ModuleName, &r.ObjectType, &r.Trigger, &r.Kind, &assigneeType, &r.AssigneeID, &r.Title, &r.Enabled)
	if errors.Is(err, pgx.ErrNoRows) {
		return Rule{}, false, nil
	}
	if err != nil {
		return Rule{}, false, fmt.Errorf("eventtrigger: regel lesen: %w", err)
	}
	r.AssigneeType = task.AssigneeType(assigneeType)
	return r, true, nil
}

type querier interface {
	QueryRow(ctx context.Context, sql string, args ...any) pgx.Row
}

// pendingEvent ist eine angenommene, noch nicht verarbeitete Zeile aus
// module_events.
type pendingEvent struct {
	ID         string
	ModuleName string
	ObjectType string
	Trigger    string
	ObjectID   string
}

// Dispatcher fuehrt die Zuordnungsregeln gegen angenommene Ereignisse
// aus und legt daraus Aufgaben/Genehmigungen an.
type Dispatcher struct {
	pool     *pgxpool.Pool
	rules    *RuleStore
	tasks    *task.Store
	approval *approval.Service
}

func NewDispatcher(pool *pgxpool.Pool, rules *RuleStore, tasks *task.Store, approvalSvc *approval.Service) *Dispatcher {
	return &Dispatcher{pool: pool, rules: rules, tasks: tasks, approval: approvalSvc}
}

// Dispatch verarbeitet alle offenen, angenommenen Ereignisse und
// liefert, wie viele davon tatsaechlich eine Aufgabe/Genehmigung
// ausgeloest haben. Ereignisse ohne passende (oder deaktivierte) Regel
// bleiben unverarbeitet liegen -- kein Fehler, aber auch kein
// Fortschritt, bis eine Regel existiert (Akzeptanzkriterium 3: nur
// konfigurierte Zustaendigkeiten erhalten Aufgaben).
func (d *Dispatcher) Dispatch(ctx context.Context) (int, error) {
	rows, err := d.pool.Query(ctx, `
		SELECT id, module_name, object_type, trigger, object_id
		FROM module_events
		WHERE accepted AND processed_at IS NULL
		ORDER BY received_at
	`)
	if err != nil {
		return 0, fmt.Errorf("eventtrigger: offene ereignisse lesen: %w", err)
	}
	var pending []pendingEvent
	for rows.Next() {
		var e pendingEvent
		if err := rows.Scan(&e.ID, &e.ModuleName, &e.ObjectType, &e.Trigger, &e.ObjectID); err != nil {
			rows.Close()
			return 0, fmt.Errorf("eventtrigger: ereignis-zeile lesen: %w", err)
		}
		pending = append(pending, e)
	}
	rows.Close()
	if err := rows.Err(); err != nil {
		return 0, fmt.Errorf("eventtrigger: ereignisse iterieren: %w", err)
	}

	n := 0
	for _, e := range pending {
		dispatched, err := d.dispatchOne(ctx, e)
		if err != nil {
			return n, err
		}
		if dispatched {
			n++
		}
	}
	return n, nil
}

// dispatchOne verarbeitet ein Ereignis nach dem Muster
// Reservieren-Ausfuehren-Bestaetigen, weil task.Store und
// approval.Service ihre eigene Transaktion fuehren und sich deshalb
// nicht in dieselbe Transaktion wie das Markieren des Ereignisses
// einschliessen lassen:
//  1. event_dispatch_log wird per INSERT ... ON CONFLICT DO NOTHING
//     reserviert (event_id ist Primary Key) -- ein zweiter Aufruf fuer
//     dasselbe Ereignis reserviert nicht erneut und legt daher auch
//     keine zweite Aufgabe an (Akzeptanzkriterium 2).
//  2. Aufgabe/Genehmigung wird angelegt.
//     3a. Erfolg: die Reservierung wird mit der erzeugten Objekt-ID
//     bestaetigt und module_events als verarbeitet markiert.
//     3b. Fehlschlag: die Reservierung wird wieder geloescht, damit ein
//     spaeterer Dispatch-Lauf es erneut versuchen kann -- das
//     verbleibende Restrisiko ist ein Absturz exakt zwischen 2 und 3a,
//     das Ereignis bliebe dann dauerhaft reserviert ohne erzeugtes
//     Objekt (dokumentiert in ENG-09-PRUEFPROTOKOLL.md).
func (d *Dispatcher) dispatchOne(ctx context.Context, e pendingEvent) (bool, error) {
	rule, ok, err := d.rules.ruleFor(ctx, d.pool, e.ModuleName, e.ObjectType, e.Trigger)
	if err != nil {
		return false, err
	}
	if !ok {
		return false, nil
	}

	tag, err := d.pool.Exec(ctx, `
		INSERT INTO event_dispatch_log (event_id, kind, created_object_id)
		VALUES ($1, $2, '')
		ON CONFLICT (event_id) DO NOTHING
	`, e.ID, rule.Kind)
	if err != nil {
		return false, fmt.Errorf("eventtrigger: dispatch reservieren: %w", err)
	}
	if tag.RowsAffected() == 0 {
		return false, nil
	}

	ref := task.Reference{ModuleName: e.ModuleName, ObjectType: e.ObjectType, ObjectID: e.ObjectID}
	var createdObjectID string
	switch rule.Kind {
	case KindTask:
		createdObjectID, err = d.tasks.Create(ctx, task.Task{
			Title:        rule.Title,
			AssigneeType: rule.AssigneeType,
			AssigneeID:   rule.AssigneeID,
			References:   []task.Reference{ref},
		})
	case KindApproval:
		subject := fmt.Sprintf("%s:%s:%s", e.ModuleName, e.ObjectType, e.ObjectID)
		var t approval.Task
		t, err = d.approval.Create(ctx, rule.AssigneeID, subject, "eventtrigger")
		createdObjectID = t.ID
	default:
		err = fmt.Errorf("eventtrigger: unbekannte regelart %q", rule.Kind)
	}
	if err != nil {
		if _, delErr := d.pool.Exec(ctx, `DELETE FROM event_dispatch_log WHERE event_id = $1`, e.ID); delErr != nil {
			return false, fmt.Errorf("eventtrigger: erzeugen fehlgeschlagen (%v) und reservierung konnte nicht zurueckgenommen werden: %w", err, delErr)
		}
		return false, fmt.Errorf("eventtrigger: erzeugen: %w", err)
	}

	if _, err := d.pool.Exec(ctx, `
		UPDATE event_dispatch_log SET created_object_id = $2 WHERE event_id = $1
	`, e.ID, createdObjectID); err != nil {
		return false, fmt.Errorf("eventtrigger: dispatch bestaetigen: %w", err)
	}
	if _, err := d.pool.Exec(ctx, `UPDATE module_events SET processed_at = now() WHERE id = $1`, e.ID); err != nil {
		return false, fmt.Errorf("eventtrigger: ereignis als verarbeitet markieren: %w", err)
	}
	return true, nil
}

func isUniqueViolation(err error) bool {
	var pgErr *pgconn.PgError
	return errors.As(err, &pgErr) && pgErr.Code == "23505"
}