talk/internal/daemon/daemon.go

// Package daemon ist der Hintergrundprozess von talk: HTTP-API auf
// 127.0.0.1, serialisierte Ansage-Queue mit Prioritäten und now-Sprung,
// persistenter Mute-State, die Anti-Spam-Gesetze (Rate-Limit, Ruhezeiten),
// CORS-Tür und MCP-Endpoint. Es gibt genau ein Binary: `talk daemon run`
// betreibt den Daemon im Vordergrund, die CLI startet ihn mit Spawn bei
// Bedarf selbst im Hintergrund (daemon.autostart).
package daemon

import (
	"context"
	"fmt"
	"io"
	"log"
	"net"
	"net/http"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"sync"
	"time"

	"code.m3ta.dev/m3tam3re/talk/internal/api"
	"code.m3ta.dev/m3tam3re/talk/internal/config"
	"code.m3ta.dev/m3tam3re/talk/internal/notify"
	"code.m3ta.dev/m3tam3re/talk/internal/provider"
	"code.m3ta.dev/m3tam3re/talk/internal/queue"
	"code.m3ta.dev/m3tam3re/talk/internal/speak"
)

// Dateien des Daemons im Config-Verzeichnis.
const (
	// LogFileName ist das Protokoll eines im Hintergrund gestarteten
	// Daemons (Spawn leitet stdout/stderr dorthin; pro Start neu).
	LogFileName = "daemon.log"

	// PIDFileName hält die Prozess-ID des laufenden Daemons (für
	// Diagnose und als Notbremse, wenn die HTTP-API nicht antwortet).
	PIDFileName = "daemon.pid"
)

// Options sind die Abhängigkeiten von Run.
type Options struct {
	// Config ist die geladene Konfiguration (Port, Gesetze, Sink, …).
	Config config.Config

	// Getenv liefert Keys und Endpunkt-Overrides (nil = os.Getenv).
	Getenv func(string) string

	// Version ist die Daemon-Version für /status und MCP serverInfo.
	Version string

	// Log empfängt das Protokoll (Vordergrund: stderr; Hintergrund: die
	// Log-Datei). nil = os.Stderr.
	Log io.Writer
}

