talk/internal/queue/queue_test.go

package queue

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

// controlledSpeaker is a testable Speaker: every announcement reports its
// start on started (buffered channel: the worker never blocks) and then
// waits until the test releases it via step, so processing is fully
// deterministic. A cancelled ctx ends the wait (Cancel/Close paths). All
// fields are read only by the test; started order == start order.
type controlledSpeaker struct {
	started  chan Item
	step     chan struct{}
	consumed int // starts consumed by waitUntil (test goroutine only)
}

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

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

// waitUntil waits until n announcements have started in total and returns
// the ones started since the previous waitUntil call.
func (g *controlledSpeaker) waitUntil(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("only %d of %d announcements started", g.consumed, n)
		}
	}
	return items
}

// noFurther checks that no further announcement starts (poll window).
func (g *controlledSpeaker) noFurther(t *testing.T) {
	t.Helper()
	select {
	case it := <-g.started:
		g.consumed++
		t.Fatalf("unexpected announcement %q started", it.Text)
	case <-time.After(150 * time.Millisecond):
	}
}

// TestProcessingOrder: now before high before normal before low, FIFO per
// level. The first blocker occupies the worker; then all levels are
// enqueued at once and processing must follow the class.
// (No laws: zero Laws = unlimited, no quiet hours.)
func TestProcessingOrder(t *testing.T) {
	g := newSpeaker()
	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 admission with ID", adm)
	}
	g.waitUntil(t, 1) // blocker is playing, worker is busy

	for _, it := range []Item{
		{Text: "low", Priority: PriorityLow},
		{Text: "second-low", Priority: PriorityLow},
		{Text: "normal", Priority: PriorityNormal},
		{Text: "urgent", Priority: PriorityHigh},
		{Text: "right-now", Now: true},
	} {
		if _, err := q.Enqueue(it); err != nil {
			t.Fatal(err)
		}
	}
	g.noFurther(t) // serialization: the worker is still on the blocker

	// Release: everything else runs through in class order.
	close(g.step)
	rest := g.waitUntil(t, 6)

	var texts []string
	for _, it := range rest {
		texts = append(texts, it.Text)
	}
	want := []string{"right-now", "urgent", "normal", "low", "second-low"}
	if !slices.Equal(texts, want) {
		t.Errorf("processing after the blocker = %v, want %v", texts, want)
	}
}

// TestCurrentAndPending: status view of the playing item and the queue.
func TestCurrentAndPending(t *testing.T) {
	g := newSpeaker()
	q := New(Laws{}, g.speak, nil)
	defer q.Close()

	if _, ok := q.Current(); ok {
		t.Fatal("Current before the first announcement: want idle")
	}

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

	q.Enqueue(Item{Text: "high", Priority: PriorityHigh})
	q.Enqueue(Item{Text: "low", Priority: PriorityLow})

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

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

// TestCancel: the playing announcement is aborted via ctx, the queue
// emptied, the total reported; afterwards the queue admits and plays again.
func TestCancel(t *testing.T) {
	g := newSpeaker()
	q := New(Laws{}, g.speak, nil)
	defer q.Close()

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

	q.Enqueue(Item{Text: "waiting-one"})
	q.Enqueue(Item{Text: "waiting-two"})

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

	g.noFurther(t) // blocker ended via ctx, waiting items are gone
	waitIdle(t, q)

	if n := q.Cancel(); n != 0 {
		t.Errorf("Cancel on empty queue = %d, want 0", n)
	}

	// The queue admits and plays again:
	adm, err := q.Enqueue(Item{Text: "afterwards"})
	if err != nil {
		t.Fatal(err)
	}
	g.step <- struct{}{} // releases the waiting announcement in the speaker
	started := g.waitUntil(t, 2)
	if len(started) != 1 || started[0].ID != adm.ID {
		t.Errorf("new announcement not started: %+v", started)
	}
}

// TestClose: waiting items are dropped, the playing announcement ends via
// the cancelled ctx, Close waits for the worker to exit, afterwards Enqueue
// is an error (and Close idempotent).
func TestClose(t *testing.T) {
	g := newSpeaker()
	q := New(Laws{}, g.speak, nil)

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

	q.Close()

	g.noFurther(t)
	if _, ok := q.Current(); ok {
		t.Error("Current after Close: want idle")
	}
	if _, err := q.Enqueue(Item{Text: "too late"}); err == nil {
		t.Error("Enqueue after Close: want error")
	}
	q.Close() // idempotent
}

// TestEnqueueAssignsIDs: IDs are short and unique.
func TestEnqueueAssignsIDs(t *testing.T) {
	g := newSpeaker()
	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 too short or duplicate", adm.ID)
		}
		seen[adm.ID] = true
	}
}

