fenrir/internal/speech/service.go

// Fenrir console screen reader.
// Copyright (c) 2026 Storm Dragon
// SPDX-License-Identifier: GPL-3.0-or-later
//
// This program is free software: you can redistribute it and/or modify it
// under the terms of the GNU General Public License as published by the Free
// Software Foundation, either version 3 of the License, or (at your option)
// any later version.
//
// This program is distributed in the hope that it will be useful, but WITHOUT
// ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
// FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
// more details.
//
// You should have received a copy of the GNU General Public License along with
// this program. If not, see <https://www.gnu.org/licenses/gpl-3.0.html>.

package speech

import (
	"context"
	"errors"
	"log/slog"

	"git.stormux.org/storm/fenrir/internal/audio"
	"strings"
	"sync"
	"time"

	"git.stormux.org/storm/fenrir/internal/diagnostic"
)

type Transport interface {
	// Say receives prepared text; backends must disable additional punctuation
	// naming so the shared punctuation processor's names are spoken once.
	Say(string) error
	Configure(Parameters) error
	Set_pitch(int) error
	Cancel() error
	Close()
}

type Service struct {
	transport            Transport
	transportMu          sync.RWMutex
	transportEpoch       uint64
	faults               chan transportFailure
	recoveryContext      context.Context
	stopRecovery         context.CancelFunc
	recovering           bool
	replacements         []replacement
	mu                   sync.Mutex
	dispatch             sync.Mutex
	queue                []utterance
	parameters           *Parameters
	generation, canceled uint64
	muted, closed        bool
	wake, cancelWake     chan struct{}
	done                 chan struct{}
	errors               chan error
	workers              sync.WaitGroup
}

type transportFailure struct {
	epoch uint64
	err   error
}

type replacement struct {
	transport Transport
	done      chan error
}

// Replace commits an already prepared client on the serialized speech worker.
// Mute, cancellation generations and pending numeric adjustments survive.
// Ownership transfers to the service even if shutdown races the request.
func (s *Service) Replace(transport Transport) error {
	request := replacement{transport: transport, done: make(chan error, 1)}
	s.mu.Lock()
	if s.closed {
		s.mu.Unlock()
		transport.Close()
		return errors.New("speech service closed")
	}
	s.replacements = append(s.replacements, request)
	signal(s.wake)
	s.mu.Unlock()
	return <-request.done
}

type utterance struct {
	text       string
	queuedAt   time.Time
	pitch      *int
	parameters *Parameters
}

func New(transport Transport) *Service {
	ctx, cancel := context.WithCancel(context.Background())
	s := &Service{recoveryContext: ctx, stopRecovery: cancel, faults: make(chan transportFailure, 1), transport: transport, wake: make(chan struct{}, 1), cancelWake: make(chan struct{}, 1), done: make(chan struct{}), errors: make(chan error, 1)}
	s.workers.Add(2)
	go s.deliver()
	go s.cancel()
	return s
}

func signal(wake chan struct{}) {
	select {
	case wake <- struct{}{}:
	default:
	}
}

func (s *Service) Errors() <-chan error { return s.errors }

