nexarch/dms/internal/workflowevent/workflowevent_test.go

package workflowevent

import (
	"context"
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"os"
	"sync"
	"testing"
	"time"

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

	"gitea.perlbach24.de/scripte/nexarch/common/testdb"
	"gitea.perlbach24.de/scripte/nexarch/dms/internal/jobqueue"
)

// fakeWorkflow simuliert den ENG-02-Adapter (workflow/internal/moduleadapter,
// Handler.go) exakt: gleiche Routen, gleiche Header-Pruefung, gleiche
// Statuscodes. Eine eigenständig laufende Workflow-Instanz existiert noch
// nicht (das Modul ist bislang reine Bibliothek ohne cmd/-Dienst), daher
// bildet dieser Fake den echten, aus dem Quellcode gelesenen HTTP-Vertrag
// nach — dokumentiert im PRUEFPROTOKOLL als Grenze dieses Tests.
type fakeWorkflow struct {
	mu              sync.Mutex
	token           string
	registrations   []registerRequest
	events          []publishEventRequest
	unreachable     bool
	rejectAllEvents bool
}

func newFakeWorkflow(token string) *fakeWorkflow {
	return &fakeWorkflow{token: token}
}

func (f *fakeWorkflow) server() *httptest.Server {
	mux := http.NewServeMux()
	mux.HandleFunc("POST /register", func(w http.ResponseWriter, r *http.Request) {
		if r.Header.Get("X-Service-Token") != f.token {
			w.WriteHeader(http.StatusUnauthorized)
			return
		}
		var req registerRequest
		_ = json.NewDecoder(r.Body).Decode(&req)
		f.mu.Lock()
		f.registrations = append(f.registrations, req)
		f.mu.Unlock()
		w.WriteHeader(http.StatusOK)
	})
	mux.HandleFunc("POST /events", func(w http.ResponseWriter, r *http.Request) {
		if r.Header.Get("X-Service-Token") != f.token {
			w.WriteHeader(http.StatusUnauthorized)
			return
		}
		f.mu.Lock()
		reject := f.rejectAllEvents
		f.mu.Unlock()
		if reject {
			w.WriteHeader(http.StatusInternalServerError)
			return
		}
		var req publishEventRequest
		_ = json.NewDecoder(r.Body).Decode(&req)
		f.mu.Lock()
		f.events = append(f.events, req)
		f.mu.Unlock()
		w.WriteHeader(http.StatusAccepted)
	})
	return httptest.NewServer(mux)
}

func (f *fakeWorkflow) eventCount() int {
	f.mu.Lock()
	defer f.mu.Unlock()
	return len(f.events)
}

func (f *fakeWorkflow) registrationCount() int {
	f.mu.Lock()
	defer f.mu.Unlock()
	return len(f.registrations)
}

// TestRegister_SendsObjectTypeAndTriggers prueft Akzeptanzkriterium 1 und
// Pruefung 1: der Objekttyp "document" samt allen Auslöser-Ereignissen wird
// beim Start bei Workflow registriert.
func TestRegister_SendsObjectTypeAndTriggers(t *testing.T) {
	fw := newFakeWorkflow("secret-token")
	srv := fw.server()
	defer srv.Close()

	cfg := Config{BaseURL: srv.URL, ServiceToken: "secret-token", TenantSlug: "mandant-a"}
	if err := Register(context.Background(), cfg); err != nil {
		t.Fatalf("register: %v", err)
	}
	if fw.registrationCount() != 1 {
		t.Fatalf("registrationCount = %d, want 1", fw.registrationCount())
	}
	got := fw.registrations[0]
	if got.ModuleName != ModuleName || got.ObjectType != ObjectType {
		t.Fatalf("unerwartete registrierung: %+v", got)
	}
	if len(got.Triggers) != len(Triggers) {
		t.Fatalf("triggers = %v, want %v", got.Triggers, Triggers)
	}
	// ENG-06: Workflow prueft tenant_slug unabhaengig gegen Core, DMS
	// muss ihn also mitschicken, sonst wird die Registrierung abgelehnt.
	if got.TenantSlug != "mandant-a" {
		t.Fatalf("tenant_slug = %q, want mandant-a", got.TenantSlug)
	}
}

// TestRegister_Disabled prueft: ohne BaseURL passiert nichts (Standard aus).
func TestRegister_Disabled(t *testing.T) {
	if err := Register(context.Background(), Config{}); err != nil {
		t.Fatalf("register mit leerer config soll nichts tun: %v", err)
	}
}