// waitIdle polls until no item is playing (Cancel/Close release that
// synchronously after the worker exits; polling covers the transition
// without trusting timing).
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 does not become idle")
		}
		time.Sleep(5 * time.Millisecond)
	}
}

// TestMuteStore: file round trip, missing means not muted, atomic Set.
func TestMuteStore(t *testing.T) {
	dir := t.TempDir()
	path := filepath.Join(dir, MuteFileName)

	var s MuteStore
	if s.Muted() {
		t.Error("MuteStore without path: Muted() = true, want false")
	}
	if err := s.Set(true); err == nil {
		t.Error("Set without path: want error")
	}

	s = MuteStore{Path: path}
	if s.Muted() {
		t.Error("without file: Muted() = true, want false (never silent by accident)")
	}

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

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

	// Garbage in the file means not muted (only "true" is muted).
	if err := os.WriteFile(path, []byte("maybe\n"), 0o644); err != nil {
		t.Fatal(err)
	}
	if s.Muted() {
		t.Error("file content \"maybe\": Muted() = true, want false")
	}
}

// TestRateLimiter: the limiter's own semantics.
func TestRateLimiter(t *testing.T) {
	// Unlimited: interval 0 always allows.
	unlimited := &RateLimiter{}
	for i := 0; i < 3; i++ {
		if !unlimited.Allow() {
			t.Fatal("interval 0: Allow = false, want true (unlimited)")
		}
		unlimited.Reserve()
	}

	// Limited: after Reserve, Allow only after the interval.
	l := &RateLimiter{Interval: 30 * time.Millisecond}
	if !l.Allow() {
		t.Fatal("fresh limiter: Allow = false")
	}
	l.Reserve()
	if l.Allow() {
		t.Fatal("right after Reserve: Allow = true, want false")
	}
	if next := l.Next(); time.Now().After(next) {
		t.Errorf("Next() = %v lies in the past", next)
	}
	time.Sleep(40 * time.Millisecond)
	if !l.Allow() {
		t.Fatal("after the interval: Allow = false, want true")
	}
}

// windowAround builds an "HH:MM-HH:MM" quiet window that safely contains t
// (three minutes before to four minutes after; minute resolution).
func windowAround(t time.Time) string {
	return t.Add(-3*time.Minute).Format("15:04") + "-" + t.Add(4*time.Minute).Format("15:04")
}

// waitForCond polls a condition (time laws are real; the controlled speaker
// alone cannot see them).
func waitForCond(t *testing.T, timeout time.Duration, cond func() bool, what string) {
	t.Helper()
	deadline := time.Now().Add(timeout)
	for !cond() {
		if time.Now().After(deadline) {
			t.Fatalf("%s did not happen within %s", what, timeout)
		}
		time.Sleep(5 * time.Millisecond)
	}
}

