fm-rds-tx/internal/control/control.go

package control

import (
	_ "embed"
	"encoding/json"
	"errors"
	"io"
	"mime"
	"net/http"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/jan/fm-rds-tx/internal/audio"
	"github.com/jan/fm-rds-tx/internal/config"
	drypkg "github.com/jan/fm-rds-tx/internal/dryrun"
	"github.com/jan/fm-rds-tx/internal/ingest"
	"github.com/jan/fm-rds-tx/internal/platform"
)

//go:embed ui.html
var uiHTML []byte

// TXController is an optional interface the Server uses to start/stop TX
// and apply live config changes.
type TXController interface {
	StartTX() error
	StopTX() error
	TXStats() map[string]any
	UpdateConfig(patch LivePatch) error
	ResetFault() error
}

// LivePatch mirrors the patchable fields from ConfigPatch for the engine.
// nil = no change.
type LivePatch struct {
	FrequencyMHz   *float64
	OutputDrive    *float64
	StereoEnabled  *bool
	PilotLevel     *float64
	RDSInjection   *float64
	RDSEnabled     *bool
	LimiterEnabled *bool
	LimiterCeiling *float64
	PS             *string
	RadioText      *string
}

type Server struct {
	mu           sync.RWMutex
	cfg          config.Config
	tx           TXController
	drv          platform.SoapyDriver // optional, for runtime stats
	streamSrc    *audio.StreamSource  // optional, for live audio ring stats
	audioIngress AudioIngress         // optional, for /audio/stream
	ingestRt     IngestRuntime        // optional, for /runtime ingest stats
	saveConfig   func(config.Config) error
	hardReload   func()
	audit        auditCounters
}

type AudioIngress interface {
	WritePCM16(data []byte) (int, error)
}

type IngestRuntime interface {
	Stats() ingest.Stats
}

type auditEvent string

const (
	auditMethodNotAllowed     auditEvent = "methodNotAllowed"
	auditUnsupportedMediaType auditEvent = "unsupportedMediaType"
	auditBodyTooLarge         auditEvent = "bodyTooLarge"
	auditUnexpectedBody       auditEvent = "unexpectedBody"
)

type auditCounters struct {
	methodNotAllowed     uint64
	unsupportedMediaType uint64
	bodyTooLarge         uint64
	unexpectedBody       uint64
}

const (
	maxConfigBodyBytes          = 64 << 10 // 64 KiB
	configContentTypeHeader     = "application/json"
	noBodyErrMsg                = "request must not include a body"
	audioStreamContentTypeError = "Content-Type must be application/octet-stream or audio/L16"
	audioStreamBodyLimitDefault = 512 << 20 // 512 MiB
)

var audioStreamAllowedMediaTypes = []string{
	"application/octet-stream",
	"audio/l16",
}

var audioStreamBodyLimit = int64(audioStreamBodyLimitDefault) // bytes allowed per /audio/stream request; tests may override.

func isJSONContentType(r *http.Request) bool {
	ct := strings.TrimSpace(r.Header.Get("Content-Type"))
	if ct == "" {
		return false
	}
	mediaType, _, err := mime.ParseMediaType(ct)
	if err != nil {
		return false
	}
	return strings.EqualFold(mediaType, configContentTypeHeader)
}

type ConfigPatch struct {
	FrequencyMHz     *float64 `json:"frequencyMHz,omitempty"`
	OutputDrive      *float64 `json:"outputDrive,omitempty"`
	StereoEnabled    *bool    `json:"stereoEnabled,omitempty"`
	PilotLevel       *float64 `json:"pilotLevel,omitempty"`
	RDSInjection     *float64 `json:"rdsInjection,omitempty"`
	RDSEnabled       *bool    `json:"rdsEnabled,omitempty"`
	ToneLeftHz       *float64 `json:"toneLeftHz,omitempty"`
	ToneRightHz      *float64 `json:"toneRightHz,omitempty"`
	ToneAmplitude    *float64 `json:"toneAmplitude,omitempty"`
	PS               *string  `json:"ps,omitempty"`
	RadioText        *string  `json:"radioText,omitempty"`
	PreEmphasisTauUS *float64 `json:"preEmphasisTauUS,omitempty"`
	LimiterEnabled   *bool    `json:"limiterEnabled,omitempty"`
	LimiterCeiling   *float64 `json:"limiterCeiling,omitempty"`
}

