nexarch/mail/internal/workflowevent/queue.go

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
}