Merge remote-tracking branch 'origin/int10' into int11-merged-base

mail/docs/INT-10-PRUEFPROTOKOLL.md

@@ -0,0 +1,104 @@
+# INT-10 – Prüfprotokoll: Workflow-Anbindung (Objekttyp- & Auslöser-Registrierung, Kopplung Workflow ENG-02)
+
+Stand: 2026-10-16, Branch `feature/int-10-workflow-anbindung-objekttyp-ausloeser-registrierung-kopplung-workflow-eng-02`.
+
+Baut auf ARC-01 (Objekt-Speicher-Anbindung, bereits Fertig) auf. Setzt
+Workflow ENG-02 (Modul-Adapter-Schnittstelle, `workflow/internal/moduleadapter`)
+voraus — bereits auf `main` gemergt (`git log --oneline --all | grep ENG-02`
+zeigt u. a. `9c31ed0 ENG-02: modul adapter schnittstelle dms mail
+registrieren objekttypen trigger`, gemergt über
+`feature/eng-02-workflow-modul-adapter-schnittstelle`).
+
+## Umsetzung
+
+Neues, eigenständiges Paket `mail/internal/workflowevent`. Mail und
+Workflow sind physisch getrennte Go-Module (eigenes `go.mod` je Modul)
+und koppeln sich ausschließlich über HTTP an den ENG-02-Adapter
+(`workflow/internal/moduleadapter`, Routen `/register` und `/events`),
+nie über einen Go-Import.
+
+- `ModuleName = "mail"`, Objekttypen `ObjectTypeMessage = "message"` und
+  `ObjectTypeThread = "thread"` mit je eigenen Auslöser-Ereignissen:
+  - `message`: `mail.received` ("Mail eingegangen"), `mail.delivery_failed` ("Zustellung fehlgeschlagen")
+  - `thread`: `mail.deadline_reached` ("Frist erreicht")
+- `RegisterAll(ctx, cfg)` registriert beide Objekttypen beim Dienststart
+  (Akzeptanzkriterium 1); `Config.BaseURL` leer bedeutet abgeschaltet
+  (Standard aus, analog `mail/internal/syncalert`). HTTP 409 (unveränderte
+  Wiederholung) ist kein Fehler, damit ein Neustart mit gleichbleibenden
+  Triggern nicht scheitert.
+- Eigene Postgres-Jobqueue `mail_workflow_events`
+  (`FOR UPDATE SKIP LOCKED`, `pending`/`processing`/`succeeded`/`failed`) —
+  dasselbe Muster wie Core `internal/webhook` (API-07) und
+  `mail/internal/indexworker` (SRC-02), referenziert im Ticket als
+  Vorbild. `Publisher.PublishMailReceived` / `PublishDeliveryFailed` /
+  `PublishDeadlineReached` reihen nur ein (reine, schnelle
+  Datenbankeinfügung, kein HTTP-Aufruf) — das ist die einzige Schnittstelle,
+  die eine auslösende Mail-Aktion braucht (Akzeptanzkriterium 3).
+- `Dispatcher.RunOnce`/`Run` stellen im Hintergrund zu (`POST /events`).
+  Ein Fehlschlag markiert die Zeile mit `last_error` und plant nach
+  arithmetischem Backoff einen erneuten Versuch (`pending`), nach
+  `max_attempts` bleibt sie sichtbar `failed` — nie stillschweigend
+  verloren (Akzeptanzkriterium 3, Prüfung 3).
+
+## Echter Fund und Behebung (im ENG-02-Adapter, nicht in Mail behoben)
+
+Der erste reale Zustellversuch gegen die echte Workflow-Instanz schlug mit
+HTTP 400 fehl: `moduleadapter: ereignis speichern: ERROR: null value in
+column "dedupe_key" of relation "module_events" violates not-null
+constraint`. Ursache in `workflow/internal/moduleadapter/events.go`
+(`PublishEvent`): die Spalte `dedupe_key` ist `NOT NULL DEFAULT ''`
+(Migration `0013_event_dispatch.up.sql`), die INSERT-Anweisung setzt sie
+aber über `NULLIF($7, '')` bei leerem Schlüssel auf `NULL` — ein
+Widerspruch zum eigenen Dokumentationskommentar des `Event`-Typs ("leer
+ist zulässig, bedeutet dann aber keinen Dedup-Schutz"). Jede Zustellung
+ohne Dedup-Schlüssel schlägt serverseitig fehl.
+
+Das ist ein Fehler im ENG-02-Adapter selbst (anderes Board/Modul,
+bereits als "Fertig" gemergt) — laut Arbeitsweise dieser Kachel ("kein
+Umbau angrenzender Bereiche") NICHT dort behoben, sondern in Mail
+umgangen: `deliverEvent` setzt jetzt immer einen abgeleiteten,
+nie-leeren `dedupe_key` (`mail:{object_type}:{trigger}:{object_id}`)
+statt sich auf den dokumentierten leeren Fall zu verlassen. Dieser Fund
+und die Umgehung sind hier dokumentiert; eine Korrektur des Adapters
+selbst wäre ein eigenes Workflow-Ticket.
+
+## Prüfungen (alle real gegen Postgres-Testcluster 192.168.1.204, `TEST_TENANT_DSN`)
+
+| # | Prüfung | Ergebnis |
+|---|---|---|
+| 1 | Registrierung gegen eine echte Workflow-Testinstanz, Objekttyp und Ereignisse erscheinen dort | **bestanden** – `TestRegisterAll_RealWorkflowInstance_ObjectTypesAppear`: `workflow/cmd/workflow-app` als echter Subprozess gegen ein frisches Postgres-Testschema gestartet (alle Tenant-Migrationen `workflow/migrations/tenant/*.up.sql` vorab angewendet), `RegisterAll` über echtes HTTP, Ergebnis per SQL direkt aus `module_registrations`/`module_registration_versions` der laufenden Instanz gelesen: `message` → `[mail.received, mail.delivery_failed]`, `thread` → `[mail.deadline_reached]`. Zusätzlich: erneuter `RegisterAll`-Aufruf (simulierter Dienst-Neustart) scheitert nicht. |
+| 2 | Test: Mail eingegangen – Ereignis kommt nachweislich bei Workflow an | **bestanden** – `TestPublishMailReceived_DeliversToRealWorkflow`: `Publisher.PublishMailReceived` reiht ein, `Dispatcher.RunOnce` liefert an dieselbe echte Workflow-Instanz aus; Ankunft direkt per SQL aus `module_events` der Workflow-Instanz verifiziert (`accepted = true`, exakt 1 Zeile für `module=mail, object_type=message, trigger=mail.received, object_id=<Test-ID>`). |
+| 3 | Test: Workflow-Endpunkt zum Zustellzeitpunkt nicht erreichbar – Mailzustellung bleibt erfolgreich, Ereignis wird nachgeliefert oder sichtbar verworfen | **bestanden** – `TestDispatcher_UnreachableWorkflow_DoesNotBlockAndRetriesVisibly`: `Config.BaseURL` zeigt auf eine Adresse ohne Listener; `PublishMailReceived` (die "auslösende Mail-Aktion") kehrt in < 2 s fehlerfrei zurück, `Dispatcher.RunOnce` schlägt beim Zustellversuch fehl, Job bleibt sichtbar `status = pending` mit nicht-leerem `last_error` statt zu verschwinden. |
+
+## Build/Test-Ergebnis (192.168.1.204)
+
+```
+go build ./...                              -> clean (mail-Modul)
+go build ./...                              -> clean (workflow-Modul, für die Testinstanz)
+go vet ./...                                -> clean
+gofmt -l ./internal/workflowevent/          -> keine Ausgabe
+go test ./... -count=1 (gesamtes mail-Modul) -> alle Pakete bestanden, keine Regression
+go test ./internal/workflowevent/... -v     -> 3/3 PASS
+```
+
+`-race` war auf diesem Host nicht verfügbar (`CGO_ENABLED=0`, kein
+C-Toolchain installiert — Umgebungsvorgabe: keine Toolchain-Installationen
+auf diesem Host) und ist nicht Bestandteil der geforderten Prüfungen.
+
+## Nicht Bestandteil dieser Kachel
+
+- Eine konkrete Verdrahtung in den tatsächlichen SMTP-/Hotfolder-
+  Zustellpfad existiert im Mail-Modul noch nicht: `mail/internal/smtp`
+  kennt nur die Schnittstelle `MessageSink.Accept` und
+  `mail/internal/hotfolder` nur `Handler.ProcessFile` — beide sind laut
+  eigenem Kommentar ("Echte Ablage/Indexierung ist Sache späterer
+  Kacheln") bewusst noch ohne konkrete Implementierung. INT-10 liefert
+  den vollständigen, eigenständig testbaren Publisher/Dispatcher
+  (`Publisher.PublishMailReceived` etc.) als die Schnittstelle, die ein
+  künftiger konkreter Zustellpfad aufrufen muss — ihn selbst zu bauen
+  wäre Umbau eines angrenzenden Bereichs und nicht Teil dieses Tickets.
+  Dieselbe Lücke bestand bereits bei DMS INT-13 (dort ebenfalls ohne
+  Anbindung an eine konkrete Dokument-Anlage-Stelle).
+- Der oben beschriebene `dedupe_key`-Fehler im ENG-02-Adapter selbst ist
+  nicht behoben, nur umgangen (anderes Modul/Board).
+- Keine Änderung an `workflow/` oder anderen Modulen.

mail/internal/workflowevent/client.go

@@ -0,0 +1,126 @@
+package workflowevent
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"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"`
+	DedupeKey  string `json:"dedupe_key"`
+}
+
+// dedupeKeyFor liefert einen stabilen, nie leeren Dedup-Schlüssel je
+// Ereignis-Auslösung: der Workflow-Adapter speichert dedupe_key NOT NULL
+// und wandelt einen leeren Wert intern über NULLIF in NULL um, was dort
+// einen Datenbankfehler auslöst — ein leerer Schlüssel würde die
+// Zustellung also abweisen, obwohl der Adapter ihn als "zulässig, aber
+// ohne Dedup-Schutz" dokumentiert.
+func dedupeKeyFor(j *job) string {
+	return fmt.Sprintf("mail:%s:%s:%s", j.ObjectType, j.Trigger, j.ObjectID)
+}
+
+// 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,
+		DedupeKey:  dedupeKeyFor(j),
+	})
+	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 {
+		detail, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<16))
+		return fmt.Errorf("workflowevent: zustellung: unerwarteter status %d: %s", resp.StatusCode, string(detail))
+	}
+	return nil
+}

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 {
+			}
+		}
+	}
+}

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()
+)

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
+}

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)
+}

mail/internal/workflowevent/workflowevent_test.go

@@ -0,0 +1,205 @@
+package workflowevent
+
+import (
+	"context"
+	"os"
+	"testing"
+	"time"
+
+	"gitea.perlbach24.de/scripte/nexarch/mail/internal/testdb"
+)
+
+const testTenantSlug = "mandant-int10"
+
+func realDSN(t *testing.T) string {
+	t.Helper()
+	dsn := os.Getenv("TEST_TENANT_DSN")
+	if dsn == "" {
+		t.Skip("TEST_TENANT_DSN nicht gesetzt, Integrationstest übersprungen")
+	}
+	return dsn
+}
+
+// setupMailQueue legt die mail-eigene Jobqueue in einem eigenen Testschema
+// an -- getrennt vom Testschema der Workflow-Instanz, wie es auch in
+// Produktion zwei getrennte Datenbanken/Schemas wären.
+func setupMailQueue(t *testing.T, dsn string) *Queue {
+	t.Helper()
+	pool := testdb.Pool(t, dsn)
+	q := NewQueue(pool)
+	if err := q.EnsureSchema(context.Background()); err != nil {
+		t.Fatalf("schema: %v", err)
+	}
+	return q
+}
+
+// TestRegisterAll_RealWorkflowInstance_ObjectTypesAppear prüft
+// Akzeptanzkriterium 1 und Prüfung 1: die Objekttypen "message" und
+// "thread" samt Auslöser-Ereignissen erscheinen nach RegisterAll
+// tatsächlich bei einer echten, laufenden Workflow-Testinstanz.
+func TestRegisterAll_RealWorkflowInstance_ObjectTypesAppear(t *testing.T) {
+	dsn := realDSN(t)
+	wf := startRealWorkflowInstance(t, dsn)
+	ctx := context.Background()
+
+	cfg := Config{BaseURL: wf.BaseURL, ServiceToken: wf.ServiceToken}
+	if err := RegisterAll(ctx, cfg); err != nil {
+		t.Fatalf("registerall: %v", err)
+	}
+
+	rows, err := wf.Pool.Query(ctx, `
+		SELECT r.object_type, v.triggers
+		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 NOT r.deregistered
+		ORDER BY r.object_type
+	`, ModuleName)
+	if err != nil {
+		t.Fatalf("registrierungen abfragen: %v", err)
+	}
+	defer rows.Close()
+
+	got := map[string][]string{}
+	for rows.Next() {
+		var objectType string
+		var triggers []string
+		if err := rows.Scan(&objectType, &triggers); err != nil {
+			t.Fatalf("zeile lesen: %v", err)
+		}
+		got[objectType] = triggers
+	}
+	if err := rows.Err(); err != nil {
+		t.Fatalf("rows: %v", err)
+	}
+
+	if len(got) != 2 {
+		t.Fatalf("erwarte 2 registrierte objekttypen (message, thread), habe %v", got)
+	}
+	if !sameSet(got[ObjectTypeMessage], MessageTriggers) {
+		t.Fatalf("message-trigger = %v, want %v", got[ObjectTypeMessage], MessageTriggers)
+	}
+	if !sameSet(got[ObjectTypeThread], ThreadTriggers) {
+		t.Fatalf("thread-trigger = %v, want %v", got[ObjectTypeThread], ThreadTriggers)
+	}
+
+	// Ein erneuter Aufruf beim nächsten Dienststart darf nicht scheitern
+	// (Akzeptanzkriterium 1: "beim Start automatisch" -- auch bei jedem
+	// Neustart, nicht nur beim allerersten).
+	if err := RegisterAll(ctx, cfg); err != nil {
+		t.Fatalf("erneutes registerall beim naechsten start: %v", err)
+	}
+}
+
+// TestPublishMailReceived_DeliversToRealWorkflow prüft Akzeptanzkriterium
+// 2 und Prüfung 2: "Mail eingegangen" wird eingereiht und vom Dispatcher
+// tatsächlich an eine echte Workflow-Instanz zugestellt.
+func TestPublishMailReceived_DeliversToRealWorkflow(t *testing.T) {
+	dsn := realDSN(t)
+	wf := startRealWorkflowInstance(t, dsn)
+	queue := setupMailQueue(t, dsn)
+	ctx := context.Background()
+
+	cfg := Config{BaseURL: wf.BaseURL, ServiceToken: wf.ServiceToken}
+	if err := RegisterAll(ctx, cfg); err != nil {
+		t.Fatalf("registerall: %v", err)
+	}
+
+	pub := NewPublisher(queue)
+	messageID := "msg-int10-" + testdb.Suffix()
+	jobID, err := pub.PublishMailReceived(ctx, testTenantSlug, messageID)
+	if err != nil {
+		t.Fatalf("publish: %v", err)
+	}
+
+	dispatcher := NewDispatcher(queue, cfg, "test-dispatcher")
+	if err := dispatcher.RunOnce(ctx); err != nil {
+		t.Fatalf("runonce: %v", err)
+	}
+	if st, serr := queue.Status(ctx, jobID); serr == nil {
+		t.Logf("debug job status=%s attempts=%d lasterr=%s", st.State, st.Attempts, st.LastError)
+	}
+
+	var count int
+	var accepted bool
+	err = wf.Pool.QueryRow(ctx, `
+		SELECT count(*), COALESCE(bool_and(accepted), false) FROM module_events
+		WHERE module_name = $1 AND object_type = $2 AND trigger = $3 AND object_id = $4
+	`, ModuleName, ObjectTypeMessage, TriggerReceived, messageID).Scan(&count, &accepted)
+	if err != nil {
+		t.Fatalf("ereignis abfragen: %v", err)
+	}
+	if count != 1 {
+		rows2, _ := wf.Pool.Query(ctx, "SELECT module_name, object_type, trigger, object_id, tenant_slug, accepted, rejected_reason FROM module_events")
+		for rows2.Next() {
+			var mn, ot, tr, oid, ts string
+			var acc bool
+			var rr *string
+			_ = rows2.Scan(&mn, &ot, &tr, &oid, &ts, &acc, &rr)
+			t.Logf("row: module=%s type=%s trigger=%s object=%s tenant=%s accepted=%v reason=%v", mn, ot, tr, oid, ts, acc, rr)
+		}
+		t.Fatalf("erwarte genau 1 angekommenes ereignis, habe %d", count)
+	}
+	if !accepted {
+		t.Fatal("ereignis wurde nicht als 'accepted' bei workflow verbucht")
+	}
+}
+
+// TestDispatcher_UnreachableWorkflow_DoesNotBlockAndRetriesVisibly prüft
+// Akzeptanzkriterium 3 und Prüfung 3: ein zum Zustellzeitpunkt nicht
+// erreichbarer Workflow-Endpunkt blockiert nicht die auslösende Mail-
+// Aktion (Enqueue bleibt schnell und fehlerfrei), und der Fehlschlag wird
+// sichtbar nachgehalten (last_error, Status bleibt "pending" für einen
+// erneuten Versuch) statt stillschweigend verworfen zu werden.
+func TestDispatcher_UnreachableWorkflow_DoesNotBlockAndRetriesVisibly(t *testing.T) {
+	dsn := realDSN(t)
+	queue := setupMailQueue(t, dsn)
+	ctx := context.Background()
+
+	// Absichtlich nicht erreichbare Adresse (kein Listener):
+	cfg := Config{BaseURL: "http://127.0.0.1:1", ServiceToken: "irrelevant"}
+
+	pub := NewPublisher(queue)
+	start := time.Now()
+	id, err := pub.PublishMailReceived(ctx, testTenantSlug, "msg-int10-unreachable-"+testdb.Suffix())
+	if err != nil {
+		t.Fatalf("publish: %v", err)
+	}
+	if elapsed := time.Since(start); elapsed > 2*time.Second {
+		t.Fatalf("publish blockierte auf dem nicht erreichbaren workflow, dauer=%v", elapsed)
+	}
+
+	dispatcher := NewDispatcher(queue, cfg, "test-dispatcher")
+	if err := dispatcher.RunOnce(ctx); err != nil {
+		t.Fatalf("runonce: %v", err)
+	}
+
+	status, err := queue.Status(ctx, id)
+	if err != nil {
+		t.Fatalf("status: %v", err)
+	}
+	if status.State != StatusPending {
+		t.Fatalf("status = %q, want %q (erneuter versuch eingeplant statt verloren)", status.State, StatusPending)
+	}
+	if status.LastError == "" {
+		t.Fatal("last_error ist leer, fehlschlag ist nicht sichtbar dokumentiert")
+	}
+}
+
+func sameSet(a, b []string) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	seen := map[string]int{}
+	for _, v := range a {
+		seen[v]++
+	}
+	for _, v := range b {
+		seen[v]--
+	}
+	for _, n := range seen {
+		if n != 0 {
+			return false
+		}
+	}
+	return true
+}

mail/internal/workflowevent/workflowtestinstance_test.go

@@ -0,0 +1,209 @@
+package workflowevent
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"net/http"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"runtime"
+	"sort"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"gitea.perlbach24.de/scripte/nexarch/mail/internal/testdb"
+)
+
+// TestMain entfernt das einmalig gebaute workflow-app-Binary wieder, damit
+// wiederholte Testläufe nicht Temp-Verzeichnisse ansammeln.
+func TestMain(m *testing.M) {
+	code := m.Run()
+	if workflowBinaryPath != "" {
+		_ = os.RemoveAll(filepath.Dir(workflowBinaryPath))
+	}
+	os.Exit(code)
+}
+
+// realWorkflowInstance ist eine ECHTE, lauffähige Workflow-App
+// (workflow/cmd/workflow-app), gestartet als eigener Prozess gegen ein
+// frisches Postgres-Testschema — keine Simulation des HTTP-Vertrags, wie
+// von der Ticket-Prüfung "gegen eine echte Workflow-Testinstanz"
+// gefordert. Mail und Workflow sind physisch getrennte Go-Module
+// (eigene go.mod je Modul), deshalb Kopplung ausschließlich über HTTP
+// wie in Produktion, nicht über einen Go-Import.
+type realWorkflowInstance struct {
+	BaseURL      string
+	ServiceToken string
+	Pool         *pgxpool.Pool // dasselbe Testschema, fuer Pruefungen direkt per SQL
+	cmd          *exec.Cmd
+}
+
+var (
+	workflowBinaryOnce sync.Once
+	workflowBinaryPath string
+	workflowBinaryErr  error
+)
+
+// workflowModuleRoot findet das Wurzelverzeichnis des Workflow-Go-Moduls
+// (nexarch/workflow) relativ zu dieser Testdatei — Mail und Workflow
+// liegen als Geschwisterverzeichnisse im selben Checkout.
+func workflowModuleRoot(t *testing.T) string {
+	t.Helper()
+	_, thisFile, _, ok := runtime.Caller(0)
+	if !ok {
+		t.Fatal("eigener dateipfad nicht ermittelbar")
+	}
+	dir := filepath.Dir(thisFile)
+	for {
+		if _, err := os.Stat(filepath.Join(dir, "go.mod")); err == nil {
+			break
+		}
+		parent := filepath.Dir(dir)
+		if parent == dir {
+			t.Fatal("go.mod des mail-moduls nicht gefunden")
+		}
+		dir = parent
+	}
+	repoRoot := filepath.Dir(dir)
+	workflowRoot := filepath.Join(repoRoot, "workflow")
+	if _, err := os.Stat(filepath.Join(workflowRoot, "go.mod")); err != nil {
+		t.Skipf("workflow-modul nicht neben mail-modul gefunden (%s): %v", workflowRoot, err)
+	}
+	return workflowRoot
+}
+
+func buildWorkflowApp(t *testing.T) string {
+	t.Helper()
+	workflowBinaryOnce.Do(func() {
+		workflowRoot := workflowModuleRoot(t)
+		dir, err := os.MkdirTemp("", "workflow-app-bin-*")
+		if err != nil {
+			workflowBinaryErr = err
+			return
+		}
+		bin := filepath.Join(dir, "workflow-app")
+		cmd := exec.Command("go", "build", "-o", bin, "./cmd/workflow-app")
+		cmd.Dir = workflowRoot
+		out, err := cmd.CombinedOutput()
+		if err != nil {
+			workflowBinaryErr = fmt.Errorf("workflow-app bauen: %w\n%s", err, out)
+			return
+		}
+		workflowBinaryPath = bin
+	})
+	if workflowBinaryErr != nil {
+		t.Fatalf("%v", workflowBinaryErr)
+	}
+	return workflowBinaryPath
+}
+
+func freeLocalPort(t *testing.T) int {
+	t.Helper()
+	l, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("freien port ermitteln: %v", err)
+	}
+	defer l.Close()
+	return l.Addr().(*net.TCPAddr).Port
+}
+
+// applyWorkflowMigrations wendet alle Workflow-Tenant-Migrationen in
+// Dateireihenfolge an -- reine SQL-Dateien, kein Import von
+// workflow/internal/migrate (anderes Go-Modul).
+func applyWorkflowMigrations(ctx context.Context, t *testing.T, pool *pgxpool.Pool, workflowRoot string) {
+	t.Helper()
+	dir := filepath.Join(workflowRoot, "migrations", "tenant")
+	entries, err := os.ReadDir(dir)
+	if err != nil {
+		t.Fatalf("migrationsverzeichnis lesen: %v", err)
+	}
+	var files []string
+	for _, e := range entries {
+		if strings.HasSuffix(e.Name(), ".up.sql") {
+			files = append(files, e.Name())
+		}
+	}
+	sort.Strings(files)
+	for _, f := range files {
+		sql, err := os.ReadFile(filepath.Join(dir, f))
+		if err != nil {
+			t.Fatalf("migration %q lesen: %v", f, err)
+		}
+		if _, err := pool.Exec(ctx, string(sql)); err != nil {
+			t.Fatalf("migration %q anwenden: %v", f, err)
+		}
+	}
+}
+
+// startRealWorkflowInstance baut (einmal pro Testlauf) und startet die
+// echte workflow-app gegen ein frisches Testschema. Prüfung 1: der
+// Objekttyp und seine Ereignisse landen in einer tatsächlich laufenden
+// Workflow-Instanz, nicht in einem Fake.
+func startRealWorkflowInstance(t *testing.T, dsn string) *realWorkflowInstance {
+	t.Helper()
+	workflowRoot := workflowModuleRoot(t)
+	bin := buildWorkflowApp(t)
+
+	pool, schema := testdb.PoolWithSchema(t, dsn, nil)
+	ctx := context.Background()
+	applyWorkflowMigrations(ctx, t, pool, workflowRoot)
+
+	port := freeLocalPort(t)
+	addr := fmt.Sprintf("127.0.0.1:%d", port)
+	serviceToken := "int10-test-token-" + testdb.Suffix()
+	schemaDSN := dsn + "&search_path=" + schema + ",public"
+
+	cmd := exec.Command(bin)
+	cmd.Env = append(os.Environ(),
+		"NEXARCH_WORKFLOW_APP_TENANT_DSN="+schemaDSN,
+		"NEXARCH_WORKFLOW_APP_SERVICE_TOKEN="+serviceToken,
+		"NEXARCH_WORKFLOW_APP_TENANT_SLUG=mandant-int10",
+		"NEXARCH_TENANT_REGISTRY_URL=http://127.0.0.1:1",
+		"NEXARCH_WORKFLOW_APP_LISTEN_ADDR="+addr,
+	)
+	var stderr strings.Builder
+	cmd.Stderr = &stderr
+	if err := cmd.Start(); err != nil {
+		t.Fatalf("workflow-app starten: %v", err)
+	}
+
+	baseURL := "http://" + addr
+	if err := waitForHealthz(baseURL, 10*time.Second); err != nil {
+		_ = cmd.Process.Kill()
+		t.Fatalf("workflow-app wurde nicht rechtzeitig erreichbar: %v\nstderr:\n%s", err, stderr.String())
+	}
+
+	t.Cleanup(func() {
+		if cmd.Process != nil {
+			_ = cmd.Process.Kill()
+			_ = cmd.Wait()
+		}
+	})
+
+	return &realWorkflowInstance{BaseURL: baseURL, ServiceToken: serviceToken, Pool: pool, cmd: cmd}
+}
+
+func waitForHealthz(baseURL string, timeout time.Duration) error {
+	deadline := time.Now().Add(timeout)
+	var lastErr error
+	for time.Now().Before(deadline) {
+		resp, err := http.Get(baseURL + "/healthz")
+		if err == nil {
+			resp.Body.Close()
+			if resp.StatusCode == http.StatusOK {
+				return nil
+			}
+			lastErr = fmt.Errorf("status %d", resp.StatusCode)
+		} else {
+			lastErr = err
+		}
+		time.Sleep(100 * time.Millisecond)
+	}
+	return lastErr
+}