type IngestSaveRequest struct {
	Ingest config.IngestConfig `json:"ingest"`
}

func NewServer(cfg config.Config) *Server {
	return &Server{cfg: cfg}
}

func hasRequestBody(r *http.Request) bool {
	if r.ContentLength > 0 {
		return true
	}
	for _, te := range r.TransferEncoding {
		if strings.EqualFold(te, "chunked") {
			return true
		}
	}
	return false
}

func (s *Server) rejectBody(w http.ResponseWriter, r *http.Request) bool {
	if !hasRequestBody(r) {
		return true
	}
	s.recordAudit(auditUnexpectedBody)
	http.Error(w, noBodyErrMsg, http.StatusBadRequest)
	return false
}

func (s *Server) recordAudit(evt auditEvent) {
	switch evt {
	case auditMethodNotAllowed:
		atomic.AddUint64(&s.audit.methodNotAllowed, 1)
	case auditUnsupportedMediaType:
		atomic.AddUint64(&s.audit.unsupportedMediaType, 1)
	case auditBodyTooLarge:
		atomic.AddUint64(&s.audit.bodyTooLarge, 1)
	case auditUnexpectedBody:
		atomic.AddUint64(&s.audit.unexpectedBody, 1)
	}
}

func (s *Server) auditSnapshot() map[string]uint64 {
	return map[string]uint64{
		"methodNotAllowed":     atomic.LoadUint64(&s.audit.methodNotAllowed),
		"unsupportedMediaType": atomic.LoadUint64(&s.audit.unsupportedMediaType),
		"bodyTooLarge":         atomic.LoadUint64(&s.audit.bodyTooLarge),
		"unexpectedBody":       atomic.LoadUint64(&s.audit.unexpectedBody),
	}
}

func isAudioStreamContentType(r *http.Request) bool {
	ct := strings.TrimSpace(r.Header.Get("Content-Type"))
	if ct == "" {
		return false
	}
	mediaType, _, err := mime.ParseMediaType(ct)
	if err != nil {
		return false
	}
	for _, allowed := range audioStreamAllowedMediaTypes {
		if strings.EqualFold(mediaType, allowed) {
			return true
		}
	}
	return false
}

func (s *Server) SetTXController(tx TXController) {
	s.mu.Lock()
	s.tx = tx
	s.mu.Unlock()
}

func (s *Server) SetDriver(drv platform.SoapyDriver) {
	s.mu.Lock()
	s.drv = drv
	s.mu.Unlock()
}

func (s *Server) SetStreamSource(src *audio.StreamSource) {
	s.mu.Lock()
	s.streamSrc = src
	s.mu.Unlock()
}

func (s *Server) SetAudioIngress(ingress AudioIngress) {
	s.mu.Lock()
	s.audioIngress = ingress
	s.mu.Unlock()
}

func (s *Server) SetIngestRuntime(rt IngestRuntime) {
	s.mu.Lock()
	s.ingestRt = rt
	s.mu.Unlock()
}

func (s *Server) SetConfigSaver(save func(config.Config) error) {
	s.mu.Lock()
	s.saveConfig = save
	s.mu.Unlock()
}

func (s *Server) SetHardReload(fn func()) {
	s.mu.Lock()
	s.hardReload = fn
	s.mu.Unlock()
}

func (s *Server) Handler() http.Handler {
	mux := http.NewServeMux()
	mux.HandleFunc("/", s.handleUI)
	mux.HandleFunc("/healthz", s.handleHealth)
	mux.HandleFunc("/status", s.handleStatus)
	mux.HandleFunc("/dry-run", s.handleDryRun)
	mux.HandleFunc("/config", s.handleConfig)
	mux.HandleFunc("/config/ingest/save", s.handleIngestSave)
	mux.HandleFunc("/runtime", s.handleRuntime)
	mux.HandleFunc("/runtime/fault/reset", s.handleRuntimeFaultReset)
	mux.HandleFunc("/tx/start", s.handleTXStart)
	mux.HandleFunc("/tx/stop", s.handleTXStop)
	mux.HandleFunc("/audio/stream", s.handleAudioStream)
	return mux
}

func (s *Server) handleHealth(w http.ResponseWriter, _ *http.Request) {
	w.Header().Set("Content-Type", "application/json")
	_ = json.NewEncoder(w).Encode(map[string]any{"ok": true})
}