// TestPublishAndDeliver_EventArrivesAtWorkflow prueft Akzeptanzkriterium 2
// und Pruefung 2: "document.created" wird eingereiht und vom Worker
// tatsaechlich an Workflow zugestellt.
func TestPublishAndDeliver_EventArrivesAtWorkflow(t *testing.T) {
	pool := setupTest(t)
	fw := newFakeWorkflow("secret-token")
	srv := fw.server()
	defer srv.Close()

	cfg := Config{BaseURL: srv.URL, ServiceToken: "secret-token"}
	queue := jobqueue.NewQueue(pool, time.Minute)
	pub := NewPublisher(queue)

	if err := pub.Publish(context.Background(), TriggerCreated, "doc-123"); err != nil {
		t.Fatalf("publish: %v", err)
	}

	worker := jobqueue.NewWorker(queue, "test-worker", []string{JobType}, time.Second, Handler(cfg))
	worker.RunOnce(context.Background())

	if fw.eventCount() != 1 {
		t.Fatalf("eventCount = %d, want 1", fw.eventCount())
	}
	got := fw.events[0]
	if got.ModuleName != ModuleName || got.ObjectType != ObjectType || got.Trigger != TriggerCreated || got.ObjectID != "doc-123" {
		t.Fatalf("unerwartetes ereignis: %+v", got)
	}
}

// TestPublish_DoesNotBlockOnUnreachableWorkflow prueft Akzeptanzkriterium 3
// und Pruefung 3: Publish selbst spricht Workflow nie synchron an — ein
// nicht erreichbarer Dienst wirkt sich erst beim (asynchronen) Zustellversuch
// aus, nie auf die auslösende DMS-Aktion. Der fehlgeschlagene Zustellversuch
// wird ueber die bestehende jobqueue-Retry/Dead-Letter-Logik sichtbar
// nachgehalten, nicht stillschweigend verworfen.
func TestPublish_DoesNotBlockOnUnreachableWorkflow(t *testing.T) {
	pool := setupTest(t)
	queue := jobqueue.NewQueue(pool, time.Minute)
	pub := NewPublisher(queue)

	// Absichtlich nicht erreichbare Adresse (kein Listener):
	cfg := Config{BaseURL: "http://127.0.0.1:1", ServiceToken: "secret-token"}

	start := time.Now()
	id, err := publishWithID(context.Background(), pub, TriggerCreated, "doc-999")
	if err != nil {
		t.Fatalf("publish: %v", err)
	}
	if time.Since(start) > time.Second {
		t.Fatalf("publish blockierte, dauer=%v", time.Since(start))
	}

	worker := jobqueue.NewWorker(queue, "test-worker", []string{JobType}, time.Second, Handler(cfg))
	worker.RunOnce(context.Background())

	job, err := queue.Status(context.Background(), id)
	if err != nil {
		t.Fatalf("status: %v", err)
	}
	if job.Status != jobqueue.StatusPending {
		t.Fatalf("status = %q, want pending (retry geplant, nicht stillschweigend verloren)", job.Status)
	}
	if job.LastError == nil || *job.LastError == "" {
		t.Fatalf("last_error leer, Fehlschlag ist nicht sichtbar dokumentiert")
	}
}

// publishWithID reiht wie Publisher.Publish ein, liefert aber zusaetzlich
// die Job-ID fuer den Statusabgleich in der Pruefung.
func publishWithID(ctx context.Context, p *Publisher, trigger, objectID string) (string, error) {
	return p.queue.Enqueue(ctx, JobType, eventPayload{Trigger: trigger, ObjectID: objectID}, jobqueue.EnqueueOptions{})
}

func setupTest(t *testing.T) *pgxpool.Pool {
	t.Helper()
	dsn := os.Getenv("TEST_TENANT_DSN")
	if dsn == "" {
		t.Skip("TEST_TENANT_DSN nicht gesetzt, Integrationstest uebersprungen")
	}
	ctx := context.Background()

	pool := testdb.Pool(t, dsn)
	t.Cleanup(func() { pool.Close() })

	if _, err := pool.Exec(ctx, `
		CREATE EXTENSION IF NOT EXISTS pgcrypto;
		CREATE TABLE IF NOT EXISTS processing_jobs (
			id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
			job_type        TEXT NOT NULL,
			payload         JSONB NOT NULL DEFAULT '{}'::jsonb,
			idempotency_key TEXT UNIQUE,
			status          TEXT NOT NULL DEFAULT 'pending'
			                CHECK (status IN ('pending', 'processing', 'succeeded', 'failed', 'dead_letter')),
			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()
		);
	`); err != nil {
		t.Fatalf("schema: %v", err)
	}
	t.Cleanup(func() {
		_, _ = pool.Exec(context.Background(), `DROP TABLE IF EXISTS processing_jobs`)
	})
	return pool
}