nexarch/internal/resync/resync_test.go

package resync

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

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

	"gitea.perlbach24.de/scripte/nexarch/internal/flag"
	"gitea.perlbach24.de/scripte/nexarch/internal/moduleregistry"
)

// fakeAudit steht fuer internal/audit.Log (AUD-01, nicht Abhaengigkeit
// dieser Kachel) — zeichnet Aufrufe in Empfangsreihenfolge auf, damit
// Vollstaendigkeit UND Reihenfolge geprueft werden koennen.
type fakeAudit struct {
	mu     sync.Mutex
	events []auditEventDTO
	failAt int // -1 = nie fehlschlagen; sonst: ab diesem Index (0-basiert) schlaegt Record fehl
}

func (f *fakeAudit) Record(ctx context.Context, tenantSlug, actor, action, target string, metadata map[string]any, occurredAt time.Time) error {
	f.mu.Lock()
	defer f.mu.Unlock()
	if f.failAt >= 0 && len(f.events) == f.failAt {
		return fmt.Errorf("simulierter core-ausfall waehrend der nachlieferung")
	}
	f.events = append(f.events, auditEventDTO{TenantSlug: tenantSlug, Actor: actor, Action: action, Target: target, Metadata: metadata, OccurredAt: occurredAt})
	return nil
}

// fakeUsage steht fuer internal/usage.Store (LIC-03, nicht Abhaengigkeit
// dieser Kachel) — summiert Deltas wie der echte Store.
type fakeUsage struct {
	mu     sync.Mutex
	totals map[string]int64
}

func newFakeUsage() *fakeUsage { return &fakeUsage{totals: map[string]int64{}} }

func (f *fakeUsage) Increment(ctx context.Context, tenantSlug, metric string, delta int64) error {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.totals[tenantSlug+"|"+metric] += delta
	return nil
}

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

	pool, err := pgxpool.New(ctx, adminDSN)
	if err != nil {
		t.Fatalf("pool: %v", err)
	}
	if _, err := pool.Exec(ctx, `
		CREATE TABLE IF NOT EXISTS resync_audit_buffer (
			id BIGSERIAL PRIMARY KEY, seq BIGSERIAL, tenant_slug TEXT NOT NULL, actor TEXT NOT NULL,
			action TEXT NOT NULL, target TEXT NOT NULL DEFAULT '', metadata JSONB NOT NULL DEFAULT '{}',
			created_at TIMESTAMPTZ NOT NULL DEFAULT now()
		);
		CREATE TABLE IF NOT EXISTS resync_usage_buffer (
			id BIGSERIAL PRIMARY KEY, seq BIGSERIAL, tenant_slug TEXT NOT NULL, metric TEXT NOT NULL,
			delta BIGINT NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT now()
		);
		CREATE TABLE IF NOT EXISTS feature_flags (
			key TEXT PRIMARY KEY, enabled BOOLEAN NOT NULL DEFAULT false,
			rollout_percentage INT NOT NULL DEFAULT 0, target_tenant_slugs TEXT[] NOT NULL DEFAULT '{}',
			updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
		);
		CREATE TABLE IF NOT EXISTS modules (
			name TEXT PRIMARY KEY, version TEXT NOT NULL CHECK (version <> ''),
			required_flags TEXT[] NOT NULL DEFAULT '{}', registered_at TIMESTAMPTZ NOT NULL DEFAULT now()
		);
		CREATE TABLE IF NOT EXISTS module_credentials (
			module_name TEXT PRIMARY KEY REFERENCES modules(name),
			client_id TEXT NOT NULL UNIQUE, secret_hash BYTEA NOT NULL,
			issued_at TIMESTAMPTZ NOT NULL DEFAULT now()
		);
	`); err != nil {
		t.Fatalf("schema: %v", err)
	}

	cleanup := func() { pool.Close() }
	return NewBuffer(pool), pool, cleanup
}

func uniqueModuleName() string {
	return fmt.Sprintf("resync-test-%d", time.Now().UnixNano())
}

