nexarch/mail/internal/workflowevent/workflowtestinstance_test.go

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
}