talk/internal/queue/queue_test.go

package queue

import (
	"context"
	"os"
	"path/filepath"
	"slices"
	"testing"
	"time"
)

// gesteuerteSpielerin ist ein testbarer Speaker: Jede Ansage meldet ihren
// Start auf started (Kanal-Puffer: der Worker blockiert nie) und wartet
// dann, bis der Test sie per step freigibt — so ist die Abarbeitung
// vollständig deterministisch steuerbar. Ein abgebrochener ctx beendet das
// Warten (Cancel-/Close-Pfade). Alle Felder werden nur vom Test gelesen;
// started-Reihenfolge == Start-Reihenfolge.
type gesteuerteSpielerin struct {
	started  chan Item
	step     chan struct{}
	consumed int // von warteBis konsumierte Starts (nur Test-Goroutine)
}

func neueSpielerin() *gesteuerteSpielerin {
	return &gesteuerteSpielerin{started: make(chan Item, 16), step: make(chan struct{})}
}

func (g *gesteuerteSpielerin) speak(ctx context.Context, item Item) error {
	g.started <- item
	select {
	case <-g.step:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

// warteBis wartet, bis insgesamt n Ansagen gestartet sind, und liefert die
// seit dem letzten warteBis-Aufruf neu gestarteten.
func (g *gesteuerteSpielerin) warteBis(t *testing.T, n int) []Item {
	t.Helper()
	var items []Item
	deadline := time.After(2 * time.Second)
	for g.consumed < n {
		select {
		case it := <-g.started:
			g.consumed++
			items = append(items, it)
		case <-deadline:
			t.Fatalf("nur %d von %d Ansagen gestartet", g.consumed, n)
		}
	}
	return items
}

// keineWeitere prüft, dass keine weitere Ansage startet (Poll-Fenster).
func (g *gesteuerteSpielerin) keineWeitere(t *testing.T) {
	t.Helper()
	select {
	case it := <-g.started:
		g.consumed++
		t.Fatalf("unerwartete Ansage %q gestartet", it.Text)
	case <-time.After(150 * time.Millisecond):
	}
}

// TestAbarbeitungsReihenfolge: now vor high vor normal vor low, FIFO je
// Level. Der erste Blocker besetzt den Worker; danach werden alle Level
// gleichzeitig eingereiht — die Abarbeitung muss der Klasse folgen.
// (Ohne Gesetze: Null-Laws = unbegrenzt, keine Ruhezeit.)
func TestAbarbeitungsReihenfolge(t *testing.T) {
	g := neueSpielerin()
	q := New(Laws{}, g.speak, nil)
	defer q.Close()

	adm, err := q.Enqueue(Item{Text: "blocker", Priority: PriorityNormal})
	if err != nil {
		t.Fatal(err)
	}
	if adm.ID == "" || adm.Discarded != "" {
		t.Fatalf("Enqueue = %+v, want Aufnahme mit ID", adm)
	}
	g.warteBis(t, 1) // Blocker läuft, Worker gebunden.

	for _, it := range []Item{
		{Text: "niedrig", Priority: PriorityLow},
		{Text: "zweit-niedrig", Priority: PriorityLow},
		{Text: "mittel", Priority: PriorityNormal},
		{Text: "dringend", Priority: PriorityHigh},
		{Text: "sofort", Now: true},
	} {
		if _, err := q.Enqueue(it); err != nil {
			t.Fatal(err)
		}
	}
	g.keineWeitere(t) // Serialisierung: Worker spielt noch den Blocker.

	// Freigeben: alles Übrige läuft in Klassen-Reihenfolge durch.
	close(g.step)
	rest := g.warteBis(t, 6)

	var texts []string
	for _, it := range rest {
		texts = append(texts, it.Text)
	}
	want := []string{"sofort", "dringend", "mittel", "niedrig", "zweit-niedrig"}
	if !slices.Equal(texts, want) {
		t.Errorf("Abarbeitung nach dem Blocker = %v, want %v", texts, want)
	}
}

// TestCurrentUndPending: Status-Sicht auf laufendes Item + Warteschlange.
func TestCurrentUndPending(t *testing.T) {
	g := neueSpielerin()
	q := New(Laws{}, g.speak, nil)
	defer q.Close()

	if _, ok := q.Current(); ok {
		t.Fatal("Current vor erster Ansage: want idle")
	}

	q.Enqueue(Item{Text: "blocker"})
	g.warteBis(t, 1)

	q.Enqueue(Item{Text: "hoch", Priority: PriorityHigh})
	q.Enqueue(Item{Text: "niedrig", Priority: PriorityLow})

	if cur, ok := q.Current(); !ok || cur.Text != "blocker" {
		t.Errorf("Current = %+v (ok=%v), will Blocker", cur, ok)
	}
	var pendingTexts []string
	for _, it := range q.Pending() {
		pendingTexts = append(pendingTexts, it.Text)
	}
	if !slices.Equal(pendingTexts, []string{"hoch", "niedrig"}) {
		t.Errorf("Pending = %v, want [hoch niedrig]", pendingTexts)
	}

	close(g.step)
	g.warteBis(t, 3)
}

// TestCancel: laufende Ansage wird über ctx abgebrochen, Queue geleert,
// Gesamtzahl zurückgemeldet; danach ist die Queue wieder aufnahme- und
// spielbereit.
func TestCancel(t *testing.T) {
	g := neueSpielerin()
	q := New(Laws{}, g.speak, nil)
	defer q.Close()

	q.Enqueue(Item{Text: "blocker"})
	g.warteBis(t, 1)

	q.Enqueue(Item{Text: "wartet-eins"})
	q.Enqueue(Item{Text: "wartet-zwei"})

	if n := q.Cancel(); n != 3 {
		t.Errorf("Cancel = %d, want 3 (laufende + 2 wartende)", n)
	}

	g.keineWeitere(t) // Blocker durch ctx beendet, wartende Items sind weg.
	waitIdle(t, q)

	if n := q.Cancel(); n != 0 {
		t.Errorf("Cancel auf leerer Queue = %d, want 0", n)
	}

	// Queue nimmt wieder an und spielt:
	adm, err := q.Enqueue(Item{Text: "danach"})
	if err != nil {
		t.Fatal(err)
	}
	g.step <- struct{}{} // gibt die wartende Ansage im Speaker frei
	gestartet := g.warteBis(t, 2)
	if len(gestartet) != 1 || gestartet[0].ID != adm.ID {
		t.Errorf("neue Ansage nicht gestartet: %+v", gestartet)
	}
}

// TestClose: wartende Items werden verworfen, die laufende Ansage über den
// abgebrochenen ctx beendet, Close wartet auf den Worker-Exit, danach ist
// Enqueue ein Fehler (und Close idempotent).
func TestClose(t *testing.T) {
	g := neueSpielerin()
	q := New(Laws{}, g.speak, nil)

	q.Enqueue(Item{Text: "blocker"})
	g.warteBis(t, 1)
	q.Enqueue(Item{Text: "wartet"})

	q.Close()

	g.keineWeitere(t)
	if _, ok := q.Current(); ok {
		t.Error("Current nach Close: want idle")
	}
	if _, err := q.Enqueue(Item{Text: "zu spät"}); err == nil {
		t.Error("Enqueue nach Close: want Fehler")
	}
	q.Close() // idempotent
}

// TestEnqueueVergibtIDs: IDs sind kurz und eindeutig.
func TestEnqueueVergibtIDs(t *testing.T) {
	g := neueSpielerin()
	q := New(Laws{}, g.speak, nil)
	defer q.Close()

	seen := map[string]bool{}
	for i := 0; i < 20; i++ {
		adm, err := q.Enqueue(Item{Text: "x"})
		if err != nil {
			t.Fatal(err)
		}
		if len(adm.ID) < 4 || seen[adm.ID] {
			t.Errorf("ID %q zu kurz oder doppelt", adm.ID)
		}
		seen[adm.ID] = true
	}
}

// waitIdle pollt, bis kein Item mehr läuft (Cancel-/Close-Pfade geben das
// nach dem Worker-Exit synchron frei; das Pollen deckt die Übergangsphase
// ab, ohne auf Timing zu trauen).
func waitIdle(t *testing.T, q *Queue) {
	t.Helper()
	deadline := time.Now().Add(time.Second)
	for {
		if _, ok := q.Current(); !ok {
			return
		}
		if time.Now().After(deadline) {
			t.Fatal("Queue wird nicht idle")
		}
		time.Sleep(5 * time.Millisecond)
	}
}

// TestMuteStore: Datei-Roundtrip, Fehlend-Heißt-Nicht-Stumm, atomares Set.
func TestMuteStore(t *testing.T) {
	dir := t.TempDir()
	path := filepath.Join(dir, MuteFileName)

	var s MuteStore
	if s.Muted() {
		t.Error("MuteStore ohne Pfad: Muted() = true, want false")
	}
	if err := s.Set(true); err == nil {
		t.Error("Set ohne Pfad: want Fehler")
	}

	s = MuteStore{Path: path}
	if s.Muted() {
		t.Error("ohne Datei: Muted() = true, want false (nie still aus Versehen)")
	}

	if err := s.Set(true); err != nil {
		t.Fatal(err)
	}
	if !s.Muted() {
		t.Error("nach Set(true): Muted() = false")
	}
	if b, _ := os.ReadFile(path); string(b) != "true\n" {
		t.Errorf("Datei-Inhalt = %q, want %q", b, "true\n")
	}

	if err := s.Set(false); err != nil {
		t.Fatal(err)
	}
	if s.Muted() {
		t.Error("nach Set(false): Muted() = true")
	}

	// Müll in der Datei heißt: nicht stumm (nur „true" ist stumm).
	if err := os.WriteFile(path, []byte("vielleicht\n"), 0o644); err != nil {
		t.Fatal(err)
	}
	if s.Muted() {
		t.Error("Datei-Inhalt „vielleicht“: Muted() = true, want false")
	}
}

// TestRateLimiter: Struktur-Semantik (seit Task 05 in den Loop verwired).
func TestRateLimiter(t *testing.T) {
	// Unbegrenzt (Task-04-Default): Interval 0 erlaubt immer.
	unlimited := &RateLimiter{}
	for i := 0; i < 3; i++ {
		if !unlimited.Allow() {
			t.Fatal("Interval 0: Allow = false, want true (unbegrenzt)")
		}
		unlimited.Reserve()
	}

	// Begrenzt: nach Reserve ist erst nach Interval wieder Allow.
	l := &RateLimiter{Interval: 30 * time.Millisecond}
	if !l.Allow() {
		t.Fatal("frisches Limit: Allow = false")
	}
	l.Reserve()
	if l.Allow() {
		t.Fatal("sofort nach Reserve: Allow = true, want false")
	}
	if next := l.Next(); time.Now().After(next) {
		t.Errorf("Next() = %v liegt in der Vergangenheit", next)
	}
	time.Sleep(40 * time.Millisecond)
	if !l.Allow() {
		t.Fatal("nach Interval: Allow = false, want true")
	}
}

// fensterUm baut ein "HH:MM-HH:MM"-Ruhefenster, das t sicher einschließt
// (eine Minute davor bis eine Minute danach; Minute-Auflösung reicht).
func fensterUm(t time.Time) string {
	return t.Add(-time.Minute).Format("15:04") + "-" + t.Add(time.Minute).Format("15:04")
}

// warteAuf verifies mit Pollen eine Bedingung (Zeit-Gesetze sind real, der
// gesteuerte Speaker allein sieht sie nicht).
func warteAuf(t *testing.T, timeout time.Duration, cond func() bool, was string) {
	t.Helper()
	deadline := time.Now().Add(timeout)
	for !cond() {
		if time.Now().After(deadline) {
			t.Fatalf("%s innerhalb %s nicht eingetreten", was, timeout)
		}
		time.Sleep(5 * time.Millisecond)
	}
}

// TestRateLimitLowVerworfenNormalParkt: mit kurzem Fenster spielt die
// erste Ansage sofort; low über dem Budget wird schon beim Eintritt
// verworfen (Admission nennt den Grund), normal parkt und spielt nach
// Fenster-Ablauf.
func TestRateLimitLowVerworfenNormalParkt(t *testing.T) {
	g := neueSpielerin()
	q := New(Laws{RateLimit: 400 * time.Millisecond}, g.speak, nil)
	defer q.Close()

	if adm, err := q.Enqueue(Item{Text: "erste"}); err != nil || adm.ID == "" || adm.Discarded != "" {
		t.Fatalf("erste Enqueue = %+v (%v), want Aufnahme", adm, err)
	}
	g.warteBis(t, 1) // erste läuft — Fenster ist gebucht.

	adm, err := q.Enqueue(Item{Text: "low-eins", Priority: PriorityLow})
	if err != nil {
		t.Fatal(err)
	}
	if adm.ID != "" || adm.Discarded != DiscardRateLimit || adm.Toast {
		t.Errorf("low über Budget = %+v, want Verwurf %q", adm, DiscardRateLimit)
	}
	if adm, err = q.Enqueue(Item{Text: "low-zwei", Priority: PriorityLow}); err != nil || adm.Discarded != DiscardRateLimit {
		t.Errorf("zweiter low-Verwurf = %+v (%v)", adm, err)
	}

	adm, err = q.Enqueue(Item{Text: "zweite"})
	if err != nil || adm.ID == "" || adm.Discarded != "" {
		t.Fatalf("normal = %+v (%v), want Aufnahme (parkt bis zum Slot)", adm, err)
	}

	// Status-Sicht während der erste läuft: zweite wartet (normal, geparkt).
	warteAuf(t, time.Second, func() bool {
		for _, it := range q.Pending() {
			if it.Text == "zweite" {
				return true
			}
		}
		return false
	}, "geparkte normale Ansage in Pending")

	close(g.step) // erste beenden — zweite muss weiter parken …
	g.keineWeitere(t)
	// … und nach Fenster-Ablauf starten (Deadline deckt Jitter locker).
	gestartet := g.warteBis(t, 2)
	if len(gestartet) != 1 || gestartet[0].Text != "zweite" {
		t.Errorf("nach Fenster-Ablauf = %+v, will zweite", gestartet)
	}
	// low-eins/low-zwei spielen nie: genau 2 Starts insgesamt.
	g.keineWeitere(t)
}

// TestHighBurstEinUndAus: mit Burst-Recht startet high sofort nach der
// laufenden Ansage, obwohl das Fenster voll ist; ohne Recht parkt high
// wie normal (kein Verstoß).
func TestHighBurstEinUndAus(t *testing.T) {
	g := neueSpielerin()
	q := New(Laws{RateLimit: 10 * time.Second, AllowHighBurst: true}, g.speak, nil)

	q.Enqueue(Item{Text: "erste"})
	g.warteBis(t, 1)
	if adm, err := q.Enqueue(Item{Text: "dringend", Priority: PriorityHigh}); err != nil || adm.ID == "" {
		t.Fatalf("high mit Burst-Recht = %+v (%v), want Aufnahme", adm, err)
	}
	g.step <- struct{}{}          // erste beenden
	gestartet := g.warteBis(t, 2) // high startet trotz vollem Fenster
	if len(gestartet) != 1 || gestartet[0].Text != "dringend" {
		t.Errorf("burst = %+v, will dringend sofort", gestartet)
	}
	q.Close()

	g2 := neueSpielerin()
	q2 := New(Laws{RateLimit: 10 * time.Second, AllowHighBurst: false}, g2.speak, nil)
	defer q2.Close()

	q2.Enqueue(Item{Text: "erste"})
	g2.warteBis(t, 1) // erste läuft — Fenster gebucht
	if adm, err := q2.Enqueue(Item{Text: "dringend", Priority: PriorityHigh}); err != nil || adm.ID == "" {
		t.Fatalf("high ohne Burst-Recht = %+v (%v), want Aufnahme (parkt)", adm, err)
	}
	g2.step <- struct{}{} // erste beenden
	g2.keineWeitere(t)    // high wartet den Slot ab — kein Verstoß ohne Recht
}

// TestNowBypassesRateLimit: now startet trotz vollem Fensters sofort —
// der dokumentierte Ausbruch (er startet trotzdem ein neues Fenster).
func TestNowBypassesRateLimit(t *testing.T) {
	g := neueSpielerin()
	q := New(Laws{RateLimit: 10 * time.Second}, g.speak, nil)
	defer q.Close()

	q.Enqueue(Item{Text: "erste"})
	g.warteBis(t, 1) // Fenster gebucht (10 s)

	if adm, err := q.Enqueue(Item{Text: "sofort", Now: true}); err != nil || adm.ID == "" || adm.Discarded != "" {
		t.Fatalf("now über vollem Fenster = %+v (%v), want Aufnahme", adm, err)
	}
	g.step <- struct{}{}
	gestartet := g.warteBis(t, 2) // lange vor Fenster-Ablauf (10 s) — sonst Timeout
	if len(gestartet) != 1 || gestartet[0].Text != "sofort" {
		t.Errorf("now = %+v, will sofort", gestartet)
	}

	// Der now-Bruch hat ein neues Fenster gebucht: normal parkt jetzt.
	if adm, err := q.Enqueue(Item{Text: "danach"}); err != nil || adm.ID == "" {
		t.Fatalf("normal nach now = %+v (%v)", adm, err)
	}
	close(g.step)
	g.keineWeitere(t) // parkt bis zum Slot — nicht sofort
}

// TestRuhezeitDiscardUndToast: aktives Fenster verwirft normal und high
// (Burst ist ein Rate-Recht, kein Ruhe-Recht) — toast ersetzt stattdessen;
// now bricht bewusst aus.
func TestRuhezeitDiscardUndToast(t *testing.T) {
	fenster := fensterUm(time.Now())

	g := neueSpielerin()
	q := New(Laws{Quiet: mustQuiet(t, fenster), QuietMode: QuietModeDiscard}, g.speak, nil)
	defer q.Close()

	adm, err := q.Enqueue(Item{Text: "normal"})
	if err != nil || adm.ID != "" || adm.Discarded != DiscardQuietHours || adm.Toast {
		t.Errorf("normal in Ruhezeit = %+v (%v), want stiller Verwurf", adm, err)
	}
	if adm, err = q.Enqueue(Item{Text: "dringend", Priority: PriorityHigh}); err != nil || adm.Discarded != DiscardQuietHours {
		t.Errorf("high in Ruhezeit = %+v (%v), want auch Verwurf", adm, err)
	}
	adm, err = q.Enqueue(Item{Text: "sofort", Now: true})
	if err != nil || adm.ID == "" || adm.Discarded != "" {
		t.Fatalf("now in Ruhezeit = %+v (%v), want Ausbruch", adm, err)
	}
	gestartet := g.warteBis(t, 1)
	if len(gestartet) != 1 || gestartet[0].Text != "sofort" {
		t.Errorf("in Ruhezeit = %+v, will nur sofort", gestartet)
	}
	g.keineWeitere(t)

	g2 := neueSpielerin()
	q2 := New(Laws{Quiet: mustQuiet(t, fenster), QuietMode: QuietModeToast}, g2.speak, nil)
	defer q2.Close()

	adm, err = q2.Enqueue(Item{Text: "normal"})
	if err != nil || adm.Discarded != DiscardQuietHours || !adm.Toast {
		t.Errorf("normal in Ruhezeit (toast) = %+v (%v), want Toast-Ersetzung", adm, err)
	}
	g2.keineWeitere(t)
}

// mustQuiet parst ein Ruhefenster oder bricht den Test ab.
func mustQuiet(t *testing.T, fenster string) QuietHours {
	t.Helper()
	q, err := ParseQuietHours(fenster)
	if err != nil {
		t.Fatal(err)
	}
	return q
}

// TestLawState: unbegrenzt liefert freien Slot; nach gebuchtem Fenster
// liegt der nächste Slot in der Zukunft; Ruhezeit-Fenster ist sichtbar.
func TestLawState(t *testing.T) {
	g := neueSpielerin()
	q := New(Laws{}, g.speak, nil)
	st := q.LawState()
	if !st.NextSlot.IsZero() || st.RateLimit != 0 {
		t.Errorf("unbegrenzt = %+v, want Nullzeit-Slot", st)
	}
	q.Close()

	// Das Ruhezeit-Fenster muss JETZT inaktiv sein, sonst würde die Ansage
	// unten getoastet statt gesprochen — ein festes Fenster wäre eine
	// Zeitbombe (rot zu bestimmten Tageszeiten). Deshalb dynamisch: eine
	// Stunde, die zwei Stunden in der Zukunft beginnt.
	window := inactiveQuietWindow(time.Now())
	fenster := mustQuiet(t, window)
	if fenster.Active(time.Now()) {
		t.Fatalf("Testfenster %s ist aktiv — darf nicht sein", window)
	}
	q2 := New(Laws{RateLimit: time.Hour, AllowHighBurst: true, Quiet: fenster, QuietMode: QuietModeToast}, g.speak, nil)
	defer q2.Close()
	st2 := q2.LawState()
	if st2.RateLimit != time.Hour || !st2.AllowHighBurst {
		t.Errorf("LawState = %+v, want 1h-Fenster mit Burst-Recht", st2)
	}
	if st2.Quiet.String() != window || st2.QuietMode != QuietModeToast {
		t.Errorf("Ruhezeit in LawState = %q/%q, want %q/toast", st2.Quiet, st2.QuietMode, window)
	}
	if st2.QuietActive || !st2.QuietUntil.IsZero() {
		t.Errorf("Ruhezeit inaktiv erwartet, LawState meldet active=%v until=%v", st2.QuietActive, st2.QuietUntil)
	}

	q2.Enqueue(Item{Text: "erste"})
	g.warteBis(t, 1)
	if next := q2.LawState().NextSlot; !next.After(time.Now()) {
		t.Errorf("nach Start = %v, want Slot in der Zukunft", next)
	}
	g.step <- struct{}{}
}

// inactiveQuietWindow liefert ein "HH:MM-HH:MM"-Fenster von einer Stunde,
// das zwei Stunden nach now beginnt — es enthält now garantiert nicht, egal
// zu welcher Tageszeit der Test läuft (auch über Mitternacht hinweg).
func inactiveQuietWindow(now time.Time) string {
	start := now.Add(2 * time.Hour)
	end := now.Add(3 * time.Hour)
	return start.Format("15:04") + "-" + end.Format("15:04")
}

// TestInactiveQuietWindow: Das Test-Hilfsfenster darf zu KEINER Tageszeit
// aktiv sein — sonst wäre TestLawState wieder eine Zeitbombe.
func TestInactiveQuietWindow(t *testing.T) {
	base := time.Date(2026, 10, 24, 0, 0, 0, 0, time.Local)
	for m := 0; m < 24*60; m += 7 {
		now := base.Add(time.Duration(m) * time.Minute)
		w := inactiveQuietWindow(now)
		q, err := ParseQuietHours(w)
		if err != nil {
			t.Fatalf("%s: %v", now.Format("15:04"), err)
		}
		if q.Active(now) {
			t.Errorf("um %s ist Fenster %s aktiv", now.Format("15:04"), w)
		}
	}
}