feat(talkd): Daemon mit HTTP-API auf 127.0.0.1, Queue-Betrieb und Shutdown
- HTTP-Server nur auf 127.0.0.1 (Port aus config server.port/TALK_PORT), klarer Fehler wenn der Port belegt ist - Ansagen laufen über die unveränderte Provider-Kette (pro Item gebaut, Pin/Voice je Anfrage); /status meldet den aktiven Provider - SIGINT/SIGTERM: erst HTTP dicht, dann Queue schließen (laufende Ansage wird abgebrochen, Worker-Exit wird abgewartet) - Mute ohne Config-Dir: dokumentierte Warnung statt Absturz
This commit is contained in:
m3ta-chiron committed
1 parent
39c3563195
commit
d09aff6874
1 file changed
+242
-7
+242
-7
@@ -1,13 +1,248 @@
|
||||
// talkd ist der Daemon für talk 2.0 (Queue, Mute-State, HTTP-API, MCP).
|
||||
// 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).
|
||||
//
|
||||
// Dieser Stand (Task 01) ist ein Platzhalter: talkd startet, loggt eine Zeile
|
||||
// und beendet sich mit Exit-Code 0. Die eigentliche Daemon-Implementierung
|
||||
// (HTTP-Server auf 127.0.0.1, serialisierte Ansage-Queue, persistentes Mute)
|
||||
// entsteht in Task 04.
|
||||
// Task 04: POST /speak, GET /status, POST /cancel, POST /mute, POST /unmute.
|
||||
// Rate-Limit/Ruhezeiten folgen in Task 05, CORS/MCP in Task 06 — die
|
||||
// Rate-Limit-Struktur existiert bereits (internal/queue/rate.go), ist hier
|
||||
// aber bewusst noch nicht angebunden (Task-04-Default: unbegrenzt).
|
||||
package main
|
||||
|
||||
import "log"
|
||||
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.Println("talkd: Platzhalter — Daemon-Implementierung folgt in Task 04")
|
||||
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)
|
||||
}
|
||||
|
||||
q := queue.New(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)",
|
||||
addr, versionString(), sink.Name(), mute.Muted())
|
||||
|
||||
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
|
||||
}
|
||||
Reference in new issue
Block a user