talk/internal/queue/queue.go

package queue

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"errors"
	"fmt"
	"io"
	"sort"
	"sync"
	"time"
)

// Priority eines Queue-Items — öffentlicher Vertrag der HTTP-API
// (low | normal | high; leer wird als normal behandelt).
type Priority string

// Erlaubte Prioritäts-Werte (Spec: high > normal > low, FIFO je Level).
const (
	PriorityLow    Priority = "low"
	PriorityNormal Priority = "normal"
	PriorityHigh   Priority = "high"
)

// Valid meldet, ob p ein erlaubter Prioritäts-Wert ist.
func (p Priority) Valid() bool {
	switch p {
	case PriorityLow, PriorityNormal, PriorityHigh:
		return true
	default:
		return false
	}
}

// ErrClosed meldet einen Enqueue-Versuch nach Close (Daemon fährt herunter).
var ErrClosed = errors.New("queue: bereits geschlossen")

// Item ist eine Ansage in der Warteschlange. Alle Felder kommen aus der
// HTTP-API (POST /speak); ID und Ablauf-Verwaltung liegen bei der Queue.
type Item struct {
	// ID ist die kurze Ansage-Kennung (Antwort von POST /speak; Grundlage
	// des --wait-Pollings der CLI). Wird von Enqueue vergeben, wenn leer.
	ID string

	// Text ist der Ansage-Text.
	Text string

	// Lang ist die Sprach-Pin (de | en); leer = automatische Erkennung
	// durch den Daemon (langdetect).
	Lang string

	// Provider pinnt die Provider-Kette auf eine Stufe (--provider); leer
	// = konfigurierte Default-Kette.
	Provider string

	// Voice pinnt eine Voice-ID (--voice); leer = Voice-Mapping/Defaults.
	Voice string

	// Priority steuert die Abarbeitungs-Reihenfolge (high > normal > low,
	// FIFO je Level).
	Priority Priority

	// Now lässt die Ansage vor alle wartenden Items springen (auch vor
	// high) — sie unterbricht aber nie die laufende Ansage.
	Now bool

	seq uint64 // Enqueue-Reihenfolge: FIFO-Tiebrecher innerhalb eines Levels
}

// rank liefert die Abarbeitungs-Klasse: now über high über normal über low.
func (it Item) rank() int {
	if it.Now {
		return 3
	}
	switch it.Priority {
	case PriorityHigh:
		return 2
	case PriorityNormal:
		return 1
	default:
		return 0
	}
}

// Speaker spielt ein Item ab. Im Daemon: Provider-Kette bauen und Speak
// aufrufen (Synthese mit Fallback + Wiedergabe über den Audio-Sink — die
// Kette aus internal/provider wird unverändert wiederverwendet). ctx wird
// abgebrochen, wenn die Ansage abgebrochen (POST /cancel) oder der Daemon
// heruntergefahren wird; die Kette bricht ihre Netz- und Prozess-Arbeit
// dann ab. Der Rückgabe-Fehler wird protokolliert, nie an die API gereicht.
type Speaker func(ctx context.Context, item Item) error

// Queue ist die serialisierte Ansage-Queue des Daemons: genau eine Ansage
// gleichzeitig, Reihenfolge now > high > normal > low (FIFO je Level) —
// seit Task 05 unter dem Anti-Spam-Gesetz (Laws): normal parkt bis zum
// Rate-Limit-Slot, low wird über dem Budget verworfen, high darf mit
// Burst-Recht verstoßen, now bricht bewusst aus, und Ruhezeiten verwerfen
// bzw. toasten statt Audio.
//
// Die Synchronisation ist bewusst simpel und offensichtlich korrekt statt
// clever: EIN Mutex + EIN Cond schützen alles (Items, aktuelles Item,
// Closed-Flag, cancelCur, Slot-Wecker). Der Worker ist der einzige, der
// Items entnimmt; take und das Setzen von current/cancelCur geschehen unter
// demselben Lock, damit ein Item nie zwischen „wartet" und „läuft" der
// Cancelierung entwischen kann. Cancel/Close rufen cancelCur unter dem Lock
// ab — der Worker bemerkt die Abbrechung in seiner speak-Funktion über ctx.
// Geparkte Items bleiben in der Warteschlange liegen; ein Slot-Wecker
// (armWakeLocked) broadcastet zum Fenster-Ablauf, damit der Worker auch ohne
// neuen Enqueue aufwacht — die Wait-Bedingung prüft jedes Mal die Gesetze
// neu, verlorene Signale sind dadurch harmlos.
type Queue struct {
	laws  Laws
	limit *RateLimiter
	speak Speaker
	log   io.Writer

	mu        sync.Mutex
	notEmpty  *sync.Cond // Signal: neues Item; Broadcast: Close/Slot frei
	items     []Item
	current   *Item
	cancelCur context.CancelFunc
	seq       uint64
	closed    bool
	wake      *time.Timer   // weckt den geparkten Worker zum Slot-Termin
	wakeAt    time.Time     // Termin des laufenden Weckers
	done      chan struct{} // geschlossen, wenn der Worker-Loop endet
}

