nexarch/mail/internal/search/reindex_test.go

// Integrationstest (SRC-09): echte Manticore-Instanz, TEST_MANTICORE_URL
// (gleiche Konvention wie integration_test.go/ranking_test.go/facets_test.go).
package search

import (
	"context"
	"errors"
	"sync"
	"sync/atomic"
	"testing"
	"time"
)

// TestRebuild_SearchKeepsWorkingDuringReindex ist die geforderte
// Pflichtprüfung 1: Reindex während laufender Suchanfragen unterbricht die
// Suche nicht.
func TestRebuild_SearchKeepsWorkingDuringReindex(t *testing.T) {
	client := setupClient(t)
	ctx := context.Background()
	tenant := "mandant-src09-parallel"

	for i := 0; i < 30; i++ {
		messageID := "msg-parallel-" + string(rune('a'+i))
		indexFacetDoc(t, client, ctx, tenant, Document{MessageID: messageID, Subject: "Zwiebelfisch " + messageID, Body: "Text"})
	}

	stop := make(chan struct{})
	var searchErrors int64
	var searchesDone int64
	var wg sync.WaitGroup
	wg.Add(1)
	go func() {
		defer wg.Done()
		for {
			select {
			case <-stop:
				return
			default:
			}
			if _, err := client.Search(ctx, tenant, "Zwiebelfisch"); err != nil {
				atomic.AddInt64(&searchErrors, 1)
				t.Logf("suchfehler während reindex: %v", err)
			}
			atomic.AddInt64(&searchesDone, 1)
			time.Sleep(2 * time.Millisecond)
		}
	}()

	reindexer := NewReindexer(client)
	result, err := reindexer.Rebuild(ctx, nil)
	close(stop)
	wg.Wait()

	if err != nil {
		t.Fatalf("rebuild: %v", err)
	}
	if result.OldCount != result.NewCount {
		t.Fatalf("erwartete gleiche trefferzahlen, habe alt=%d neu=%d", result.OldCount, result.NewCount)
	}
	if atomic.LoadInt64(&searchesDone) == 0 {
		t.Fatal("keine einzige parallele suche ausgeführt — test aussagelos")
	}
	if errs := atomic.LoadInt64(&searchErrors); errs != 0 {
		t.Fatalf("erwartete 0 fehlgeschlagene suchen während des reindex, habe %d von %d", errs, atomic.LoadInt64(&searchesDone))
	}

	// Suche funktioniert auch NACH dem Umschalten weiterhin real.
	afterResults, err := client.Search(ctx, tenant, "Zwiebelfisch")
	if err != nil {
		t.Fatalf("search nach reindex: %v", err)
	}
	if len(afterResults) != 30 {
		t.Fatalf("erwartete 30 treffer nach reindex, habe %d", len(afterResults))
	}
}

// TestRebuild_AbortedReindexLeavesNoInconsistentState ist die geforderte
// Pflichtprüfung 2: abgebrochener Reindex hinterlässt keinen
// inkonsistenten Zustand.
func TestRebuild_AbortedReindexLeavesNoInconsistentState(t *testing.T) {
	client := setupClient(t)
	ctx := context.Background()
	tenant := "mandant-src09-abbruch"

	indexFacetDoc(t, client, ctx, tenant, Document{MessageID: "msg-abbruch-1", Subject: "Vertragsentwurf Abbruchtest", Body: "Text"})

	before, err := client.Search(ctx, tenant, "Abbruchtest")
	if err != nil || len(before) != 1 {
		t.Fatalf("voraussetzung nicht erfüllt: %v / %d treffer", err, len(before))
	}

	cancelCtx, cancel := context.WithCancel(ctx)
	cancel() // sofort abgebrochen, simuliert Absturz/Abbruch mitten im Kopiervorgang

	reindexer := NewReindexer(client)
	_, err = reindexer.Rebuild(cancelCtx, nil)
	if err == nil {
		t.Fatal("erwartete fehler bei abgebrochenem kontext, habe nil")
	}
	if !errors.Is(err, context.Canceled) {
		// Manticore-Fehler durch den abgebrochenen Request sind ebenfalls
		// akzeptabel, solange überhaupt ein Fehler zurückkommt.
		t.Logf("fehler war nicht context.Canceled, sondern: %v (akzeptiert, solange real ein fehler zurückkommt)", err)
	}

	// Die lebende Tabelle muss trotz Abbruch unverändert und abfragbar sein.
	after, err := client.Search(ctx, tenant, "Abbruchtest")
	if err != nil {
		t.Fatalf("search nach abgebrochenem reindex: %v", err)
	}
	if len(after) != 1 {
		t.Fatalf("erwartete weiterhin 1 treffer nach abgebrochenem reindex, habe %d — inkonsistenter zustand", len(after))
	}

	// Keine verwaisten Zwischentabellen (kein inkonsistenter Zustand auf
	// Manticore-Ebene): kurz warten, damit ein eventuell noch laufender
	// CREATE-TABLE-Aufruf durchlaufen kann, dann prüfen, dass keine
	// mail_documents_reindex_*-Tabelle übrig geblieben ist.
	time.Sleep(200 * time.Millisecond)
	orphaned := listOrphanedReindexTables(t, client)
	if len(orphaned) > 0 {
		t.Fatalf("verwaiste zwischentabellen nach abbruch gefunden: %v", orphaned)
	}
}