func (s *Server) handleUI(w http.ResponseWriter, r *http.Request) {
	if r.URL.Path != "/" {
		http.NotFound(w, r)
		return
	}
	w.Header().Set("Content-Type", "text/html; charset=utf-8")
	w.Header().Set("Cache-Control", "no-cache")
	w.Write(uiHTML)
}

func (s *Server) handleStatus(w http.ResponseWriter, _ *http.Request) {
	s.mu.RLock()
	cfg := s.cfg
	tx := s.tx
	s.mu.RUnlock()

	status := map[string]any{
		"service":             "fm-rds-tx",
		"backend":             cfg.Backend.Kind,
		"frequencyMHz":        cfg.FM.FrequencyMHz,
		"stereoEnabled":       cfg.FM.StereoEnabled,
		"rdsEnabled":          cfg.RDS.Enabled,
		"preEmphasisTauUS":    cfg.FM.PreEmphasisTauUS,
		"limiterEnabled":      cfg.FM.LimiterEnabled,
		"fmModulationEnabled": cfg.FM.FMModulationEnabled,
	}
	if tx != nil {
		if stats := tx.TXStats(); stats != nil {
			if ri, ok := stats["runtimeIndicator"]; ok {
				status["runtimeIndicator"] = ri
			}
			if alert, ok := stats["runtimeAlert"]; ok {
				status["runtimeAlert"] = alert
			}
			if queue, ok := stats["queue"]; ok {
				status["queue"] = queue
			}
			if runtimeState, ok := stats["state"]; ok {
				status["runtimeState"] = runtimeState
			}
		}
	}

	w.Header().Set("Content-Type", "application/json")
	_ = json.NewEncoder(w).Encode(status)
}

func (s *Server) handleRuntime(w http.ResponseWriter, _ *http.Request) {
	s.mu.RLock()
	drv := s.drv
	tx := s.tx
	stream := s.streamSrc
	ingestRt := s.ingestRt
	s.mu.RUnlock()

	result := map[string]any{}
	if drv != nil {
		result["driver"] = drv.Stats()
	}
	if tx != nil {
		if stats := tx.TXStats(); stats != nil {
			result["engine"] = stats
		}
	}
	if stream != nil {
		result["audioStream"] = stream.Stats()
	}
	if ingestRt != nil {
		result["ingest"] = ingestRt.Stats()
	}
	result["controlAudit"] = s.auditSnapshot()
	w.Header().Set("Content-Type", "application/json")
	_ = json.NewEncoder(w).Encode(result)
}

func (s *Server) handleRuntimeFaultReset(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		s.recordAudit(auditMethodNotAllowed)
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
		return
	}
	if s.rejectBody(w, r) { // BUG-01 fix: rejectBody returns true when rejected
		return
	}
	s.mu.RLock()
	tx := s.tx
	s.mu.RUnlock()
	if tx == nil {
		http.Error(w, "tx controller not available", http.StatusServiceUnavailable)
		return
	}
	if err := tx.ResetFault(); err != nil {
		http.Error(w, err.Error(), http.StatusConflict)
		return
	}
	w.Header().Set("Content-Type", "application/json")
	_ = json.NewEncoder(w).Encode(map[string]any{"ok": true})
}

