talk/internal/queue/queue.go

package queue

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

// Priority of a queue item; public contract of the HTTP API
// (low | normal | high; empty is treated as normal).
type Priority string

// Allowed priority values (high > normal > low, FIFO per level).
const (
	PriorityLow    Priority = "low"
	PriorityNormal Priority = "normal"
	PriorityHigh   Priority = "high"
)

// Valid reports whether p is an allowed priority value.
func (p Priority) Valid() bool {
	switch p {
	case PriorityLow, PriorityNormal, PriorityHigh:
		return true
	default:
		return false
	}
}

// ErrClosed reports an Enqueue attempt after Close (daemon shutting down).
var ErrClosed = errors.New("queue: already closed")

// Item is an announcement in the queue. All fields come from the HTTP API
// (POST /speak); ID and lifecycle belong to the queue.
type Item struct {
	// ID is the short announcement identifier (returned by POST /speak; the
	// CLI polls it for --wait). Assigned by Enqueue when empty.
	ID string

	// Text is the announcement text.
	Text string

	// Lang pins the language (de | en); empty = detected by the daemon.
	Lang string

	// Provider pins the chain to one tier (--provider); empty = the
	// configured default chain.
	Provider string

	// Voice pins a voice ID (--voice); empty = voice mapping/defaults.
	Voice string

	// Priority controls processing order (high > normal > low, FIFO per
	// level).
	Priority Priority

	// Now lets the announcement jump ahead of every waiting item (including
	// high); it never interrupts the announcement that is playing.
	Now bool

	seq      uint64    // enqueue order: FIFO tiebreaker within a level
	enqueued time.Time // admission time (basis for Laws.MaxAge)
}

// rank returns the processing class: now over high over normal over 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 plays an item. In the daemon: build the provider chain and call
// Speak (synthesis with fallback plus playback through the audio sink). ctx
// is cancelled when the announcement is cancelled (POST /cancel) or the
// daemon shuts down; the chain then aborts its network and process work.
// The returned error is logged, never passed to the API.
type Speaker func(ctx context.Context, item Item) error

// Queue is the daemon's serialized announcement queue: exactly one
// announcement at a time, order now > high > normal > low (FIFO per level),
// under the anti-spam laws: normal parks until the rate-limit slot, low
// over budget is discarded, high may break the limit with burst rights, now
// deliberately breaks out, quiet hours discard or toast instead of audio,
// stale items expire.
//
// Synchronization is deliberately simple and obviously correct rather than
// clever: ONE mutex and ONE cond protect everything (items, current item,
// closed flag, cancelCur, slot timer). The worker is the only one taking
// items; take and setting current/cancelCur happen under the same lock, so
// an item can never slip past cancellation between "waiting" and "playing".
// Cancel/Close call cancelCur under the lock; the worker notices via ctx in
// its speak function. Parked items stay in the queue; a slot timer
// (armWakeLocked) broadcasts when the window elapses so the worker wakes
// without a new Enqueue. The wait condition re-evaluates the laws every
// time, so lost signals are harmless.
type Queue struct {
	laws  Laws
	limit *RateLimiter
	speak Speaker
	log   io.Writer
	toast func(Item) // quiet-hours toast (optional; set before the first Enqueue)

	mu        sync.Mutex
	notEmpty  *sync.Cond // Signal: new item; Broadcast: Close / slot open
	items     []Item
	current   *Item
	cancelCur context.CancelFunc
	seq       uint64
	closed    bool
	wake      *time.Timer   // wakes the parked worker at the slot time
	wakeAt    time.Time     // time of the armed timer
	done      chan struct{} // closed when the worker loop ends
}

// New starts the queue with its worker loop (one announcement at a time).
// laws carries the anti-spam laws (zero value = everything allowed; the
// daemon supplies the configured laws). log is optional (nil = silent) and
// receives the items' lifecycle lines. Close ends the loop and waits for it.
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
}