// setupModuleCredential registriert ein echtes Modul + Service-Credential
// ueber internal/moduleregistry (API-02) — dieselbe Authentifizierung wird
// vom Handler tatsaechlich geprueft, kein Mock.
func setupModuleCredential(t *testing.T, pool *pgxpool.Pool) (registry *moduleregistry.Registry, clientID, secret string) {
	t.Helper()
	ctx := context.Background()
	flagService := flag.NewService(flag.NewStore(pool), 10*time.Millisecond)
	registry = moduleregistry.NewRegistry(pool, flagService)
	name := uniqueModuleName()
	if _, err := registry.Register(ctx, name, "1.0.0", nil); err != nil {
		t.Fatalf("modul registrieren: %v", err)
	}
	clientID, secret, err := registry.Provision(ctx, name)
	if err != nil {
		t.Fatalf("credential provisionieren: %v", err)
	}
	return registry, clientID, secret
}

// Akzeptanzkriterium 2 + Pruefung 1: waehrend eines simulierten Ausfalls
// lokal gepufferte Audit-Events sind nach Wiederanlauf vollstaendig und in
// korrekter Reihenfolge in Core's Audit-Log vorhanden.
func TestFlushAll_DeliversBufferedAuditEventsCompleteAndInOrder(t *testing.T) {
	buffer, pool, cleanup := setupTest(t)
	defer cleanup()
	ctx := context.Background()
	registry, clientID, secret := setupModuleCredential(t, pool)

	tenantSlug := "acme"
	// Ereignisse "waehrend core down" lokal puffern.
	actions := []string{"login", "upload", "delete", "logout"}
	for _, action := range actions {
		if err := buffer.EnqueueAuditEvent(ctx, tenantSlug, "user-1", action, "res-1", nil); err != nil {
			t.Fatalf("enqueue %s: %v", action, err)
		}
	}

	audit := &fakeAudit{failAt: -1}
	usage := newFakeUsage()
	handler := NewHandler(registry, audit, usage)

	mux := http.NewServeMux()
	mux.HandleFunc("/internal/resync/audit", handler.AuditHandler)
	mux.HandleFunc("/internal/resync/usage", handler.UsageHandler)
	mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) })
	server := httptest.NewServer(mux)
	defer server.Close()

	client := NewCoreClient(server.URL, clientID, secret)
	worker := NewWorker(buffer, client)

	if err := worker.FlushAll(ctx); err != nil {
		t.Fatalf("flushall: %v", err)
	}

	audit.mu.Lock()
	defer audit.mu.Unlock()
	if len(audit.events) != len(actions) {
		t.Fatalf("erwartet %d zugestellte events, habe %d", len(actions), len(audit.events))
	}
	for i, e := range audit.events {
		if e.Action != actions[i] {
			t.Fatalf("reihenfolge falsch: position %d = %q, want %q", i, e.Action, actions[i])
		}
	}

	// Puffer muss nach bestaetigter Zustellung leer sein.
	remaining, err := buffer.PendingAuditEvents(ctx)
	if err != nil {
		t.Fatalf("pending: %v", err)
	}
	if len(remaining) != 0 {
		t.Fatalf("erwartet leeren puffer nach bestaetigter zustellung, habe %d verbleibende", len(remaining))
	}
}

// Bekannter-Fehler-Praevention: bricht die Zustellung waehrend der
// Nachlieferung erneut ab (Core faellt wieder aus), bleiben die NICHT
// bestaetigten Events sicher im Puffer erhalten statt verloren zu gehen.
func TestFlushAll_KeepsUnconfirmedEventsInBufferOnPartialFailure(t *testing.T) {
	buffer, pool, cleanup := setupTest(t)
	defer cleanup()
	ctx := context.Background()
	registry, clientID, secret := setupModuleCredential(t, pool)

	for i := 0; i < 5; i++ {
		if err := buffer.EnqueueAuditEvent(ctx, "acme", "user-1", fmt.Sprintf("action-%d", i), "", nil); err != nil {
			t.Fatalf("enqueue %d: %v", i, err)
		}
	}

	// Core-Fake schlaegt AB dem 3. Event fehl -> simuliert erneuten Ausfall
	// mitten in der Nachlieferung.
	audit := &fakeAudit{failAt: 3}
	handler := NewHandler(registry, audit, newFakeUsage())
	mux := http.NewServeMux()
	mux.HandleFunc("/internal/resync/audit", handler.AuditHandler)
	mux.HandleFunc("/internal/resync/usage", handler.UsageHandler)
	server := httptest.NewServer(mux)
	defer server.Close()

	client := NewCoreClient(server.URL, clientID, secret)
	worker := NewWorker(buffer, client)

	if err := worker.FlushAll(ctx); err == nil {
		t.Fatal("erwartet fehler, da core nur teilweise bestaetigt hat")
	}

	remaining, err := buffer.PendingAuditEvents(ctx)
	if err != nil {
		t.Fatalf("pending: %v", err)
	}
	if len(remaining) != 2 {
		t.Fatalf("erwartet 2 verbleibende (nicht bestaetigte) events im puffer, habe %d", len(remaining))
	}
}

