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 <noreply@anthropic.com>
This commit is contained in:
patrickandClaude Sonnet 5 committed
1 parent
dac7fff504
commit
1cef097271
5 files changed
+431
No files matched your search
@@ -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
|
||||
}
|
||||
@@ -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 {
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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()
|
||||
)
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
Reference in new issue
Block a user