// OnToast sets the notification fired during quiet hours with quiet_mode =
// "toast" instead of audio (daemon: desktop notification). Unset, only the
// log line remains. Set it before the first Enqueue.
func (q *Queue) OnToast(fn func(Item)) { q.toast = fn }

// fireToast calls the toast hook (never under the mutex).
func (q *Queue) fireToast(item Item) {
	if q.toast != nil {
		q.toast(item)
	}
}

// Enqueue puts item into the queue and returns the admission decision:
// admitted items carry their ID (assigned when empty: short and random).
// Quiet-hours discard/toast and low over the window budget are decided on
// admission (the 202 response of POST /speak names the reason); everything
// else is decided by the worker when the item comes due. After Close,
// Enqueue is an error.
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
	}
	var truncated bool
	if item.Text, truncated = truncateText(item.Text, q.laws.MaxText); truncated {
		defer q.logf("queue: announcement (%q …) truncated to %d characters (max_text)", firstRunes(item.Text, 40), q.laws.MaxText)
	}
	item.enqueued = time.Now()
	switch q.laws.decide(item, item.enqueued, q.limit.Allow()) {
	case verdictDiscardQuiet:
		q.mu.Unlock()
		q.logf("queue: announcement (%q, priority %s) discarded: quiet hours %s active (mode discard)",
			item.Text, item.Priority, q.laws.Quiet)
		return Admission{Discarded: DiscardQuietHours}, nil
	case verdictToast:
		q.mu.Unlock()
		q.logf("queue: announcement (%q, priority %s) replaced by toast: quiet hours %s active (mode toast)",
			item.Text, item.Priority, q.laws.Quiet)
		q.fireToast(item)
		return Admission{Discarded: DiscardQuietHours, Toast: true}, nil
	case verdictDiscardRate:
		q.mu.Unlock()
		q.logf("queue: announcement (%q, priority low) discarded: 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, Truncated: truncated}, nil
}

// firstRunes returns the first n runes of s (log preview).
func firstRunes(s string, n int) string {
	r := []rune(s)
	if len(r) <= n {
		return s
	}
	return string(r[:n])
}

// Pending returns a snapshot of the waiting items in processing order (best
// item first); basis for 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 reports whether a is processed before b: higher class first,
// earlier enqueue order within the same class.
func itemBefore(a, b Item) bool {
	if a.rank() != b.rank() {
		return a.rank() > b.rank()
	}
	return a.seq < b.seq
}

// Current returns the item that is playing (false: queue is 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 aborts the playing announcement and empties the queue. It returns
// the number of cancelled announcements (playing + waiting).
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
}

// CancelItem aborts the announcement with the given id if it is the one
// playing right now (the Cancel button of its notification). Waiting items
// and the rest of the queue are untouched. Reports whether it hit.
func (q *Queue) CancelItem(id string) bool {
	q.mu.Lock()
	defer q.mu.Unlock()
	if q.current == nil || q.current.ID != id || q.cancelCur == nil {
		return false
	}
	q.cancelCur()
	return true
}

// Close shuts the queue down: waiting items are dropped, the playing
// announcement is aborted, the worker loop ends. Close waits until the loop
// has ended (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 returns the /status view of the laws: the configured rate limit
// with the next open slot (zero = open or unlimited), the quiet window with
// its end if active, and the length and age limits.
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),
		MaxText:        q.laws.MaxText,
		MaxAge:         q.laws.MaxAge,
	}
}

