talk/cmd/talkd/main.go

// talkd ist der Daemon für talk 2.0: HTTP-API auf 127.0.0.1 (Default-Port
// 7800, server.port/TALK_PORT), serialisierte Ansage-Queue mit Prioritäten
// und now-Sprung, persistenter Mute-State (Datei mute im Config-Dir) —
// und seit Task 05 das Anti-Spam-Gesetz: Rate-Limit (queue.rate_limit,
// Default 1 Ansage / 30 s; low über Budget wird verworfen, normal parkt,
// high darf mit queue.allow_high_burst verstoßen) plus Ruhezeiten
// (quiet_hours/quiet_mode; now:true bricht bewusst aus beiden aus).
// Task 06 ergänzt die CORS-Tür für POST /speak (server.allowed_origins /
// TALK_ALLOWED_ORIGINS; leer = kein CORS) und den MCP-Endpoint POST /mcp
// (Streamable HTTP, ein speak-Tool).
package main

import (
	"context"
	"fmt"
	"log"
	"net"
	"net/http"
	"os"
	"os/signal"
	"path/filepath"
	"strings"
	"sync"
	"syscall"
	"time"

	"code.m3ta.dev/m3tam3re/talk/internal/api"
	"code.m3ta.dev/m3tam3re/talk/internal/audio"
	"code.m3ta.dev/m3tam3re/talk/internal/config"
	"code.m3ta.dev/m3tam3re/talk/internal/langdetect"
	"code.m3ta.dev/m3tam3re/talk/internal/provider"
	"code.m3ta.dev/m3tam3re/talk/internal/queue"
)

// Build-Informationen. Werden beim Release per -ldflags injiziert (gleicher
// Vertrag wie cmd/talk, siehe .goreleaser.yaml in Task 11). Default: dev.
var (
	version = "dev"
	commit  = ""
)

func main() {
	log.SetFlags(log.LstdFlags)
	log.SetPrefix("talkd: ")
	if err := run(); err != nil {
		log.Fatal(err)
	}
}

// run startet den Daemon und kehrt erst beim Herunterfahren (SIGINT/
// SIGTERM) zurück. Abbrüche der laufenden Ansage beim Shutdown sind
// beabsichtigt: talkd soll auf Signale prompt reagieren.
func run() error {
	getenv := os.Getenv

	cfg, err := config.Load(config.Options{Getenv: getenv})
	if err != nil {
		return err
	}

	// Audio-Sink (audio.sink; "null" verwirft tonlos — CI/Server).
	// Ketten-, Queue- und Sink-Protokoll laufen über denselben Logger.
	lw := lineLogger{}
	sink, err := audio.NewSink(cfg.Audio.Sink, lw)
	if err != nil {
		return err
	}

	// Basiskette nur für die /status-Auflösung „aktiver Provider" (erste
	// verfügbare Stufe; Available ist billig — Key-/Endpunkt-Binary-Checks
	// ohne Netz-Call). Das eigentliche Abspielen baut pro Item eine eigene
	// Kette (Pin/Voice je Anfrage) — die Kette aus internal/provider wird
	// unverändert wiederverwendet.
	baseChain, err := provider.NewChain(chainConfigFor(cfg, "", "", getenv, nil))
	if err != nil {
		return err
	}

	state := &providerState{}
	speak := func(ctx context.Context, item queue.Item) error {
		state.set(resolveProviderName(baseChain, item))
		lang := item.Lang
		if lang == "" {
			lang = langdetect.Detect(item.Text)
		}
		chain, err := provider.NewChain(chainConfigFor(cfg, item.Provider, item.Voice, getenv, sink))
		if err != nil {
			return err
		}
		return chain.Speak(ctx, item.Text, lang)
	}

	// Anti-Spam-Gesetz (Task 05): Rate-Limit + High-Burst + Ruhezeiten.
	// config.Load hat quiet_hours bereits validiert — das Parsen hier kann
	// praktisch nicht scheitern, aber ein Fehler darf nie still werden.
	quiet, err := queue.ParseQuietHours(cfg.QuietHours)
	if err != nil {
		return err
	}
	laws := queue.Laws{
		RateLimit:      cfg.Queue.RateLimit,
		AllowHighBurst: cfg.Queue.AllowHighBurst,
		Quiet:          quiet,
		QuietMode:      cfg.QuietMode,
	}

	q := queue.New(laws, speak, lw)

	// Persistenter Mute-State (Datei mute im Config-Dir). Ohne Config-Dir
	// läuft der Daemon ohne Mute weiter (dokumentierte Warnung) statt zu
	// scheitern — alles andere an talk darf ja auch ohne Config-Datei.
	var mutePath string
	if dir, err := config.Dir(getenv); err == nil {
		mutePath = filepath.Join(dir, queue.MuteFileName)
	} else {
		log.Printf("warnung: Config-Dir nicht bestimmbar — Mute-State deaktiviert (%v)", err)
	}
	mute := queue.MuteStore{Path: mutePath}

	handler := api.NewHandler(api.HandlerConfig{
		Queue:          q,
		Mute:           mute,
		Version:        versionString(),
		KnownProviders: provider.Names(),
		AllowedOrigins: cfg.Server.AllowedOrigins,
		Provider: func() string {
			if cur, ok := q.Current(); ok {
				return resolveProviderName(baseChain, cur)
			}
			if last := state.get(); last != "" {
				return last
			}
			return resolveProviderName(baseChain, queue.Item{})
		},
	})

	addr := fmt.Sprintf("127.0.0.1:%d", cfg.Server.Port)
	ln, err := net.Listen("tcp", addr)
	if err != nil {
		return fmt.Errorf("Port belegen (%s): %w — läuft bereits ein talkd?", addr, err)
	}
	srv := &http.Server{Handler: handler, ReadHeaderTimeout: 5 * time.Second}
	errCh := make(chan error, 1)
	go func() { errCh <- srv.Serve(ln) }()
	corsLog := "zu (kein CORS)"
	if len(cfg.Server.AllowedOrigins) > 0 {
		corsLog = fmt.Sprintf("%v", cfg.Server.AllowedOrigins)
	}
	log.Printf("listening on http://%s (version %s, audio sink %s, mute=%v, rate-limit %s, high-burst=%v, ruhezeit=%q modus %s, cors=%s, mcp=/mcp)",
		addr, versionString(), sink.Name(), mute.Muted(),
		cfg.Queue.RateLimit, cfg.Queue.AllowHighBurst, cfg.QuietHours, cfg.QuietMode, corsLog)

	sig := make(chan os.Signal, 1)
	signal.Notify(sig, os.Interrupt, syscall.SIGTERM)

	select {
	case err := <-errCh:
		q.Close()
		return fmt.Errorf("http server: %w", err)
	case s := <-sig:
		log.Printf("Signal %v empfangen — fahre herunter (laufende Ansage wird abgebrochen)", s)
	}

	// Reihenfolge: erst HTTP dicht (keine neuen Anfragen), dann die Queue
	// schließen (laufende Ansage abbrechen, Worker beenden und darauf
	// warten). So läuft nie eine Ansage nach dem Daemon-Ende weiter.
	shutdownCtx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()
	if err := srv.Shutdown(shutdownCtx); err != nil {
		log.Printf("http shutdown: %v", err)
	}
	q.Close()
	return nil
}

