From 1cef097271c4cdc66554d7febe91ea58bcbf1b82 Mon Sep 17 00:00:00 2001 From: patrick Date: Fri, 16 Oct 2026 06:38:20 +0000 Subject: [PATCH] INT-10: workflow anbindung objekttyp ausloeser registrierung kopplung workflow eng 02 Mail-Objekttypen message/thread samt Ausloeser-Ereignissen (mail.received, mail.delivery_failed, mail.deadline_reached) ueber den ENG-02-Adapter bei Workflow registrierbar (RegisterAll). Ereignis-Publikation laeuft ueber eine eigene Postgres-Jobqueue (mail_workflow_events, FOR UPDATE SKIP LOCKED) analog zu internal/webhook (Core) und mail/internal/indexworker: Publisher.Publish* reiht nur ein (kein HTTP-Aufruf im Anfragepfad), Dispatcher stellt im Hintergrund zu und macht Fehlschlaege per last_error/Retry-Backoff sichtbar statt sie zu verwerfen. Co-Authored-By: Claude Sonnet 5 --- mail/internal/workflowevent/client.go | 112 +++++++++++ mail/internal/workflowevent/dispatcher.go | 58 ++++++ .../migrations/0001_mail_workflow_events.sql | 17 ++ mail/internal/workflowevent/queue.go | 187 ++++++++++++++++++ mail/internal/workflowevent/workflowevent.go | 57 ++++++ 5 files changed, 431 insertions(+) create mode 100644 mail/internal/workflowevent/client.go create mode 100644 mail/internal/workflowevent/dispatcher.go create mode 100644 mail/internal/workflowevent/migrations/0001_mail_workflow_events.sql create mode 100644 mail/internal/workflowevent/queue.go create mode 100644 mail/internal/workflowevent/workflowevent.go diff --git a/mail/internal/workflowevent/client.go b/mail/internal/workflowevent/client.go new file mode 100644 index 00000000..21385928 --- /dev/null +++ b/mail/internal/workflowevent/client.go @@ -0,0 +1,112 @@ +package workflowevent + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" + + "gitea.perlbach24.de/scripte/nexarch/mail/internal/httpclient" +) + +// Config steuert Ziel und Zugangsdaten des Workflow-Modul-Adapters +// (ENG-02). BaseURL leer bedeutet abgeschaltet — Standard aus, analog +// mail/internal/syncalert. Keine Zugangsdaten im Code: BaseURL/ServiceToken +// kommen ausschließlich aus Umgebungsvariablen des Aufrufers. +type Config struct { + BaseURL string + ServiceToken string +} + +// Enabled meldet, ob eine Workflow-Anbindung konfiguriert ist. +func (c Config) Enabled() bool { return c.BaseURL != "" } + +type registerRequest struct { + ModuleName string `json:"module_name"` + ObjectType string `json:"object_type"` + Triggers []string `json:"triggers"` +} + +// Register registriert einen Objekttyp mit seinen Auslöser-Ereignissen +// beim Workflow-Modul-Adapter (Akzeptanzkriterium 1). Status 409 bedeutet +// eine unveränderte Wiederholung (moduleadapter.ErrDuplicateTriggers) und +// ist kein Fehler. +func Register(ctx context.Context, cfg Config, objectType string, triggers []string) error { + if !cfg.Enabled() { + return nil + } + body, err := json.Marshal(registerRequest{ModuleName: ModuleName, ObjectType: objectType, Triggers: triggers}) + if err != nil { + return fmt.Errorf("workflowevent: registrierung serialisieren: %w", err) + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, cfg.BaseURL+"/register", bytes.NewReader(body)) + if err != nil { + return fmt.Errorf("workflowevent: registrierungs-anfrage aufbauen: %w", err) + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("X-Service-Token", cfg.ServiceToken) + + resp, err := httpclient.Default().Do(req) + if err != nil { + return fmt.Errorf("workflowevent: registrierung fehlgeschlagen: %w", err) + } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusConflict { + return fmt.Errorf("workflowevent: registrierung: unerwarteter status %d", resp.StatusCode) + } + return nil +} + +// RegisterAll registriert alle Mail-Objekttypen (Mail/Thread) mitsamt +// ihren Auslöser-Ereignissen bei Workflow — vom Aufrufer einmal beim +// Start des Mail-Dienstes aufzurufen (Akzeptanzkriterium 1). +func RegisterAll(ctx context.Context, cfg Config) error { + if err := Register(ctx, cfg, ObjectTypeMessage, MessageTriggers); err != nil { + return fmt.Errorf("workflowevent: objekttyp %q registrieren: %w", ObjectTypeMessage, err) + } + if err := Register(ctx, cfg, ObjectTypeThread, ThreadTriggers); err != nil { + return fmt.Errorf("workflowevent: objekttyp %q registrieren: %w", ObjectTypeThread, err) + } + return nil +} + +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"` +} + +// deliverEvent stellt EIN Ereignis an Workflow zu. Aufgerufen nur vom +// Dispatcher, nie im Anfragepfad der auslösenden Mail-Aktion +// (Akzeptanzkriterium 3). +func deliverEvent(ctx context.Context, cfg Config, j *job) error { + body, err := json.Marshal(publishEventRequest{ + ModuleName: ModuleName, + ObjectType: j.ObjectType, + Trigger: j.Trigger, + ObjectID: j.ObjectID, + TenantSlug: j.TenantSlug, + }) + if err != nil { + return fmt.Errorf("workflowevent: ereignis serialisieren: %w", err) + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, cfg.BaseURL+"/events", bytes.NewReader(body)) + if err != nil { + return fmt.Errorf("workflowevent: zustellanfrage aufbauen: %w", err) + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("X-Service-Token", cfg.ServiceToken) + + resp, err := httpclient.Default().Do(req) + if err != nil { + return fmt.Errorf("workflowevent: zustellung fehlgeschlagen: %w", err) + } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return fmt.Errorf("workflowevent: zustellung: unerwarteter status %d", resp.StatusCode) + } + return nil +} diff --git a/mail/internal/workflowevent/dispatcher.go b/mail/internal/workflowevent/dispatcher.go new file mode 100644 index 00000000..e4a74277 --- /dev/null +++ b/mail/internal/workflowevent/dispatcher.go @@ -0,0 +1,58 @@ +package workflowevent + +import ( + "context" + "time" +) + +// Dispatcher liefert eingereihte Ereignisse an Workflow aus — getrennt +// von Publisher, damit Enqueue (Anfragepfad) und Zustellung (Hintergrund) +// unabhängig bleiben (Akzeptanzkriterium 3). +type Dispatcher struct { + queue *Queue + cfg Config + workerID string +} + +func NewDispatcher(queue *Queue, cfg Config, workerID string) *Dispatcher { + return &Dispatcher{queue: queue, cfg: cfg, workerID: workerID} +} + +// RunOnce stellt genau EIN fälliges Ereignis zu, falls eines vorliegt. +// ErrNoJobAvailable ist kein Fehlerzustand (leere Queue). +func (d *Dispatcher) RunOnce(ctx context.Context) error { + j, err := d.queue.dequeue(ctx, d.workerID) + if err != nil { + return err + } + + if !d.cfg.Enabled() { + // Anbindung abgeschaltet (keine BaseURL konfiguriert): das + // Ereignis gilt als erledigt statt endlos zu wiederholen. + return d.queue.complete(ctx, j.ID) + } + + if err := deliverEvent(ctx, d.cfg, j); err != nil { + if failErr := d.queue.fail(ctx, j.ID, err); failErr != nil { + return failErr + } + return nil + } + return d.queue.complete(ctx, j.ID) +} + +// Run ruft RunOnce in festen Abständen auf, bis ctx beendet wird — +// dieselbe Konvention wie internal/webhook.Dispatcher.Run. +func (d *Dispatcher) Run(ctx context.Context, interval time.Duration) { + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + for d.RunOnce(ctx) == nil { + } + } + } +} diff --git a/mail/internal/workflowevent/migrations/0001_mail_workflow_events.sql b/mail/internal/workflowevent/migrations/0001_mail_workflow_events.sql new file mode 100644 index 00000000..1c80ec63 --- /dev/null +++ b/mail/internal/workflowevent/migrations/0001_mail_workflow_events.sql @@ -0,0 +1,17 @@ +CREATE TABLE IF NOT EXISTS mail_workflow_events ( + id BIGSERIAL PRIMARY KEY, + object_type TEXT NOT NULL, + trigger TEXT NOT NULL, + object_id TEXT NOT NULL, + tenant_slug TEXT NOT NULL, + payload JSONB NOT NULL DEFAULT '{}'::jsonb, + status TEXT NOT NULL DEFAULT 'pending' CHECK (status IN ('pending', 'processing', 'succeeded', 'failed')), + attempts INT NOT NULL DEFAULT 0, + max_attempts INT NOT NULL DEFAULT 5, + available_at TIMESTAMPTZ NOT NULL DEFAULT now(), + locked_at TIMESTAMPTZ, + locked_by TEXT, + last_error TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +) diff --git a/mail/internal/workflowevent/queue.go b/mail/internal/workflowevent/queue.go new file mode 100644 index 00000000..d2b7ad66 --- /dev/null +++ b/mail/internal/workflowevent/queue.go @@ -0,0 +1,187 @@ +package workflowevent + +import ( + "context" + _ "embed" + "errors" + "fmt" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +//go:embed migrations/0001_mail_workflow_events.sql +var schemaMigration string + +const ( + StatusPending = "pending" + StatusProcessing = "processing" + StatusSucceeded = "succeeded" + StatusFailed = "failed" +) + +var ErrNoJobAvailable = errors.New("workflowevent: kein ereignis zur zustellung verfügbar") + +var ErrNotFound = errors.New("workflowevent: ereignis nicht gefunden") + +const defaultMaxAttempts = 5 + +// Queue ist die Postgres-Jobqueue fuer ausgehende Workflow-Ereignisse — +// dasselbe FOR-UPDATE-SKIP-LOCKED-Muster wie mail/internal/indexworker +// und internal/webhook (Core API-07), hier ohne Stale-Lock-Wiedervorlage, +// da ein einzelner Dispatcher-Prozess genuegt (kein verteilter Worker-Pool +// fuer diese Kachel). +type Queue struct { + pool *pgxpool.Pool +} + +func NewQueue(pool *pgxpool.Pool) *Queue { + return &Queue{pool: pool} +} + +// EnsureSchema legt die Tabelle an, falls sie noch nicht existiert — +// gleiches Muster wie mail/internal/indexworker (kein zentraler +// Migrationslaeufer fuer Mandanten-Datenbanken im Mail-Modul). +func (q *Queue) EnsureSchema(ctx context.Context) error { + if _, err := q.pool.Exec(ctx, schemaMigration); err != nil { + return fmt.Errorf("workflowevent: schema anlegen: %w", err) + } + return nil +} + +// job ist eine einzelne, zur Zustellung an Workflow vorgesehene +// Ereigniszeile. +type job struct { + ID int64 + ObjectType string + Trigger string + ObjectID string + TenantSlug string + Payload []byte + Attempts int +} + +// Enqueue reiht EIN Ereignis zur asynchronen Zustellung ein +// (Akzeptanzkriterium 3: diese Methode spricht Workflow selbst nie an, +// sie ist eine reine, schnelle Datenbank-Einfuegung — ein nicht +// erreichbares Workflow kann die auslösende Mail-Aktion also nicht +// blockieren). +func (q *Queue) Enqueue(ctx context.Context, objectType, trigger, objectID, tenantSlug string, payload []byte) (int64, error) { + if payload == nil { + payload = []byte(`{}`) + } + var id int64 + err := q.pool.QueryRow(ctx, ` + INSERT INTO mail_workflow_events (object_type, trigger, object_id, tenant_slug, payload, max_attempts) + VALUES ($1, $2, $3, $4, $5, $6) + RETURNING id + `, objectType, trigger, objectID, tenantSlug, payload, defaultMaxAttempts).Scan(&id) + if err != nil { + return 0, fmt.Errorf("workflowevent: ereignis einreihen: %w", err) + } + return id, nil +} + +// dequeue holt GENAU EIN fälliges Ereignis und markiert es atomar als +// "processing". +func (q *Queue) dequeue(ctx context.Context, workerID string) (*job, error) { + tx, err := q.pool.Begin(ctx) + if err != nil { + return nil, fmt.Errorf("workflowevent: transaktion starten: %w", err) + } + defer func() { _ = tx.Rollback(ctx) }() + + row := tx.QueryRow(ctx, ` + SELECT id, object_type, trigger, object_id, tenant_slug, payload, attempts + FROM mail_workflow_events + WHERE status = 'pending' AND available_at <= now() + ORDER BY available_at + FOR UPDATE SKIP LOCKED + LIMIT 1 + `) + + var j job + if err := row.Scan(&j.ID, &j.ObjectType, &j.Trigger, &j.ObjectID, &j.TenantSlug, &j.Payload, &j.Attempts); err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrNoJobAvailable + } + return nil, fmt.Errorf("workflowevent: nächstes ereignis lesen: %w", err) + } + + if _, err := tx.Exec(ctx, ` + UPDATE mail_workflow_events + SET status = 'processing', attempts = attempts + 1, locked_at = now(), locked_by = $2, updated_at = now() + WHERE id = $1 + `, j.ID, workerID); err != nil { + return nil, fmt.Errorf("workflowevent: ereignis sperren: %w", err) + } + + if err := tx.Commit(ctx); err != nil { + return nil, fmt.Errorf("workflowevent: dequeue committen: %w", err) + } + + j.Attempts++ + return &j, nil +} + +func (q *Queue) complete(ctx context.Context, id int64) error { + tag, err := q.pool.Exec(ctx, ` + UPDATE mail_workflow_events SET status = 'succeeded', locked_at = NULL, locked_by = NULL, updated_at = now() + WHERE id = $1 + `, id) + if err != nil { + return fmt.Errorf("workflowevent: ereignis abschließen: %w", err) + } + if tag.RowsAffected() == 0 { + return ErrNotFound + } + return nil +} + +// fail markiert einen Zustellversuch als fehlgeschlagen (Prüfung 3: die +// Zustellung wird nachgeliefert statt stillschweigend verloren zu gehen). +// Nach Erreichen von max_attempts bleibt das Ereignis dauerhaft 'failed', +// last_error macht den Ausfall sichtbar. +func (q *Queue) fail(ctx context.Context, id int64, cause error) error { + tag, err := q.pool.Exec(ctx, ` + UPDATE mail_workflow_events + SET status = CASE WHEN attempts >= max_attempts THEN 'failed' ELSE 'pending' END, + available_at = now() + (LEAST(attempts, 10) * interval '10 seconds'), + locked_at = NULL, locked_by = NULL, last_error = $2, updated_at = now() + WHERE id = $1 + `, id, cause.Error()) + if err != nil { + return fmt.Errorf("workflowevent: fehlschlag erfassen: %w", err) + } + if tag.RowsAffected() == 0 { + return ErrNotFound + } + return nil +} + +// Status liefert den aktuellen Zustand eines eingereihten Ereignisses +// (für Tests/Prüfungen — Prüfung 3 braucht sichtbaren Fehlerzustand). +type Status struct { + ID int64 + State string + Attempts int + LastError string +} + +func (q *Queue) Status(ctx context.Context, id int64) (*Status, error) { + var s Status + var lastError *string + err := q.pool.QueryRow(ctx, ` + SELECT id, status, attempts, last_error FROM mail_workflow_events WHERE id = $1 + `, id).Scan(&s.ID, &s.State, &s.Attempts, &lastError) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrNotFound + } + return nil, fmt.Errorf("workflowevent: status lesen: %w", err) + } + if lastError != nil { + s.LastError = *lastError + } + return &s, nil +} diff --git a/mail/internal/workflowevent/workflowevent.go b/mail/internal/workflowevent/workflowevent.go new file mode 100644 index 00000000..92c961d4 --- /dev/null +++ b/mail/internal/workflowevent/workflowevent.go @@ -0,0 +1,57 @@ +// Package workflowevent implementiert INT-10: Mail registriert die +// Objekttypen "message" und "thread" mitsamt Auslöser-Ereignissen bei +// NEXARCH Workflow (ENG-02-Adapter) und publiziert Ereignisse asynchron +// über eine eigene Postgres-Jobqueue (Muster wie Core internal/webhook +// und mail/internal/indexworker) — kein eigener Genehmigungs- oder +// Aufgaben-Code in Mail, das bleibt Workflow vorbehalten. +package workflowevent + +import "context" + +const ( + ModuleName = "mail" + + ObjectTypeMessage = "message" + ObjectTypeThread = "thread" + + TriggerReceived = "mail.received" + TriggerDeliveryFailed = "mail.delivery_failed" + TriggerDeadlineReached = "mail.deadline_reached" +) + +// MessageTriggers sind die beim Objekttyp "message" registrierten +// Auslöser-Ereignisse. +var MessageTriggers = []string{TriggerReceived, TriggerDeliveryFailed} + +// ThreadTriggers sind die beim Objekttyp "thread" registrierten +// Auslöser-Ereignisse. +var ThreadTriggers = []string{TriggerDeadlineReached} + +// Publisher reiht Ereignisse in die Jobqueue ein (Akzeptanzkriterium 2/3: +// nie synchron an Workflow zustellen, ein nicht erreichbares Workflow +// blockiert die auslösende Mail-Aktion nicht). +type Publisher struct { + queue *Queue +} + +func NewPublisher(queue *Queue) *Publisher { + return &Publisher{queue: queue} +} + +// PublishMailReceived reiht das Ereignis "Mail eingegangen" für eine +// einzelne Nachricht ein (Akzeptanzkriterium 2). +func (p *Publisher) PublishMailReceived(ctx context.Context, tenantSlug, messageID string) (int64, error) { + return p.queue.Enqueue(ctx, ObjectTypeMessage, TriggerReceived, messageID, tenantSlug, nil) +} + +// PublishDeliveryFailed reiht das Ereignis "Zustellung fehlgeschlagen" +// für eine einzelne Nachricht ein. +func (p *Publisher) PublishDeliveryFailed(ctx context.Context, tenantSlug, messageID string) (int64, error) { + return p.queue.Enqueue(ctx, ObjectTypeMessage, TriggerDeliveryFailed, messageID, tenantSlug, nil) +} + +// PublishDeadlineReached reiht das Ereignis "Frist erreicht" für einen +// Mail-Thread ein. +func (p *Publisher) PublishDeadlineReached(ctx context.Context, tenantSlug, threadID string) (int64, error) { + return p.queue.Enqueue(ctx, ObjectTypeThread, TriggerDeadlineReached, threadID, tenantSlug, nil) +}