// Akzeptanzkriterium 3 + Pruefung 2: Nutzungszaehler-Differenz aus der
// Ausfallzeit wird korrekt nachgebucht, ein wiederholter (fehlerhafter)
// Flush-Versuch fuehrt NICHT zu Doppelzaehlung, weil bereits bestaetigte
// Deltas aus dem Puffer entfernt sind.
func TestFlushAll_AppliesUsageDeltasWithoutDoubleCounting(t *testing.T) {
	buffer, pool, cleanup := setupTest(t)
	defer cleanup()
	ctx := context.Background()
	registry, clientID, secret := setupModuleCredential(t, pool)

	deltas := []int64{3, 5, 2}
	var want int64
	for _, d := range deltas {
		want += d
		if err := buffer.EnqueueUsageDelta(ctx, "acme", "api_calls", d); err != nil {
			t.Fatalf("enqueue delta %d: %v", d, err)
		}
	}

	usage := newFakeUsage()
	handler := NewHandler(registry, &fakeAudit{failAt: -1}, usage)
	mux := http.NewServeMux()
	mux.HandleFunc("/internal/resync/audit", handler.AuditHandler)
	mux.HandleFunc("/internal/resync/usage", handler.UsageHandler)
	server := httptest.NewServer(mux)
	defer server.Close()

	client := NewCoreClient(server.URL, clientID, secret)
	worker := NewWorker(buffer, client)

	if err := worker.FlushAll(ctx); err != nil {
		t.Fatalf("flushall 1: %v", err)
	}

	usage.mu.Lock()
	got := usage.totals["acme|api_calls"]
	usage.mu.Unlock()
	if got != want {
		t.Fatalf("nutzungsstand nach nachbuchung = %d, want %d", got, want)
	}

	// Ein zweiter Flush-Versuch (z.B. redundanter Retry) darf NICHTS mehr
	// nachbuchen, da der Puffer bereits geleert wurde.
	if err := worker.FlushAll(ctx); err != nil {
		t.Fatalf("flushall 2: %v", err)
	}
	usage.mu.Lock()
	got2 := usage.totals["acme|api_calls"]
	usage.mu.Unlock()
	if got2 != want {
		t.Fatalf("nutzungsstand nach redundantem zweiten flush = %d, want unveraendert %d (keine doppelzaehlung)", got2, want)
	}
}

// Akzeptanzkriterium 1 + Pruefung 3: bei erkannter Core-Wiedererreichbarkeit
// wird der Cache SOFORT invalidiert (naechster Zugriff refetcht), nicht
// erst nach TTL-Ablauf — Latenz wird gemessen und liegt weit unter einer
// langen TTL.
func TestCheckAndSync_InvalidatesCacheImmediatelyOnRecovery(t *testing.T) {
	buffer, pool, cleanup := setupTest(t)
	defer cleanup()
	registry, clientID, secret := setupModuleCredential(t, pool)
	_ = registry

	mux := http.NewServeMux()
	mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) })
	server := httptest.NewServer(mux)
	defer server.Close()

	client := NewCoreClient(server.URL, clientID, secret)
	worker := NewWorker(buffer, client)

	invalidated := false
	var mu sync.Mutex
	worker.OnReachable(func() {
		mu.Lock()
		invalidated = true
		mu.Unlock()
	})

	start := time.Now()
	becameReachable, err := worker.CheckAndSync(context.Background())
	elapsed := time.Since(start)
	if err != nil {
		t.Fatalf("checkandsync: %v", err)
	}
	if !becameReachable {
		t.Fatal("erwartet erkannten uebergang zu 'erreichbar' beim ersten erfolgreichen check")
	}

	mu.Lock()
	defer mu.Unlock()
	if !invalidated {
		t.Fatal("erwartet sofortigen cache-invalidierungs-callback bei core-wiedererreichbarkeit")
	}
	// Zielwert: deutlich unter einer typischen TTL (z.B. 5s beim
	// Feature-Flag-Cache, LIC-02) — hier im Millisekundenbereich, da rein
	// lokal ohne Netzwerk-Overhead.
	if elapsed > time.Second {
		t.Fatalf("cache-invalidierung brauchte %s, erwartet deutlich unter 1s", elapsed)
	}
}