// handleAudioStream accepts raw S16LE PCM via HTTP POST and pushes
// it into the configured ingest http-raw source. Use with:
//
//	curl -X POST --data-binary @- http://host:8088/audio/stream < audio.raw
//	ffmpeg ... -f s16le -ar 44100 -ac 2 - | curl -X POST --data-binary @- http://host:8088/audio/stream
func (s *Server) handleAudioStream(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		s.recordAudit(auditMethodNotAllowed)
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
		return
	}
	if !isAudioStreamContentType(r) {
		s.recordAudit(auditUnsupportedMediaType)
		http.Error(w, audioStreamContentTypeError, http.StatusUnsupportedMediaType)
		return
	}
	s.mu.RLock()
	ingress := s.audioIngress
	s.mu.RUnlock()

	if ingress == nil {
		http.Error(w, "audio ingest not configured (use --audio-http with ingest runtime)", http.StatusServiceUnavailable)
		return
	}

	// BUG-10 fix: /audio/stream is a long-lived streaming endpoint.
	// The global HTTP server ReadTimeout (5s) and WriteTimeout (10s) would
	// kill connections mid-stream. Disable them per-request via ResponseController
	// (requires Go 1.20+, confirmed Go 1.22).
	rc := http.NewResponseController(w)
	_ = rc.SetReadDeadline(time.Time{})
	_ = rc.SetWriteDeadline(time.Time{})

	r.Body = http.MaxBytesReader(w, r.Body, audioStreamBodyLimit)

	// Read body in chunks and push to ring buffer
	buf := make([]byte, 32768)
	totalFrames := 0
	for {
		n, err := r.Body.Read(buf)
		if n > 0 {
			written, writeErr := ingress.WritePCM16(buf[:n])
			totalFrames += written
			if writeErr != nil {
				http.Error(w, writeErr.Error(), http.StatusServiceUnavailable)
				return
			}
		}
		if err != nil {
			if err == io.EOF {
				break
			}
			var maxErr *http.MaxBytesError
			if errors.As(err, &maxErr) {
				s.recordAudit(auditBodyTooLarge)
				http.Error(w, maxErr.Error(), http.StatusRequestEntityTooLarge)
				return
			}
			http.Error(w, err.Error(), http.StatusInternalServerError)
			return
		}
	}

	w.Header().Set("Content-Type", "application/json")
	_ = json.NewEncoder(w).Encode(map[string]any{
		"ok":     true,
		"frames": totalFrames,
	})
}

func (s *Server) handleTXStart(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		s.recordAudit(auditMethodNotAllowed)
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
		return
	}
	if s.rejectBody(w, r) { // BUG-01 fix: rejectBody returns true when rejected
		return
	}
	s.mu.RLock()
	tx := s.tx
	s.mu.RUnlock()
	if tx == nil {
		http.Error(w, "tx controller not available", http.StatusServiceUnavailable)
		return
	}
	if err := tx.StartTX(); err != nil {
		http.Error(w, err.Error(), http.StatusConflict)
		return
	}
	w.Header().Set("Content-Type", "application/json")
	_ = json.NewEncoder(w).Encode(map[string]any{"ok": true, "action": "started"})
}

func (s *Server) handleTXStop(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		s.recordAudit(auditMethodNotAllowed)
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
		return
	}
	if s.rejectBody(w, r) { // BUG-01 fix: rejectBody returns true when rejected
		return
	}
	s.mu.RLock()
	tx := s.tx
	s.mu.RUnlock()
	if tx == nil {
		http.Error(w, "tx controller not available", http.StatusServiceUnavailable)
		return
	}
	if err := tx.StopTX(); err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	w.Header().Set("Content-Type", "application/json")
	_ = json.NewEncoder(w).Encode(map[string]any{"ok": true, "action": "stopped"})
}

func (s *Server) handleDryRun(w http.ResponseWriter, _ *http.Request) {
	s.mu.RLock()
	cfg := s.cfg
	s.mu.RUnlock()
	w.Header().Set("Content-Type", "application/json")
	_ = json.NewEncoder(w).Encode(drypkg.Generate(cfg))
}