// Configure coalesces adjustments without dropping them on speech interruption
// or mute. Delivery applies the latest values before admitting more speech.
func (s *Service) Configure(parameters Parameters) {
	if err := parameters.Validate(); err != nil {
		s.fail(err)
		return
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	if !s.closed {
		s.parameters = &parameters
		signal(s.wake)
	}
}

func (s *Service) Say(text string) {
	s.enqueue(utterance{text: text})
}

// Say_with_parameters previews one utterance without changing saved or live
// defaults. The serialized worker restores its latest normal parameters.
func (s *Service) Say_with_parameters(text string, parameters Parameters) {
	if err := parameters.Validate(); err != nil {
		s.fail(err)
		return
	}
	s.enqueue(utterance{text: text, parameters: &parameters})
}

// Say_with_pitch scopes the override and restoration to one queued request.
// Interruption discards both together; configuration never leaks into later speech.
func (s *Service) Say_with_pitch(text string, pitch int) {
	if pitch < -100 || pitch > 100 {
		s.fail(errors.New("speech pitch must be between -100 and 100"))
		return
	}
	s.enqueue(utterance{text: text, pitch: &pitch})
}

func (s *Service) enqueue(item utterance) {
	if strings.TrimSpace(item.text) == "" {
		return
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.closed || s.muted || s.recovering {
		return
	}
	item.queuedAt = time.Now()
	s.queue = append(s.queue, item)
	signal(s.wake)
}

func (s *Service) Interrupt() {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.closed {
		return
	}
	s.interrupt()
}

func (s *Service) interrupt() {
	clear(s.queue)
	s.queue = nil
	s.generation++
	signal(s.cancelWake)
}

func (s *Service) Toggle_mute() bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.closed {
		return s.muted
	}
	s.muted = !s.muted
	if s.muted {
		s.interrupt()
	}
	return s.muted
}

func (s *Service) fail(err error) {
	if err != nil {
		select {
		case s.errors <- err:
		default:
		}
	}
}

// Recovery belongs to the delivery worker. Admission and cancellation never
// wait for a new daemon, and speech from the outage is never replayed.
func (s *Service) recover_transport(problem error, parameters *Parameters) bool {
	reopener, ok := s.transport.(interface {
		Reopen(context.Context) (Transport, error)
	})
	if !ok || !audio.Temporary(problem) {
		s.fail(problem)
		return false
	}
	s.dispatch.Lock()
	defer s.dispatch.Unlock()
	s.mu.Lock()
	s.recovering = true
	s.queue = nil
	s.mu.Unlock()
	s.transportMu.Lock()
	old := s.transport
	s.transport = nil
	s.transportEpoch++
	s.transportMu.Unlock()
	old.Close()
	slog.Warn("speech connection lost; reconnecting", "error", problem, "at", time.Now())
	ctx, cancel := context.WithTimeout(s.recoveryContext, audio.RecoveryTimeout)
	defer cancel()
	var candidate Transport
	err := audio.Retry(ctx, func(ctx context.Context) error {
		var err error
		candidate, err = reopener.Reopen(ctx)
		if err == nil && parameters != nil {
			if client, ok := candidate.(*Client); ok {
				stop := context.AfterFunc(ctx, client.abort)
				err = candidate.Configure(*parameters)
				stop()
			} else {
				err = candidate.Configure(*parameters)
			}
		}
		if err != nil && candidate != nil {
			candidate.Close()
			candidate = nil
		}
		return err
	}, audio.Temporary)
	err = errors.Join(err, ctx.Err())
	s.mu.Lock()
	if err != nil || s.closed || ctx.Err() != nil {
		closed := s.closed
		s.mu.Unlock()
		if candidate != nil {
			candidate.Close()
		}
		if err != nil && !closed {
			s.fail(errors.Join(errors.New("speech recovery failed"), err))
		}
		return false
	}
	s.transportMu.Lock()
	s.transport = candidate
	s.transportMu.Unlock()
	s.recovering = false
	s.mu.Unlock()
	slog.Info("speech connection restored", "at", time.Now())
	return true
}

func (s *Service) deliver() {
	defer s.workers.Done()
	normalPitch := 0
	var currentParameters *Parameters
	for {
		select {
		case <-s.done:
			return
		case <-s.wake:
		case fault := <-s.faults:
			if fault.epoch == s.transportEpoch && !s.recover_transport(fault.err, currentParameters) {
				return
			}
		}
		for {
			s.mu.Lock()
			if !s.closed && s.generation == s.canceled && s.parameters != nil {
				parameters := *s.parameters
				currentParameters = &parameters
				normalPitch = parameters.Pitch
				s.parameters = nil
				s.dispatch.Lock()
				s.mu.Unlock()
				err := s.transport.Configure(parameters)
				s.dispatch.Unlock()
				if err != nil {
					if !s.recover_transport(err, currentParameters) {
						return
					}
				}
				continue
			}
			if !s.closed && s.generation == s.canceled && len(s.replacements) > 0 {
				request := s.replacements[0]
				s.replacements = s.replacements[1:]
				s.dispatch.Lock()
				s.mu.Unlock()
				var err error
				if currentParameters != nil {
					err = request.transport.Configure(*currentParameters)
				}
				s.mu.Lock()
				if s.closed {
					err = errors.New("speech service closed")
				}
				if err == nil {
					s.transportMu.Lock()
					old := s.transport
					s.transport = request.transport
					s.transportEpoch++
					s.transportMu.Unlock()
					s.mu.Unlock()
					old.Close()
				} else {
					s.mu.Unlock()
					request.transport.Close()
				}
				s.dispatch.Unlock()
				request.done <- err
				continue
			}
			if s.closed || s.muted || s.generation != s.canceled || len(s.queue) == 0 {
				s.mu.Unlock()
				break
			}
			item := s.queue[0]
			s.queue[0] = utterance{}
			s.queue = s.queue[1:]
			// Claim dispatch before releasing admission state. An interrupt
			// must account for this send even if it hasn't reached the socket.
			s.dispatch.Lock()
			generation := s.generation
			s.mu.Unlock()
			diagnostic.Timing("speech queue wait", item.queuedAt, "generation", generation)
			started := time.Now()
			err := s.say(item, normalPitch, currentParameters)
			diagnostic.Timing("speech send", started, "generation", generation)
			s.dispatch.Unlock()
			if err != nil {
				if !s.recover_transport(err, currentParameters) {
					return
				}
			}
		}
	}
}

func (s *Service) say(item utterance, normalPitch int, normalParameters *Parameters) error {
	if item.parameters != nil {
		if normalParameters == nil {
			return errors.New("speech parameters unavailable for preview")
		}
		if err := s.transport.Configure(*item.parameters); err != nil {
			return errors.Join(err, s.transport.Configure(*normalParameters))
		}
		err := s.transport.Say(item.text)
		return errors.Join(err, s.transport.Configure(*normalParameters))
	}
	if item.pitch == nil {
		return s.transport.Say(item.text)
	}
	if err := s.transport.Set_pitch(*item.pitch); err != nil {
		return err
	}
	err := s.transport.Say(item.text)
	return errors.Join(err, s.transport.Set_pitch(normalPitch))
}

func (s *Service) cancel_transport() error {
	s.transportMu.RLock()
	defer s.transportMu.RUnlock()
	if s.transport == nil {
		return nil
	}
	err := s.transport.Cancel()
	if err != nil {
		select {
		case s.faults <- transportFailure{epoch: s.transportEpoch, err: err}:
		default:
		}
	}
	return err
}

func (s *Service) cancel() {
	defer s.workers.Done()
	for {
		select {
		case <-s.done:
			return
		case <-s.cancelWake:
		}
		for {
			s.mu.Lock()
			generation := s.generation
			closed := s.closed
			settled := s.canceled == generation
			s.mu.Unlock()
			if closed {
				return
			}
			if settled {
				break
			}
			started := time.Now()
			if !s.dispatch.TryLock() {
				// Stop audible speech even while a stale send is blocked. A
				// second cancel after that send finishes closes the late-send race.
				_ = s.cancel_transport()
				s.dispatch.Lock()
			}
			// With dispatch held, no send can race this cancel. The usual
			// idle-send path needs only this one acknowledgement.
			_ = s.cancel_transport()
			diagnostic.Timing("speech cancel barrier", started, "generation", generation)
			s.mu.Lock()
			// Every intervening interrupt is covered: fresh sends have been
			// gated throughout. Coalesce them instead of canceling again.
			s.canceled = s.generation
			s.mu.Unlock()
			s.dispatch.Unlock()
			signal(s.wake)
			break
		}
	}
}

func (s *Service) Close() {
	s.mu.Lock()
	if s.closed {
		s.mu.Unlock()
		return
	}
	s.closed = true
	s.stopRecovery()
	s.queue = nil
	close(s.done)
	pending := s.replacements
	s.replacements = nil
	s.mu.Unlock()
	for _, request := range pending {
		request.transport.Close()
		request.done <- errors.New("speech service closed")
	}
	s.transportMu.RLock()
	if s.transport != nil {
		s.transport.Close()
	}
	s.transportMu.RUnlock()
	s.workers.Wait()
}