// 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). // CORS/MCP folgen in Task 06. 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(), 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) }() log.Printf("listening on http://%s (version %s, audio sink %s, mute=%v, rate-limit %s, high-burst=%v, ruhezeit=%q modus %s)", addr, versionString(), sink.Name(), mute.Muted(), cfg.Queue.RateLimit, cfg.Queue.AllowHighBurst, cfg.QuietHours, cfg.QuietMode) 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 }