// New startet die Queue mit ihrem Worker-Loop (eine Ansage gleichzeitig).
// laws trägt die Anti-Spam-Gesetze (Nullwert = alles erlaubt; der Daemon
// liefert die Config-Gesetze). log ist optional (nil = still) und empfängt
// die Lebenslauf-Zeilen der Items. Close beendet den Loop und wartet auf
// sein Ende.
func New(laws Laws, speak Speaker, log io.Writer) *Queue {
	q := &Queue{
		laws:  laws,
		limit: &RateLimiter{Interval: laws.RateLimit},
		speak: speak,
		log:   log,
		done:  make(chan struct{}),
	}
	q.notEmpty = sync.NewCond(&q.mu)
	go q.loop()
	return q
}

// Enqueue stellt item in die Warteschlange und liefert die Aufnahme-
// Entscheidung: Aufgenommene tragen ihre ID (bei leerer ID neu vergeben:
// kurz und zufällig). Ruhezeit-Verwurf/Toast und low über dem Fenster-Budget
// werden schon beim Eintritt entschieden (die 202-Antwort von POST /speak
// nennt den Grund); alles andere entscheidet der Worker, wenn das Item
// fällig ist. Nach Close ist Enqueue ein Fehler.
func (q *Queue) Enqueue(item Item) (Admission, error) {
	q.mu.Lock()
	if q.closed {
		q.mu.Unlock()
		return Admission{}, ErrClosed
	}
	if item.Priority == "" {
		item.Priority = PriorityNormal
	}
	switch q.laws.decide(item, time.Now(), q.limit.Allow()) {
	case verdictDiscardQuiet:
		q.mu.Unlock()
		q.logf("queue: ansage (%q, priorität %s) verworfen — ruhezeit %s aktiv (modus discard)",
			item.Text, item.Priority, q.laws.Quiet)
		return Admission{Discarded: DiscardQuietHours}, nil
	case verdictToast:
		q.mu.Unlock()
		q.logf("queue: ansage (%q, priorität %s) durch toast ersetzt — ruhezeit %s aktiv (modus toast)",
			item.Text, item.Priority, q.laws.Quiet)
		return Admission{Discarded: DiscardQuietHours, Toast: true}, nil
	case verdictDiscardRate:
		q.mu.Unlock()
		q.logf("queue: ansage (%q, priorität low) verworfen — rate-limit %s",
			item.Text, q.laws.RateLimit)
		return Admission{Discarded: DiscardRateLimit}, nil
	}
	q.seq++
	item.seq = q.seq
	if item.ID == "" {
		item.ID = newID()
	}
	q.items = append(q.items, item)
	q.mu.Unlock()
	q.notEmpty.Signal()
	return Admission{ID: item.ID}, nil
}

// Pending liefert eine Momentaufnahme der wartenden Items in
// Abarbeitungs-Reihenfolge (bestes Item zuerst) — Basis für GET /status.
func (q *Queue) Pending() []Item {
	q.mu.Lock()
	defer q.mu.Unlock()
	out := make([]Item, len(q.items))
	copy(out, q.items)
	sort.SliceStable(out, func(i, j int) bool { return itemBefore(out[i], out[j]) })
	return out
}

// itemBefore meldet, ob a vor b abgearbeitet wird: höhere Klasse zuerst,
// innerhalb derselben Klasse die frühere Enqueue-Reihenfolge.
func itemBefore(a, b Item) bool {
	if a.rank() != b.rank() {
		return a.rank() > b.rank()
	}
	return a.seq < b.seq
}