// versionString rendert die Daemon-Version für /status.
func versionString() string {
	if commit == "" {
		return version
	}
	return fmt.Sprintf("%s (%s)", version, commit)
}

// chainConfigFor überführt Config + Item-Pins in die ChainConfig — dieselbe
// Zusammensetzung wie im CLI-Pfad (cmd/talk/say.go), nur dass der Daemon
// sie pro Ansage baut (Pin/Voice sind Anfrage-Properties). player ist nur
// für die Abspiel-Kette gesetzt (die Basiskette braucht ihn nicht).
func chainConfigFor(cfg config.Config, pin, voice string, getenv func(string) string, player provider.Player) provider.ChainConfig {
	return provider.ChainConfig{
		Pin:       pickProvider(pin, cfg.Provider),
		Voice:     voice,
		Voices:    cfg.Voice,
		Endpoints: endpointsOf(cfg),
		Model:     cfg.Providers["openai"].Model,
		Getenv:    getenv,
		Player:    player,
		Log:       lineLogger{},
	}
}

// pickProvider löst den Provider-Pin auf: Anfrage-/Flag-Pin schlägt
// `provider = "…"` aus der config.toml; leer = komplette Default-Kette.
func pickProvider(pin, cfgValue string) string {
	if pin != "" {
		return pin
	}
	return cfgValue
}

// endpointsOf überführt die Provider-Einstellungen der Config in die
// Endpunkt-Map der Kette.
func endpointsOf(cfg config.Config) map[string]string {
	endpoints := make(map[string]string, len(cfg.Providers))
	for name, settings := range cfg.Providers {
		if settings.Endpoint != "" {
			endpoints[name] = settings.Endpoint
		}
	}
	return endpoints
}

// resolveProviderName löst den „aktiven Provider" für /status auf: Pin des
// Items, sonst erste verfügbare Stufe der Basiskette. Available() ist
// bewusst billig (Key-/Binary-Prüfung ohne Netz-Call), deshalb darf das
// /status on demand aufrufen.
func resolveProviderName(chain *provider.Chain, item queue.Item) string {
	if item.Provider != "" {
		return item.Provider
	}
	for _, p := range chain.Tiers() {
		if p.Available(context.Background()) {
			return p.Name()
		}
	}
	return "keine"
}

// providerState merkt sich den zuletzt aktiven Provider (für /status im
// Leerlauf). Geschrieben wird ausschließlich vom Queue-Worker, gelesen von
// HTTP-Handlern — daher der eigene kleine Mutex.
type providerState struct {
	mu   sync.Mutex
	name string
}

func (s *providerState) set(name string) {
	s.mu.Lock()
	s.name = name
	s.mu.Unlock()
}

func (s *providerState) get() string {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.name
}

// lineLogger reicht Zeilen an den Standard-Logger weiter (Zeitstempel +
// talkd-Präfix), statt sie roh zu schreiben: Provider-Kette, Queue und
// null-Sink schreiben zeilenweise, so entsteht ein einheitliches Log.
type lineLogger struct{}

func (lineLogger) Write(p []byte) (int, error) {
	for _, line := range strings.Split(strings.TrimRight(string(p), "\n"), "\n") {
		if line != "" {
			log.Print(line)
		}
	}
	return len(p), nil
}