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).
//
// Die Synchronisation ist bewusst simpel und offensichtlich korrekt statt
// clever: EIN Mutex + EIN Cond schützen alles (Items, aktuelles Item,
// Closed-Flag, cancelCur). Der Worker ist der einzige, der Items entnimmt;
// pop 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.
type Queue struct {
	speak Speaker
	log   io.Writer

	mu        sync.Mutex
	notEmpty  *sync.Cond // Signal: neues Item; Broadcast: Close
	items     []Item
	current   *Item
	cancelCur context.CancelFunc
	seq       uint64
	closed    bool
	done      chan struct{} // geschlossen, wenn der Worker-Loop endet
}

// New startet die Queue mit ihrem Worker-Loop (eine Ansage gleichzeitig).
// log ist optional (nil = still) und empfängt die Lebenslauf-Zeilen der
// Items. Close beendet den Loop und wartet auf sein Ende.
func New(speak Speaker, log io.Writer) *Queue {
	q := &Queue{speak: speak, log: log, done: make(chan struct{})}
	q.notEmpty = sync.NewCond(&q.mu)
	go q.loop()
	return q
}

// Enqueue stellt item in die Queue und liefert dessen ID (bei leerer ID neu
// vergeben: kurz und zufällig). Nach Close ist Enqueue ein Fehler.
func (q *Queue) Enqueue(item Item) (string, error) {
	q.mu.Lock()
	if q.closed {
		q.mu.Unlock()
		return "", ErrClosed
	}
	if item.Priority == "" {
		item.Priority = PriorityNormal
	}
	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 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()
	}
	q.mu.Unlock()
	q.notEmpty.Broadcast()
	<-q.done
}

// loop ist der Worker: wartet auf Items und spielt sie einzeln ab. 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 len(q.items) == 0 && !q.closed {
			q.notEmpty.Wait()
		}
		if q.closed {
			q.mu.Unlock()
			return
		}
		item := q.popLocked()
		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)
		cancel()

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

		switch {
		case ctx.Err() != nil:
			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))
		}
	}
}

// popLocked entnimmt das als nächstes fällige Item (mu muss gehalten werden).
func (q *Queue) popLocked() Item {
	best := 0
	for i := 1; i < len(q.items); i++ {
		if itemBefore(q.items[i], q.items[best]) {
			best = i
		}
	}
	item := q.items[best]
	q.items = append(q.items[:best], q.items[best+1:]...)
	return item
}

// 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...)
	}
}