hyprvoice/internal/pipeline/pipeline.go

package pipeline

import (
	"context"
	"log"
	"strings"
	"sync"
	"sync/atomic"

	"github.com/leonardotrapani/hyprvoice/internal/config"
	"github.com/leonardotrapani/hyprvoice/internal/injection"
	"github.com/leonardotrapani/hyprvoice/internal/llm"
	"github.com/leonardotrapani/hyprvoice/internal/notify"
	"github.com/leonardotrapani/hyprvoice/internal/recording"
	"github.com/leonardotrapani/hyprvoice/internal/transcriber"
)

type Status string
type Action string

type PipelineError struct {
	Title   string
	Message string
	Err     error
}

const (
	Idle         Status = "idle"
	Recording    Status = "recording"
	Transcribing Status = "transcribing"
	Processing   Status = "processing" // LLM post-processing
	Injecting    Status = "injecting"
)

const (
	Inject Action = "inject"
	Cancel Action = "cancel"
)

type Pipeline interface {
	Run(ctx context.Context)
	Stop()
	Status() Status
	GetActionCh() chan<- Action
	GetErrorCh() <-chan PipelineError
	GetNotifyCh() <-chan notify.MessageType
}

// Factory types for dependency injection
type RecorderFactory func(cfg recording.Config) recording.Recorder
type TranscriberFactory func(cfg transcriber.Config) (transcriber.Transcriber, error)
type InjectorFactory func(cfg injection.Config) injection.Injector
type LLMAdapterFactory func(cfg llm.Config) (llm.Adapter, error)

// Option configures the pipeline
type Option func(*pipeline)

// WithRecorderFactory sets a custom recorder factory
func WithRecorderFactory(f RecorderFactory) Option {
	return func(p *pipeline) {
		p.recorderFactory = f
	}
}

// WithTranscriberFactory sets a custom transcriber factory
func WithTranscriberFactory(f TranscriberFactory) Option {
	return func(p *pipeline) {
		p.transcriberFactory = f
	}
}

// WithInjectorFactory sets a custom injector factory
func WithInjectorFactory(f InjectorFactory) Option {
	return func(p *pipeline) {
		p.injectorFactory = f
	}
}

// WithLLMAdapterFactory sets a custom LLM adapter factory
func WithLLMAdapterFactory(f LLMAdapterFactory) Option {
	return func(p *pipeline) {
		p.llmAdapterFactory = f
	}
}

type pipeline struct {
	status   Status
	actionCh chan Action
	errorCh  chan PipelineError
	notifyCh chan notify.MessageType
	config   *config.Config

	mu       sync.RWMutex
	wg       sync.WaitGroup
	cancel   context.CancelFunc
	stopOnce sync.Once

	running atomic.Bool

	// dependency factories (for testing)
	recorderFactory    RecorderFactory
	transcriberFactory TranscriberFactory
	injectorFactory    InjectorFactory
	llmAdapterFactory  LLMAdapterFactory
}

func New(cfg *config.Config, opts ...Option) Pipeline {
	p := &pipeline{
		actionCh: make(chan Action, 1),
		errorCh:  make(chan PipelineError, 10),
		notifyCh: make(chan notify.MessageType, 10),
		config:   cfg,
		// default factories
		recorderFactory:    recording.NewRecorder,
		transcriberFactory: transcriber.NewTranscriber,
		injectorFactory:    injection.NewInjector,
		llmAdapterFactory:  llm.NewAdapter,
	}

	for _, opt := range opts {
		opt(p)
	}

	return p
}
func (p *pipeline) Run(ctx context.Context) {
	if !p.running.CompareAndSwap(false, true) {
		log.Printf("Pipeline: Already running, ignoring Run() call")
		return
	}

	runCtx, cancel := context.WithTimeout(ctx, p.config.Recording.Timeout)
	p.setCancel(cancel)

	p.wg.Add(1)
	go p.run(runCtx)
}

func (p *pipeline) run(ctx context.Context) {
	defer func() {
		p.running.Store(false)
		p.setStatus(Idle)
		p.wg.Done()
	}()

	log.Printf("Pipeline: Starting recording")
	p.setStatus(Recording)

	recorder := p.recorderFactory(p.config.ToRecordingConfig())
	frameCh, rErrCh, err := recorder.Start(ctx)

	if err != nil {
		log.Printf("Pipeline: Recording error: %v", err)
		p.sendError("Recording Error", "Failed to start recording", err)
		return
	}

	defer recorder.Stop()

	t, err := p.transcriberFactory(p.config.ToTranscriberConfig())
	if err != nil {
		log.Printf("Pipeline: Failed to create transcriber: %v", err)
		p.sendError("Transcription Error", "Failed to create transcriber", err)
		return
	}

	log.Printf("Pipeline: Starting transcriber")
	p.setStatus(Transcribing)

	tErrCh, err := t.Start(ctx, frameCh)
	if err != nil {
		log.Printf("Pipeline: Transcriber error: %v", err)
		p.sendError("Transcription Error", "Failed to start transcriber", err)
		return
	}

	defer func() {
		if stopErr := t.Stop(ctx); stopErr != nil {
			log.Printf("Pipeline: Error stopping transcriber: %v", stopErr)
			// Silently call an error now because on simple transcriber we just transcribe all audio when we stop, and might fail when force stop
			//p.sendError("Transcription Error", "Failed to stop transcriber cleanly", stopErr)
		}
	}()

	// Forward errors from component channels to unified pipeline error channel
	go func() {
		for err := range tErrCh {
			p.sendError("Transcription Error", "Transcription processing error", err)
		}
	}()

	go func() {
		for err := range rErrCh {
			p.sendError("Recording Error", "Recording stream error", err)
		}
	}()

	for {
		select {
		case action := <-p.actionCh:
			switch action {
			case Inject:
				p.handleInjectAction(ctx, recorder, t)
				return
			}

		case <-ctx.Done():
			return
		}
	}
}

