nexarch/archive/internal/retentionnotify/retentionnotify_test.go

package retentionnotify

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

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

	"gitea.perlbach24.de/scripte/nexarch/archive/internal/notifyclient"
	"gitea.perlbach24.de/scripte/nexarch/archive/internal/retentionengine"
)

func requireTestPool(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, err := pgxpool.New(ctx, dsn)
	if err != nil {
		t.Fatalf("pool: %v", err)
	}
	t.Cleanup(func() { pool.Close() })

	if _, err := pool.Exec(ctx, `
		CREATE EXTENSION IF NOT EXISTS pgcrypto;
		CREATE TABLE IF NOT EXISTS retention_objects (
			id UUID PRIMARY KEY DEFAULT gen_random_uuid(), object_type TEXT NOT NULL,
			object_reference TEXT NOT NULL,
			status TEXT NOT NULL DEFAULT 'active' CHECK (status IN ('active', 'expired', 'deleted')),
			created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
			UNIQUE (object_type, object_reference)
		);
		CREATE TABLE IF NOT EXISTS retention_class_assignments (
			id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
			retention_object_id UUID NOT NULL REFERENCES retention_objects(id) ON DELETE CASCADE,
			retention_class TEXT NOT NULL, assigned_at TIMESTAMPTZ NOT NULL DEFAULT now()
		);
		CREATE TABLE IF NOT EXISTS retention_class_rules (
			retention_class TEXT PRIMARY KEY, duration INTERVAL NOT NULL,
			active BOOLEAN NOT NULL DEFAULT true
		);
		ALTER TABLE retention_class_rules ADD COLUMN IF NOT EXISTS notify_lead_days INT NOT NULL DEFAULT 7;
		ALTER TABLE retention_class_rules ADD COLUMN IF NOT EXISTS notify_enabled BOOLEAN NOT NULL DEFAULT true;
		CREATE TABLE IF NOT EXISTS retention_notifications (
			retention_object_id UUID PRIMARY KEY REFERENCES retention_objects(id) ON DELETE CASCADE,
			notified_at TIMESTAMPTZ NOT NULL DEFAULT now()
		);
	`); err != nil {
		t.Fatalf("schema: %v", err)
	}
	t.Cleanup(func() {
		_, _ = pool.Exec(context.Background(), `TRUNCATE retention_notifications, retention_class_assignments, retention_objects CASCADE; TRUNCATE retention_class_rules`)
	})
	return pool
}

// fakeCFG05Server zaehlt Aufrufe und liefert eine feste Antwort - simuliert
// den echten CFG-05-Endpunkt, ohne das Core-Modul einzubinden (Archive kann
// es nicht direkt importieren, siehe notifyclient).
func fakeCFG05Server(t *testing.T, fail bool) (*notifyclient.Client, *int32) {
	t.Helper()
	var calls int32
	mux := http.NewServeMux()
	mux.HandleFunc("POST /notify/enqueue", func(w http.ResponseWriter, r *http.Request) {
		atomic.AddInt32(&calls, 1)
		if fail {
			http.Error(w, "simulierter zustellfehler", http.StatusInternalServerError)
			return
		}
		w.Header().Set("Content-Type", "application/json")
		_ = json.NewEncoder(w).Encode(map[string]any{"job_id": "fake-job-id", "skipped": false})
	})
	server := httptest.NewServer(mux)
	t.Cleanup(server.Close)
	return notifyclient.New(server.URL, "test-token"), &calls
}

// insertDueObject legt ein Objekt an, dessen Stichtag (assigned_at +
// Klassen-Frist) genau dueIn ab jetzt liegt — die Subtraktion der Frist
// erfolgt IN Postgres (dieselbe INTERVAL-Arithmetik wie ComputeDueDate),
// keine eigene Kalenderrechnung in Go.
func insertDueObject(t *testing.T, ctx context.Context, pool *pgxpool.Pool, objectRef, class string, dueIn time.Duration) string {
	t.Helper()
	var objID string
	if err := pool.QueryRow(ctx, `INSERT INTO retention_objects (object_type, object_reference) VALUES ('dms_document', $1) RETURNING id`, objectRef).Scan(&objID); err != nil {
		t.Fatalf("objekt anlegen: %v", err)
	}
	desiredDue := time.Now().UTC().Add(dueIn)
	if _, err := pool.Exec(ctx, `
		INSERT INTO retention_class_assignments (retention_object_id, retention_class, assigned_at)
		SELECT $1, $2, $3::timestamptz - r.duration FROM retention_class_rules r WHERE r.retention_class = $2
	`, objID, class, desiredDue); err != nil {
		t.Fatalf("zuordnung anlegen: %v", err)
	}
	return objID
}

var testRecipient = Recipient{TenantSlug: "acme", UserID: "tenant-admin", Email: "admin@acme.example"}