// loop is the worker: it waits for due items and plays them one by one.
// "Due" includes the laws: parked items sleep until the rate-limit slot
// (armWakeLocked wakes them), discards pass through the worker without
// audio but with a log line, now and high burst play immediately. After
// Close (also when Close hits during an announcement) it exits as soon as
// the playing announcement has returned.
func (q *Queue) loop() {
	defer close(q.done)
	for {
		q.mu.Lock()
		for !q.closed && !q.readyLocked() {
			if len(q.items) > 0 {
				// The best item parks for its slot: arm the timer.
				q.armWakeLocked(q.limit.Next())
			}
			q.notEmpty.Wait()
		}
		if q.closed {
			q.mu.Unlock()
			return
		}
		item, v := q.takeLocked()
		if v == verdictWait {
			// Woken without progress (for example a lost signal): go back to
			// sleep. Cannot really happen after readyLocked; parked rather
			// than spun, just in case.
			q.mu.Unlock()
			continue
		}
		if v == verdictPlay {
			// Book the window BEFORE speaking: as soon as the item counts as
			// playing, the next slot time is fixed.
			q.limit.Reserve()
			ctx, cancel := context.WithCancel(context.Background())
			q.current = &item
			q.cancelCur = cancel
			q.mu.Unlock()

			q.logf("queue: announcement %s (%q, priority %s, now=%v) starting", item.ID, item.Text, item.Priority, item.Now)
			start := time.Now()
			err := q.speak(ctx, item)
			dur := time.Since(start)
			// Read the cancellation BEFORE our own cancel(): afterwards
			// ctx.Err() is always set and every finished announcement would
			// read as cancelled.
			cancelled := ctx.Err() != nil
			cancel()

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

			switch {
			case cancelled:
				q.logf("queue: announcement %s cancelled", item.ID)
			case err != nil:
				q.logf("queue: announcement %s failed after %s: %v", item.ID, dur.Round(time.Millisecond), err)
			default:
				q.logf("queue: announcement %s finished after %s", item.ID, dur.Round(time.Millisecond))
			}
			continue
		}
		q.mu.Unlock()
		switch v {
		case verdictDiscardRate:
			q.logf("queue: announcement %s (%q, priority low) discarded: rate limit %s",
				item.ID, item.Text, q.laws.RateLimit)
		case verdictDiscardQuiet:
			q.logf("queue: announcement %s (%q) discarded: quiet hours %s active (mode discard)",
				item.ID, item.Text, q.laws.Quiet)
		case verdictToast:
			q.logf("queue: announcement %s (%q) replaced by toast: quiet hours %s active (mode toast)",
				item.ID, item.Text, q.laws.Quiet)
			q.fireToast(item)
		case verdictDiscardExpired:
			q.logf("queue: announcement %s (%q) discarded: waited %s, older than max_age %s",
				item.ID, item.Text, time.Since(item.enqueued).Round(time.Second), q.laws.MaxAge)
		}
	}
}

// readyLocked reports whether the worker has work (mu must be held): always
// when closed; otherwise only when the best item is not merely parked for
// its rate-limit slot. Discards count as work: they pass through the worker
// without audio but with a log line.
func (q *Queue) readyLocked() bool {
	if q.closed || len(q.items) == 0 {
		return q.closed
	}
	return q.verdictLocked(q.items[q.bestIndexLocked()]) != verdictWait
}

// takeLocked removes the next due item when it is playable or decidable
// (mu must be held). On verdictWait it is NOT removed: the item stays
// parked until the slot opens (or a better item overtakes it).
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 returns the index of the next due item (mu must be held).
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 applies the laws to an item (mu must be held; the lock order
// queue → RateLimiter is preserved).
func (q *Queue) verdictLocked(item Item) verdict {
	return q.laws.decide(item, time.Now(), q.limit.Allow())
}

// armWakeLocked arms the slot timer for until (mu must be held): it
// broadcasts the cond at that time so the parked worker wakes without a new
// Enqueue. A running timer with an earlier (or equal) time suffices;
// otherwise it is replaced.
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 returns a short, collision-resistant ID (6 hex characters). IDs only
// need to be unique per daemon run (--wait polling); the fallback is
// practically unreachable but deterministic.
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 writes a log line when a log is set. Never called under the mutex
// (callers take care of that).
func (q *Queue) logf(format string, args ...any) {
	if q.log != nil {
		fmt.Fprintf(q.log, format+"\n", args...)
	}
}