// Current liefert das gerade laufende Item (false: Queue ist idle).
func (q *Queue) Current() (Item, bool) {
	q.mu.Lock()
	defer q.mu.Unlock()
	if q.current == nil {
		return Item{}, false
	}
	return *q.current, true
}

// Cancel bricht die laufende Ansage ab und leert die Queue. Rückgabe ist
// die Zahl der abgebrochenen Ansagen (laufende + wartende).
func (q *Queue) Cancel() int {
	q.mu.Lock()
	defer q.mu.Unlock()
	n := len(q.items)
	q.items = nil
	if q.current != nil {
		n++
		if q.cancelCur != nil {
			q.cancelCur()
		}
	}
	return n
}

// Close fährt die Queue herunter: wartende Items werden verworfen, die
// laufende Ansage abgebrochen, der Worker-Loop beendet. Close wartet, bis
// der Loop sicher beendet ist (idempotent).
func (q *Queue) Close() {
	q.mu.Lock()
	if q.closed {
		q.mu.Unlock()
		<-q.done
		return
	}
	q.closed = true
	q.items = nil
	if q.cancelCur != nil {
		q.cancelCur()
	}
	if q.wake != nil {
		q.wake.Stop()
		q.wake = nil
	}
	q.mu.Unlock()
	q.notEmpty.Broadcast()
	<-q.done
}

// LawState liefert die /status-Sicht auf die Anti-Spam-Gesetze: das
// konfigurierte Rate-Limit mit dem nächsten freien Slot (Nullzeit = frei
// oder unbegrenzt) und das Ruhezeit-Fenster mit seinem Ende, wenn es
// gerade aktiv ist.
func (q *Queue) LawState() LawState {
	now := time.Now()
	return LawState{
		RateLimit:      q.laws.RateLimit,
		AllowHighBurst: q.laws.AllowHighBurst,
		NextSlot:       q.limit.Next(),
		Quiet:          q.laws.Quiet,
		QuietActive:    q.laws.Quiet.Active(now),
		QuietMode:      q.laws.QuietMode,
		QuietUntil:     q.laws.Quiet.Until(now),
	}
}

// loop ist der Worker: wartet auf fällige Items und spielt sie einzeln ab.
// „Fällig" heißt seit Task 05 auch gesetzkonform: Geparkte schlafen bis zum
// Rate-Limit-Slot (armWakeLocked weckt), low- und Ruhezeit-Verwürfe laufen
// ohne Audio durchs Protokoll, now und high-burst spielen sofort.
// Nach Close (oder wenn Close während einer Ansage zuschlägt) beendet er
// sich, sobald die laufende Ansage zurückgekehrt ist.
func (q *Queue) loop() {
	defer close(q.done)
	for {
		q.mu.Lock()
		for !q.closed && !q.readyLocked() {
			if len(q.items) > 0 {
				// Bestes Item parkt auf den Slot: Wecker zum Termin armieren.
				q.armWakeLocked(q.limit.Next())
			}
			q.notEmpty.Wait()
		}
		if q.closed {
			q.mu.Unlock()
			return
		}
		item, v := q.takeLocked()
		if v == verdictWait {
			// Kurz ohne Fortschritt geweckt (z. B. verlorenes Signal):
			// wieder schlafen legen. Kann nach readyLocked eigentlich nie
			// eintreten — sicherheitshalber geparkt statt gedreht.
			q.mu.Unlock()
			continue
		}
		if v == verdictPlay {
			// Fenster buchen, BEVOR gesprochen wird: Sobald das Item als
			// laufend gilt, steht der nächste Slot-Termin fest.
			q.limit.Reserve()
			ctx, cancel := context.WithCancel(context.Background())
			q.current = &item
			q.cancelCur = cancel
			q.mu.Unlock()

			q.logf("queue: ansage %s (%q, Priorität %s, now=%v) startet", item.ID, item.Text, item.Priority, item.Now)
			start := time.Now()
			err := q.speak(ctx, item)
			dur := time.Since(start)
			// Abbruch VOR dem eigenen cancel() ablesen — danach wäre ctx.Err()
			// immer gesetzt und jede fertige Ansage hieße „abgebrochen".
			cancelled := ctx.Err() != nil
			cancel()

			q.mu.Lock()
			q.current = nil
			q.cancelCur = nil
			q.mu.Unlock()

			switch {
			case cancelled:
				q.logf("queue: ansage %s abgebrochen", item.ID)
			case err != nil:
				q.logf("queue: ansage %s fehlgeschlagen nach %s: %v", item.ID, dur.Round(time.Millisecond), err)
			default:
				q.logf("queue: ansage %s fertig nach %s", item.ID, dur.Round(time.Millisecond))
			}
			continue
		}
		q.mu.Unlock()
		switch v {
		case verdictDiscardRate:
			q.logf("queue: ansage %s (%q, priorität low) verworfen — rate-limit %s",
				item.ID, item.Text, q.laws.RateLimit)
		case verdictDiscardQuiet:
			q.logf("queue: ansage %s (%q) verworfen — ruhezeit %s aktiv (modus discard)",
				item.ID, item.Text, q.laws.Quiet)
		case verdictToast:
			q.logf("queue: ansage %s (%q) durch toast ersetzt — ruhezeit %s aktiv (modus toast)",
				item.ID, item.Text, q.laws.Quiet)
		}
	}
}