// TestRateLimitLowDiscardedNormalParks: with a short window the first
// announcement plays at once; low over budget is discarded on admission
// (Admission names the reason), normal parks and plays after the window.
func TestRateLimitLowDiscardedNormalParks(t *testing.T) {
	g := newSpeaker()
	q := New(Laws{RateLimit: 400 * time.Millisecond}, g.speak, nil)
	defer q.Close()

	if adm, err := q.Enqueue(Item{Text: "first"}); err != nil || adm.ID == "" || adm.Discarded != "" {
		t.Fatalf("first Enqueue = %+v (%v), want admission", adm, err)
	}
	g.waitUntil(t, 1) // first is playing, the window is booked

	adm, err := q.Enqueue(Item{Text: "low-one", Priority: PriorityLow})
	if err != nil {
		t.Fatal(err)
	}
	if adm.ID != "" || adm.Discarded != DiscardRateLimit || adm.Toast {
		t.Errorf("low over budget = %+v, want discard %q", adm, DiscardRateLimit)
	}
	if adm, err = q.Enqueue(Item{Text: "low-two", Priority: PriorityLow}); err != nil || adm.Discarded != DiscardRateLimit {
		t.Errorf("second low discard = %+v (%v)", adm, err)
	}

	adm, err = q.Enqueue(Item{Text: "second"})
	if err != nil || adm.ID == "" || adm.Discarded != "" {
		t.Fatalf("normal = %+v (%v), want admission (parks until the slot)", adm, err)
	}

	// Status view while the first plays: second waits (normal, parked).
	waitForCond(t, time.Second, func() bool {
		for _, it := range q.Pending() {
			if it.Text == "second" {
				return true
			}
		}
		return false
	}, "parked normal announcement in Pending")

	close(g.step) // end the first; second must keep parking …
	g.noFurther(t)
	// … and start after the window elapses (deadline covers jitter easily).
	started := g.waitUntil(t, 2)
	if len(started) != 1 || started[0].Text != "second" {
		t.Errorf("after the window = %+v, want second", started)
	}
	// low-one/low-two never play: exactly 2 starts in total.
	g.noFurther(t)
}

// TestHighBurstOnAndOff: with burst rights high starts right after the
// playing announcement although the window is full; without rights high
// parks like normal (no violation).
func TestHighBurstOnAndOff(t *testing.T) {
	g := newSpeaker()
	q := New(Laws{RateLimit: 10 * time.Second, AllowHighBurst: true}, g.speak, nil)

	q.Enqueue(Item{Text: "first"})
	g.waitUntil(t, 1)
	if adm, err := q.Enqueue(Item{Text: "urgent", Priority: PriorityHigh}); err != nil || adm.ID == "" {
		t.Fatalf("high with burst rights = %+v (%v), want admission", adm, err)
	}
	g.step <- struct{}{}         // end the first
	started := g.waitUntil(t, 2) // high starts despite the full window
	if len(started) != 1 || started[0].Text != "urgent" {
		t.Errorf("burst = %+v, want urgent right away", started)
	}
	q.Close()

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

	q2.Enqueue(Item{Text: "first"})
	g2.waitUntil(t, 1) // first is playing, window booked
	if adm, err := q2.Enqueue(Item{Text: "urgent", Priority: PriorityHigh}); err != nil || adm.ID == "" {
		t.Fatalf("high without burst rights = %+v (%v), want admission (parks)", adm, err)
	}
	g2.step <- struct{}{} // end the first
	g2.noFurther(t)       // high waits for the slot: no violation without rights
}

// TestNowBypassesRateLimit: now starts at once despite a full window, the
// documented escape (it still books a new window).
func TestNowBypassesRateLimit(t *testing.T) {
	g := newSpeaker()
	q := New(Laws{RateLimit: 10 * time.Second}, g.speak, nil)
	defer q.Close()

	q.Enqueue(Item{Text: "first"})
	g.waitUntil(t, 1) // window booked (10 s)

	if adm, err := q.Enqueue(Item{Text: "right-now", Now: true}); err != nil || adm.ID == "" || adm.Discarded != "" {
		t.Fatalf("now over a full window = %+v (%v), want admission", adm, err)
	}
	g.step <- struct{}{}
	started := g.waitUntil(t, 2) // long before the window elapses (10 s), else timeout
	if len(started) != 1 || started[0].Text != "right-now" {
		t.Errorf("now = %+v, want right-now", started)
	}

	// The now break-out booked a new window: normal parks now.
	if adm, err := q.Enqueue(Item{Text: "afterwards"}); err != nil || adm.ID == "" {
		t.Fatalf("normal after now = %+v (%v)", adm, err)
	}
	close(g.step)
	g.noFurther(t) // parks until the slot, not immediately
}