// TestRun_ShortLeadTimeTriggersExactlyOneNotification ist die geforderte
// Pflichtpruefung 1.
func TestRun_ShortLeadTimeTriggersExactlyOneNotification(t *testing.T) {
	pool := requireTestPool(t)
	ctx := context.Background()

	if err := retentionengine.ConfigureClassRule(ctx, pool, "klasse-kurz", "1 day"); err != nil {
		t.Fatal(err)
	}
	if _, err := pool.Exec(ctx, `UPDATE retention_class_rules SET notify_lead_days = 1, notify_enabled = true WHERE retention_class = 'klasse-kurz'`); err != nil {
		t.Fatal(err)
	}
	insertDueObject(t, ctx, pool, "kurz-doc", "klasse-kurz", 12*time.Hour)

	client, calls := fakeCFG05Server(t, false)
	results, err := Run(ctx, pool, client, time.Now().UTC(), testRecipient)
	if err != nil {
		t.Fatalf("run: %v", err)
	}
	if len(results) != 1 || results[0].Err != nil {
		t.Fatalf("erwartet genau ein ergebnis ohne fehler, habe: %+v", results)
	}
	if atomic.LoadInt32(calls) != 1 {
		t.Fatalf("erwartet genau einen cfg-05-aufruf, habe %d", *calls)
	}
}

// TestRun_DisabledNotificationSendsNothing ist die geforderte
// Pflichtpruefung 2.
func TestRun_DisabledNotificationSendsNothing(t *testing.T) {
	pool := requireTestPool(t)
	ctx := context.Background()

	if err := retentionengine.ConfigureClassRule(ctx, pool, "klasse-deaktiviert", "1 day"); err != nil {
		t.Fatal(err)
	}
	if _, err := pool.Exec(ctx, `UPDATE retention_class_rules SET notify_lead_days = 1, notify_enabled = false WHERE retention_class = 'klasse-deaktiviert'`); err != nil {
		t.Fatal(err)
	}
	insertDueObject(t, ctx, pool, "deaktiviert-doc", "klasse-deaktiviert", 12*time.Hour)

	client, calls := fakeCFG05Server(t, false)
	results, err := Run(ctx, pool, client, time.Now().UTC(), testRecipient)
	if err != nil {
		t.Fatalf("run: %v", err)
	}
	if len(results) != 0 {
		t.Fatalf("erwartet keine benachrichtigung bei deaktivierter klasse, habe: %+v", results)
	}
	if atomic.LoadInt32(calls) != 0 {
		t.Fatalf("erwartet keinen cfg-05-aufruf, habe %d", *calls)
	}
}

// TestRun_FailedDeliveryIsReportedNotSwallowed ist die geforderte
// Pflichtpruefung 3.
func TestRun_FailedDeliveryIsReportedNotSwallowed(t *testing.T) {
	pool := requireTestPool(t)
	ctx := context.Background()

	if err := retentionengine.ConfigureClassRule(ctx, pool, "klasse-fehler", "1 day"); err != nil {
		t.Fatal(err)
	}
	if _, err := pool.Exec(ctx, `UPDATE retention_class_rules SET notify_lead_days = 1, notify_enabled = true WHERE retention_class = 'klasse-fehler'`); err != nil {
		t.Fatal(err)
	}
	objID := insertDueObject(t, ctx, pool, "fehler-doc", "klasse-fehler", 12*time.Hour)

	client, _ := fakeCFG05Server(t, true)
	results, err := Run(ctx, pool, client, time.Now().UTC(), testRecipient)
	if err != nil {
		t.Fatalf("run: %v", err)
	}
	if len(results) != 1 || results[0].Err == nil {
		t.Fatalf("erwartet ein ergebnis MIT protokolliertem fehler, habe: %+v", results)
	}

	var count int
	if err := pool.QueryRow(ctx, `SELECT count(*) FROM retention_notifications WHERE retention_object_id = $1`, objID).Scan(&count); err != nil {
		t.Fatal(err)
	}
	if count != 0 {
		t.Fatalf("fehlgeschlagener versand darf NICHT als benachrichtigt markiert werden (sonst kein retry)")
	}
}

// TestRun_PreventsDoubleNotificationAcrossRuns ist Akzeptanzkriterium 2 -
// Mehrfachversand wird verhindert, auch nach einem simulierten Neustart
// (zweiter Run() mit frischem Aufruf, kein In-Memory-Zustand zwischen den
// Durchlaeufen).
func TestRun_PreventsDoubleNotificationAcrossRuns(t *testing.T) {
	pool := requireTestPool(t)
	ctx := context.Background()

	if err := retentionengine.ConfigureClassRule(ctx, pool, "klasse-doppelt", "1 day"); err != nil {
		t.Fatal(err)
	}
	if _, err := pool.Exec(ctx, `UPDATE retention_class_rules SET notify_lead_days = 1, notify_enabled = true WHERE retention_class = 'klasse-doppelt'`); err != nil {
		t.Fatal(err)
	}
	insertDueObject(t, ctx, pool, "doppelt-doc", "klasse-doppelt", 12*time.Hour)

	client, calls := fakeCFG05Server(t, false)
	now := time.Now().UTC()

	if _, err := Run(ctx, pool, client, now, testRecipient); err != nil {
		t.Fatalf("erster run: %v", err)
	}
	if _, err := Run(ctx, pool, client, now, testRecipient); err != nil {
		t.Fatalf("zweiter run (simulierter neustart): %v", err)
	}

	if atomic.LoadInt32(calls) != 1 {
		t.Fatalf("erwartet genau einen cfg-05-aufruf ueber beide durchlaeufe, habe %d", *calls)
	}
}