func (p *pipeline) Status() Status {
	p.mu.RLock()
	defer p.mu.RUnlock()
	return p.status
}

func (p *pipeline) setStatus(status Status) {
	p.mu.Lock()
	defer p.mu.Unlock()
	p.status = status
}

func (p *pipeline) setCancel(cancel context.CancelFunc) {
	p.mu.Lock()
	defer p.mu.Unlock()
	p.cancel = cancel
}

func (p *pipeline) getCancel() context.CancelFunc {
	p.mu.RLock()
	defer p.mu.RUnlock()
	return p.cancel
}

func (p *pipeline) GetActionCh() chan<- Action {
	p.mu.RLock()
	defer p.mu.RUnlock()
	return p.actionCh
}

func (p *pipeline) GetErrorCh() <-chan PipelineError {
	p.mu.RLock()
	defer p.mu.RUnlock()
	return p.errorCh
}

func (p *pipeline) GetNotifyCh() <-chan notify.MessageType {
	p.mu.RLock()
	defer p.mu.RUnlock()
	return p.notifyCh
}

func (p *pipeline) sendError(title, message string, err error) {
	pipelineErr := PipelineError{
		Title:   title,
		Message: message,
		Err:     err,
	}

	select {
	case p.errorCh <- pipelineErr:
	default:
		log.Printf("Pipeline: Error channel full, dropping error: %s", message)
	}
}

func (p *pipeline) sendNotify(mt notify.MessageType) {
	select {
	case p.notifyCh <- mt:
	default:
		log.Printf("Pipeline: Notify channel full, dropping notification")
	}
}

func (p *pipeline) handleInjectAction(ctx context.Context, recorder recording.Recorder, t transcriber.Transcriber) {
	status := p.Status()

	if status != Transcribing {
		log.Printf("Pipeline: Inject action received, but not in transcribing state, ignoring")
		return
	}

	log.Printf("Pipeline: Inject action received, stopping recording and finalizing transcription")
	p.setStatus(Injecting)

	recorder.Stop()

	if err := t.Stop(ctx); err != nil {
		p.sendError("Transcription Error", "Failed to stop transcriber during injection", err)
		return
	}

	transcriptionText, err := t.GetFinalTranscription()
	if err != nil {
		p.sendError("Transcription Error", "Failed to retrieve transcription", err)
		return
	}
	log.Printf("Pipeline: Final transcription text: %s", transcriptionText)

	// LLM post-processing phase
	textToInject := transcriptionText
	if p.config.IsLLMEnabled() {
		p.setStatus(Processing)
		p.sendNotify(notify.MsgLLMProcessing)
		log.Printf("Pipeline: LLM post-processing enabled, processing text")

		llmCfg := p.config.ToLLMConfig()
		adapter, err := p.llmAdapterFactory(llm.Config{
			Provider:          llmCfg.Provider,
			APIKey:            llmCfg.APIKey,
			Model:             llmCfg.Model,
			RemoveStutters:    llmCfg.RemoveStutters,
			AddPunctuation:    llmCfg.AddPunctuation,
			FixGrammar:        llmCfg.FixGrammar,
			RemoveFillerWords: llmCfg.RemoveFillerWords,
			CustomPrompt:      llmCfg.CustomPrompt,
			Keywords:          llmCfg.Keywords,
			BaseURL:           llmCfg.BaseURL,
		})
		if err != nil {
			log.Printf("Pipeline: Failed to create LLM adapter: %v, using raw transcription", err)
		} else {
			processed, err := adapter.Process(ctx, transcriptionText)
			if err != nil {
				log.Printf("Pipeline: LLM processing failed: %v, using raw transcription", err)
			} else {
				textToInject = processed
				log.Printf("Pipeline: LLM processed text: %s", textToInject)
			}
		}
		p.setStatus(Injecting)
	}

	// Sanitize: replace line-terminating characters with spaces to prevent
	// unintended Enter keypresses during injection, which can submit forms mid-sentence.
	// Covers ASCII controls (\r, \n, \v, \f), Unicode NEL (U+0085),
	// LINE SEPARATOR (U+2028), and PARAGRAPH SEPARATOR (U+2029).
	textToInject = strings.Map(func(r rune) rune {
		switch r {
		case '\r', '\n', '\v', '\f', '\u0085', '\u2028', '\u2029':
			return ' '
		}
		return r
	}, textToInject)
	// Speech APIs occasionally include leading or trailing whitespace (for
	// example Whisper commonly returns a leading space). It is transport
	// formatting rather than dictated content, so never inject it.
	textToInject = strings.TrimSpace(textToInject)
	injector := p.injectorFactory(p.config.ToInjectionConfig())

	if err := injector.Inject(ctx, textToInject); err != nil {
		p.sendError("Injection Error", "Failed to inject text", err)
	} else {
		log.Printf("Pipeline: Text injection completed successfully")
	}

	p.setStatus(Idle)
}

func (p *pipeline) Stop() {
	p.stopOnce.Do(func() {
		cancel := p.getCancel()
		if cancel != nil {
			cancel()
		}
	})
	p.wg.Wait()
}