// readyLocked meldet, ob der Worker arbeiten kann (mu muss gehalten
// werden): geschlossen immer; sonst nur, wenn das beste Item nicht bloß
// auf seinen Rate-Limit-Slot parkt. Verwürfe zählen als Arbeit — sie
// laufen ohne Audio, aber mit Protokoll durch den Worker.
func (q *Queue) readyLocked() bool {
	if q.closed || len(q.items) == 0 {
		return q.closed
	}
	return q.verdictLocked(q.items[q.bestIndexLocked()]) != verdictWait
}

// takeLocked entnimmt das als nächstes fällige Item, wenn es spielbar oder
// entscheidbar ist (mu muss gehalten werden). Bei verdictWait wird NICHT
// entnommen — das Item bleibt geparkt, bis der Slot frei ist (oder ein
// besseres Item vorgezogen wird).
func (q *Queue) takeLocked() (Item, verdict) {
	best := q.bestIndexLocked()
	item := q.items[best]
	v := q.verdictLocked(item)
	if v != verdictWait {
		q.items = append(q.items[:best], q.items[best+1:]...)
	}
	return item, v
}

// bestIndexLocked liefert den Index des als nächstes fälligen Items
// (mu muss gehalten werden).
func (q *Queue) bestIndexLocked() int {
	best := 0
	for i := 1; i < len(q.items); i++ {
		if itemBefore(q.items[i], q.items[best]) {
			best = i
		}
	}
	return best
}

// verdictLocked wendet die Gesetze auf ein Item an (mu muss gehalten
// werden — die Lock-Ordnung Queue → RateLimiter bleibt gewahrt).
func (q *Queue) verdictLocked(item Item) verdict {
	return q.laws.decide(item, time.Now(), q.limit.Allow())
}

// armWakeLocked armiert den Slot-Wecker auf until (mu muss gehalten
// werden): Er broadcastet die Cond zum Termin, damit der geparkte Worker
// auch ohne neuen Enqueue aufwacht. Ein bereits laufender Wecker mit
// früherem (oder gleichem) Termin reicht; sonst wird er ersetzt.
func (q *Queue) armWakeLocked(until time.Time) {
	if q.wake != nil && !q.wakeAt.After(until) {
		return
	}
	if q.wake != nil {
		q.wake.Stop()
	}
	q.wakeAt = until
	d := time.Until(until)
	if d < 0 {
		d = 0
	}
	q.wake = time.AfterFunc(d, func() {
		q.mu.Lock()
		q.wake = nil
		q.mu.Unlock()
		q.notEmpty.Broadcast()
	})
}

// newID liefert eine kurze, kollisionsarme ID (6 Hex-Zeichen). IDs müssen
// nur pro Daemon-Lauf eindeutig sein (--wait-Polling); der Fallback ist
// praktisch unerreichbar, aber deterministisch sicher.
func newID() string {
	var b [3]byte
	if _, err := rand.Read(b[:]); err == nil {
		return hex.EncodeToString(b[:])
	}
	return fmt.Sprintf("id%d", time.Now().UnixNano())
}

// logf schreibt eine Protokoll-Zeile, wenn ein Log gesetzt ist. Wird nie
// unter dem Mutex gerufen (Aufrufer achten darauf).
func (q *Queue) logf(format string, args ...any) {
	if q.log != nil {
		fmt.Fprintf(q.log, format+"\n", args...)
	}
}