// TestQuietHoursDiscardAndToast: an active window discards normal and high
// (burst is a rate right, not a quiet right); toast replaces instead; now
// deliberately breaks out.
func TestQuietHoursDiscardAndToast(t *testing.T) {
	window := windowAround(time.Now())

	g := newSpeaker()
	q := New(Laws{Quiet: mustQuiet(t, window), 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 quiet hours = %+v (%v), want silent discard", adm, err)
	}
	if adm, err = q.Enqueue(Item{Text: "urgent", Priority: PriorityHigh}); err != nil || adm.Discarded != DiscardQuietHours {
		t.Errorf("high in quiet hours = %+v (%v), want discard too", adm, err)
	}
	adm, err = q.Enqueue(Item{Text: "right-now", Now: true})
	if err != nil || adm.ID == "" || adm.Discarded != "" {
		t.Fatalf("now in quiet hours = %+v (%v), want break-out", adm, err)
	}
	started := g.waitUntil(t, 1)
	if len(started) != 1 || started[0].Text != "right-now" {
		t.Errorf("in quiet hours = %+v, want only right-now", started)
	}
	g.noFurther(t)

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

	toasted := make(chan Item, 1)
	q2.OnToast(func(it Item) { toasted <- it })
	adm, err = q2.Enqueue(Item{Text: "normal"})
	if err != nil || adm.Discarded != DiscardQuietHours || !adm.Toast {
		t.Errorf("normal in quiet hours (toast) = %+v (%v), want toast replacement", adm, err)
	}
	select {
	case it := <-toasted:
		if it.Text != "normal" {
			t.Errorf("toast hook got %q", it.Text)
		}
	case <-time.After(time.Second):
		t.Error("toast hook not fired")
	}
	g2.noFurther(t)
}

// mustQuiet parses a quiet window or fails the test.
func mustQuiet(t *testing.T, window string) QuietHours {
	t.Helper()
	q, err := ParseQuietHours(window)
	if err != nil {
		t.Fatal(err)
	}
	return q
}

// TestLawState: unlimited yields an open slot; after a booked window the
// next slot lies in the future; the quiet window and limits are visible.
func TestLawState(t *testing.T) {
	g := newSpeaker()
	q := New(Laws{}, g.speak, nil)
	st := q.LawState()
	if !st.NextSlot.IsZero() || st.RateLimit != 0 {
		t.Errorf("unlimited = %+v, want zero-time slot", st)
	}
	q.Close()

	// The quiet window must be inactive NOW, otherwise the announcement
	// below would be toasted instead of spoken. A fixed window would be a
	// time bomb (red at certain hours), hence dynamic: one hour starting two
	// hours from now.
	window := inactiveQuietWindow(time.Now())
	quiet := mustQuiet(t, window)
	if quiet.Active(time.Now()) {
		t.Fatalf("test window %s is active, must not be", window)
	}
	q2 := New(Laws{RateLimit: time.Hour, AllowHighBurst: true, Quiet: quiet, QuietMode: QuietModeToast, MaxText: 400, MaxAge: 2 * time.Minute}, g.speak, nil)
	defer q2.Close()
	st2 := q2.LawState()
	if st2.RateLimit != time.Hour || !st2.AllowHighBurst || st2.MaxText != 400 || st2.MaxAge != 2*time.Minute {
		t.Errorf("LawState = %+v, want 1h window with burst rights and limits 400/2m", st2)
	}
	if st2.Quiet.String() != window || st2.QuietMode != QuietModeToast {
		t.Errorf("quiet hours in LawState = %q/%q, want %q/toast", st2.Quiet, st2.QuietMode, window)
	}
	if st2.QuietActive || !st2.QuietUntil.IsZero() {
		t.Errorf("quiet hours expected inactive, LawState reports active=%v until=%v", st2.QuietActive, st2.QuietUntil)
	}

	q2.Enqueue(Item{Text: "first"})
	g.waitUntil(t, 1)
	if next := q2.LawState().NextSlot; !next.After(time.Now()) {
		t.Errorf("after start = %v, want slot in the future", next)
	}
	g.step <- struct{}{}
}

// inactiveQuietWindow returns a one-hour "HH:MM-HH:MM" window starting two
// hours after now; it never contains now, whatever the time of day (also
// across midnight).
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: the helper window must not be active at ANY time
// of day, otherwise TestLawState would be a time bomb again.
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("at %s the window %s is active", now.Format("15:04"), w)
		}
	}
}