func (s *Server) handleConfig(w http.ResponseWriter, r *http.Request) {
	switch r.Method {
	case http.MethodGet:
		s.mu.RLock()
		cfg := s.cfg
		s.mu.RUnlock()
		w.Header().Set("Content-Type", "application/json")
		_ = json.NewEncoder(w).Encode(cfg)
	case http.MethodPost:
		if !isJSONContentType(r) {
			s.recordAudit(auditUnsupportedMediaType)
			http.Error(w, "Content-Type must be application/json", http.StatusUnsupportedMediaType)
			return
		}
		r.Body = http.MaxBytesReader(w, r.Body, maxConfigBodyBytes)
		var patch ConfigPatch
		if err := json.NewDecoder(r.Body).Decode(&patch); err != nil {
			statusCode := http.StatusBadRequest
			if strings.Contains(err.Error(), "http: request body too large") {
				statusCode = http.StatusRequestEntityTooLarge
				s.recordAudit(auditBodyTooLarge)
			}
			http.Error(w, err.Error(), statusCode)
			return
		}

		// Update the server's config snapshot (for GET /config and /status)
		s.mu.Lock()
		next := s.cfg
		if patch.FrequencyMHz != nil {
			next.FM.FrequencyMHz = *patch.FrequencyMHz
		}
		if patch.OutputDrive != nil {
			next.FM.OutputDrive = *patch.OutputDrive
		}
		if patch.ToneLeftHz != nil {
			next.Audio.ToneLeftHz = *patch.ToneLeftHz
		}
		if patch.ToneRightHz != nil {
			next.Audio.ToneRightHz = *patch.ToneRightHz
		}
		if patch.ToneAmplitude != nil {
			next.Audio.ToneAmplitude = *patch.ToneAmplitude
		}
		if patch.PS != nil {
			next.RDS.PS = *patch.PS
		}
		if patch.RadioText != nil {
			next.RDS.RadioText = *patch.RadioText
		}
		if patch.PreEmphasisTauUS != nil {
			next.FM.PreEmphasisTauUS = *patch.PreEmphasisTauUS
		}
		if patch.StereoEnabled != nil {
			next.FM.StereoEnabled = *patch.StereoEnabled
		}
		if patch.LimiterEnabled != nil {
			next.FM.LimiterEnabled = *patch.LimiterEnabled
		}
		if patch.LimiterCeiling != nil {
			next.FM.LimiterCeiling = *patch.LimiterCeiling
		}
		if patch.RDSEnabled != nil {
			next.RDS.Enabled = *patch.RDSEnabled
		}
		if patch.PilotLevel != nil {
			next.FM.PilotLevel = *patch.PilotLevel
		}
		if patch.RDSInjection != nil {
			next.FM.RDSInjection = *patch.RDSInjection
		}
		if err := next.Validate(); err != nil {
			s.mu.Unlock()
			http.Error(w, err.Error(), http.StatusBadRequest)
			return
		}
		lp := LivePatch{
			FrequencyMHz:   patch.FrequencyMHz,
			OutputDrive:    patch.OutputDrive,
			StereoEnabled:  patch.StereoEnabled,
			PilotLevel:     patch.PilotLevel,
			RDSInjection:   patch.RDSInjection,
			RDSEnabled:     patch.RDSEnabled,
			LimiterEnabled: patch.LimiterEnabled,
			LimiterCeiling: patch.LimiterCeiling,
			PS:             patch.PS,
			RadioText:      patch.RadioText,
		}
		tx := s.tx
		if tx != nil {
			if err := tx.UpdateConfig(lp); err != nil {
				s.mu.Unlock()
				http.Error(w, err.Error(), http.StatusBadRequest)
				return
			}
		}
		s.cfg = next
		live := tx != nil
		s.mu.Unlock()
		w.Header().Set("Content-Type", "application/json")
		_ = json.NewEncoder(w).Encode(map[string]any{"ok": true, "live": live})
	default:
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
	}
}

func (s *Server) handleIngestSave(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		s.recordAudit(auditMethodNotAllowed)
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
		return
	}
	if !isJSONContentType(r) {
		s.recordAudit(auditUnsupportedMediaType)
		http.Error(w, "Content-Type must be application/json", http.StatusUnsupportedMediaType)
		return
	}
	r.Body = http.MaxBytesReader(w, r.Body, maxConfigBodyBytes)

	var req IngestSaveRequest
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		statusCode := http.StatusBadRequest
		if strings.Contains(err.Error(), "http: request body too large") {
			statusCode = http.StatusRequestEntityTooLarge
			s.recordAudit(auditBodyTooLarge)
		}
		http.Error(w, err.Error(), statusCode)
		return
	}

	s.mu.Lock()
	next := s.cfg
	next.Ingest = req.Ingest
	if err := next.Validate(); err != nil {
		s.mu.Unlock()
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}
	save := s.saveConfig
	reload := s.hardReload
	if save == nil {
		s.mu.Unlock()
		http.Error(w, "config save is not configured (start with --config <path>)", http.StatusServiceUnavailable)
		return
	}
	if err := save(next); err != nil {
		s.mu.Unlock()
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	s.cfg = next
	s.mu.Unlock()

	w.Header().Set("Content-Type", "application/json")
	reloadScheduled := reload != nil
	_ = json.NewEncoder(w).Encode(map[string]any{
		"ok":              true,
		"saved":           true,
		"reloadScheduled": reloadScheduled,
	})
	if reloadScheduled {
		go func(fn func()) {
			time.Sleep(250 * time.Millisecond)
			fn()
		}(reload)
	}
}