// Run betreibt den Daemon, bis ctx endet (Signal), POST /shutdown kommt
// oder der HTTP-Server scheitert. Eine laufende Ansage wird beim
// Herunterfahren abgebrochen — der Daemon soll prompt reagieren. Ein
// belegter Port ist ein klarer Fehler („läuft bereits ein Daemon?").
func Run(ctx context.Context, opts Options) error {
	getenv := opts.Getenv
	if getenv == nil {
		getenv = os.Getenv
	}
	out := opts.Log
	if out == nil {
		out = os.Stderr
	}
	logger := log.New(out, "talk daemon: ", log.LstdFlags)
	lw := lineWriter{logger}
	cfg := opts.Config

	engine, err := speak.New(cfg, getenv, lw)
	if err != nil {
		return err
	}

	// Anti-Spam-Gesetz: config.Load hat quiet_hours validiert — ein Fehler
	// hier ist praktisch unmöglich, darf aber nie still werden.
	quiet, err := queue.ParseQuietHours(cfg.QuietHours)
	if err != nil {
		return err
	}
	q := queue.New(queue.Laws{
		RateLimit:      cfg.Queue.RateLimit,
		AllowHighBurst: cfg.Queue.AllowHighBurst,
		Quiet:          quiet,
		QuietMode:      cfg.QuietMode,
		MaxText:        cfg.Queue.MaxText,
		MaxAge:         cfg.Queue.MaxAge,
	}, func(ctx context.Context, item queue.Item) error {
		out, err := engine.Speak(ctx, speak.Request{
			Text:     item.Text,
			Lang:     item.Lang,
			Provider: item.Provider,
			Voice:    item.Voice,
		})
		if err == nil && !out.Discarded {
			logger.Printf("ansage %s über %s (synthese %s, wiedergabe %s)", item.ID, out.Provider,
				out.Synthesis.Round(time.Millisecond), out.Playback.Round(time.Millisecond))
		}
		return err
	}, lw)

	// Ruhezeit-Toast: Desktop-Benachrichtigung statt Audio (quiet_mode =
	// "toast"); best effort, Fehler nur im Protokoll. Läuft asynchron, damit
	// ein hängender Notification-Daemon nie die Queue blockiert.
	if cfg.QuietMode == queue.QuietModeToast {
		if notify.Available() {
			q.OnToast(func(item queue.Item) {
				go func() {
					if err := notify.Show(context.Background(), item.Text); err != nil {
						logger.Printf("toast: %v", err)
					}
				}()
			})
		} else {
			logger.Printf("warnung: quiet_mode = toast, aber %s — Ruhezeit-Ansagen werden nur protokolliert", notify.Describe())
		}
	}

	// Persistenter Mute-State und PID-Datei im Config-Dir. Ohne Config-Dir
	// läuft der Daemon ohne beides weiter (Warnung) statt zu scheitern.
	var mutePath, pidPath string
	if dir, err := config.Dir(getenv); err == nil {
		mutePath = filepath.Join(dir, queue.MuteFileName)
		pidPath = filepath.Join(dir, PIDFileName)
	} else {
		logger.Printf("warnung: Config-Dir nicht bestimmbar — Mute-State und PID-Datei deaktiviert (%v)", err)
	}
	mute := queue.MuteStore{Path: mutePath}

	shutdown := make(chan struct{})
	var once sync.Once
	handler := api.NewHandler(api.HandlerConfig{
		Queue:          q,
		Mute:           mute,
		Version:        opts.Version,
		KnownProviders: provider.Names(),
		AllowedOrigins: cfg.Server.AllowedOrigins,
		Provider: func() string {
			if cur, ok := q.Current(); ok {
				return engine.ActiveProvider(cur.Provider)
			}
			if last := engine.LastProvider(); last != "" {
				return last
			}
			return engine.ActiveProvider("")
		},
		Shutdown: func() { once.Do(func() { close(shutdown) }) },
	})

	addr := fmt.Sprintf("127.0.0.1:%d", cfg.Server.Port)
	ln, err := net.Listen("tcp", addr)
	if err != nil {
		q.Close()
		return fmt.Errorf("Port belegen (%s): %w — läuft bereits ein talk daemon?", addr, err)
	}
	if pidPath != "" {
		writePID(pidPath, logger)
		// Die PID-Datei verschwindet als Letztes — `talk daemon stop`
		// wartet darauf und weiß dann: Prozess ist fertig, nicht nur HTTP.
		defer os.Remove(pidPath)
	}

	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)
	}
	logger.Printf("listening on http://%s (version %s, pid %d, audio sink %s, mute=%v, rate-limit %s, high-burst=%v, max-text %d, max-age %s, ruhezeit=%q modus %s, cors=%s, mcp=/mcp)",
		addr, opts.Version, os.Getpid(), engine.SinkName(), mute.Muted(),
		cfg.Queue.RateLimit, cfg.Queue.AllowHighBurst, cfg.Queue.MaxText, cfg.Queue.MaxAge, cfg.QuietHours, cfg.QuietMode, corsLog)

	select {
	case err := <-errCh:
		q.Close()
		return fmt.Errorf("http server: %w", err)
	case <-ctx.Done():
		logger.Printf("Signal empfangen — fahre herunter (laufende Ansage wird abgebrochen)")
	case <-shutdown:
		logger.Printf("Shutdown angefordert (talk daemon stop) — fahre herunter")
	}

	// 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 {
		logger.Printf("http shutdown: %v", err)
	}
	q.Close()
	return nil
}

// writePID schreibt die eigene Prozess-ID (best effort; ein Fehler ist nur
// eine Protokollzeile — der Daemon funktioniert auch ohne PID-Datei).
func writePID(path string, logger *log.Logger) {
	if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
		logger.Printf("warnung: PID-Datei: %v", err)
		return
	}
	if err := os.WriteFile(path, []byte(strconv.Itoa(os.Getpid())+"\n"), 0o644); err != nil {
		logger.Printf("warnung: PID-Datei: %v", err)
	}
}

// ReadPID liest die PID-Datei; 0, wenn keine (lesbare) Datei existiert.
func ReadPID(dir string) int {
	b, err := os.ReadFile(filepath.Join(dir, PIDFileName))
	if err != nil {
		return 0
	}
	pid, err := strconv.Atoi(strings.TrimSpace(string(b)))
	if err != nil || pid <= 0 {
		return 0
	}
	return pid
}

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

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