// syncLog is a concurrency-safe log buffer for worker lines.
type syncLog struct {
	mu sync.Mutex
	b  strings.Builder
}

func (l *syncLog) Write(p []byte) (int, error) {
	l.mu.Lock()
	defer l.mu.Unlock()
	return l.b.Write(p)
}

func (l *syncLog) String() string {
	l.mu.Lock()
	defer l.mu.Unlock()
	return l.b.String()
}

// TestLogFinishedNotCancelled: a normally finished announcement is logged as
// "finished", not "cancelled"; the worker must not read its own cancel() as
// a cancellation. A real Cancel does log "cancelled".
func TestLogFinishedNotCancelled(t *testing.T) {
	logb := &syncLog{}
	g := newSpeaker()
	q := New(Laws{}, g.speak, logb)
	defer q.Close()

	q.Enqueue(Item{Text: "runs to the end"})
	g.waitUntil(t, 1)
	g.step <- struct{}{}
	waitFor(t, func() bool { return strings.Contains(logb.String(), "finished after") })
	if strings.Contains(logb.String(), "cancelled") {
		t.Errorf("finished announcement logged as cancelled:\n%s", logb.String())
	}

	q.Enqueue(Item{Text: "gets cancelled"})
	g.waitUntil(t, 2)
	q.Cancel()
	waitFor(t, func() bool { return strings.Contains(logb.String(), "cancelled") })
}

// waitFor polls cond for up to two seconds.
func waitFor(t *testing.T, cond func() bool) {
	t.Helper()
	deadline := time.Now().Add(2 * time.Second)
	for !cond() {
		if time.Now().After(deadline) {
			t.Fatal("condition not met within 2 s")
		}
		time.Sleep(5 * time.Millisecond)
	}
}

// TestCancelItem: cancels only the playing item with that id; a stale id or
// a waiting item is a no-op, the queue goes on with the next item.
func TestCancelItem(t *testing.T) {
	g := newSpeaker()
	q := New(Laws{}, g.speak, nil)
	defer q.Close()

	first, _ := q.Enqueue(Item{Text: "playing"})
	g.waitUntil(t, 1)
	second, _ := q.Enqueue(Item{Text: "waiting"})

	if q.CancelItem(second.ID) {
		t.Error("CancelItem on a waiting item reported a hit")
	}
	if q.CancelItem("nosuch") {
		t.Error("CancelItem with an unknown id reported a hit")
	}
	if !q.CancelItem(first.ID) {
		t.Fatal("CancelItem on the playing item reported no hit")
	}
	// The playing item ended via ctx; the waiting one starts next.
	started := g.waitUntil(t, 2)
	if len(started) != 1 || started[0].ID != second.ID {
		t.Errorf("after CancelItem started = %+v, want the waiting item", started)
	}
	g.step <- struct{}{}
}