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