func listOrphanedReindexTables(t *testing.T, client *Client) []string {
	t.Helper()
	// SHOW TABLES ist eine feste, unparametrisierte Anweisung ohne
	// jeglichen Laufzeitwert.
	rows, err := client.showTables(context.Background())
	if err != nil {
		t.Fatalf("show tables: %v", err)
	}
	names := []string{}
	for _, table := range rows {
		if tempTableNamePattern.MatchString(table) {
			names = append(names, table)
		}
	}
	return names
}

// TestRebuild_SampleComparisonMatchesOldAndNewIndex ist die geforderte
// Pflichtprüfung 3: Stichprobenvergleich Alt-/Neuindex bestätigt gleiche
// Trefferzahlen.
func TestRebuild_SampleComparisonMatchesOldAndNewIndex(t *testing.T) {
	client := setupClient(t)
	ctx := context.Background()
	tenant := "mandant-src09-stichprobe"

	subjects := []string{"Quartalsbericht", "Personalplanung", "Urlaubsantrag"}
	for i, s := range subjects {
		indexFacetDoc(t, client, ctx, tenant, Document{MessageID: "msg-sp-" + string(rune('a'+i)), Subject: s, Body: "Inhalt " + s})
	}

	beforeCounts := map[string]int{}
	for _, s := range subjects {
		results, err := client.Search(ctx, tenant, s)
		if err != nil {
			t.Fatalf("search vor reindex (%s): %v", s, err)
		}
		beforeCounts[s] = len(results)
	}

	reindexer := NewReindexer(client)
	if _, err := reindexer.Rebuild(ctx, nil); err != nil {
		t.Fatalf("rebuild: %v", err)
	}

	for _, s := range subjects {
		results, err := client.Search(ctx, tenant, s)
		if err != nil {
			t.Fatalf("search nach reindex (%s): %v", s, err)
		}
		if len(results) != beforeCounts[s] {
			t.Fatalf("stichprobe %q: vor reindex %d treffer, nach reindex %d treffer", s, beforeCounts[s], len(results))
		}
	}
}

// TestRebuild_ReportsProgress deckt Akzeptanzkriterium 2 ab (Fortschritt
// nachvollziehbar sichtbar).
func TestRebuild_ReportsProgress(t *testing.T) {
	client := setupClient(t)
	ctx := context.Background()
	tenant := "mandant-src09-fortschritt"

	for i := 0; i < 5; i++ {
		indexFacetDoc(t, client, ctx, tenant, Document{MessageID: "msg-progress-" + string(rune('a'+i)), Subject: "x"})
	}

	var updates []Progress
	var mu sync.Mutex
	reindexer := NewReindexer(client)
	_, err := reindexer.Rebuild(ctx, func(p Progress) {
		mu.Lock()
		updates = append(updates, p)
		mu.Unlock()
	})
	if err != nil {
		t.Fatalf("rebuild: %v", err)
	}
	if len(updates) == 0 {
		t.Fatal("erwartete mindestens eine fortschrittsmeldung")
	}
	last := updates[len(updates)-1]
	if last.Copied < last.Total {
		// total ist eine zu Beginn eingefrorene Momentaufnahme; die geteilte
		// Manticore-Instanz kann während des Kopierens durch andere Tests
		// weiter wachsen (real beobachtet) — copied darf total daher
		// erreichen oder minimal überschreiten, nur ein Rückstand wäre ein
		// echter Fehler.
		t.Fatalf("letzte fortschrittsmeldung sollte abgeschlossen sein, habe copied=%d total=%d", last.Copied, last.Total)
	}
}