Go-based FM stereo transmitter with RDS, Windows-first and cross-platform
選択できるのは25トピックまでです。 トピックは、先頭が英数字で、英数字とダッシュ('-')を使用した35文字以内のものにしてください。

815 行
24KB

  1. package control
  2. import (
  3. _ "embed"
  4. "encoding/json"
  5. "errors"
  6. "io"
  7. "mime"
  8. "net/http"
  9. "strings"
  10. "sync"
  11. "sync/atomic"
  12. "time"
  13. "github.com/jan/fm-rds-tx/internal/audio"
  14. "github.com/jan/fm-rds-tx/internal/config"
  15. drypkg "github.com/jan/fm-rds-tx/internal/dryrun"
  16. "github.com/jan/fm-rds-tx/internal/ingest"
  17. "github.com/jan/fm-rds-tx/internal/platform"
  18. )
  19. //go:embed ui.html
  20. var uiHTML []byte
  21. //go:embed logo.png
  22. var logoPNG []byte
  23. // TXController is an optional interface the Server uses to start/stop TX
  24. // and apply live config changes.
  25. type TXController interface {
  26. StartTX() error
  27. StopTX() error
  28. TXStats() map[string]any
  29. UpdateConfig(patch LivePatch) error
  30. ResetFault() error
  31. }
  32. // LivePatch mirrors the patchable fields from ConfigPatch for the engine.
  33. // nil = no change.
  34. type LivePatch struct {
  35. FrequencyMHz *float64
  36. OutputDrive *float64
  37. StereoEnabled *bool
  38. StereoMode *string
  39. PilotLevel *float64
  40. RDSInjection *float64
  41. RDSEnabled *bool
  42. LimiterEnabled *bool
  43. LimiterCeiling *float64
  44. PS *string
  45. RadioText *string
  46. TA *bool
  47. TP *bool
  48. ToneLeftHz *float64
  49. ToneRightHz *float64
  50. ToneAmplitude *float64
  51. AudioGain *float64
  52. CompositeClipperEnabled *bool
  53. }
  54. type Server struct {
  55. mu sync.RWMutex
  56. cfg config.Config
  57. tx TXController
  58. drv platform.SoapyDriver // optional, for runtime stats
  59. streamSrc *audio.StreamSource // optional, for live audio ring stats
  60. audioIngress AudioIngress // optional, for /audio/stream
  61. ingestRt IngestRuntime // optional, for /runtime ingest stats
  62. saveConfig func(config.Config) error
  63. hardReload func()
  64. // BUG-F fix: reloadPending prevents multiple concurrent goroutines from
  65. // calling hardReload when handleIngestSave is hit multiple times quickly.
  66. reloadPending atomic.Bool
  67. audit auditCounters
  68. }
  69. type AudioIngress interface {
  70. WritePCM16(data []byte) (int, error)
  71. }
  72. type IngestRuntime interface {
  73. Stats() ingest.Stats
  74. }
  75. type auditEvent string
  76. const (
  77. auditMethodNotAllowed auditEvent = "methodNotAllowed"
  78. auditUnsupportedMediaType auditEvent = "unsupportedMediaType"
  79. auditBodyTooLarge auditEvent = "bodyTooLarge"
  80. auditUnexpectedBody auditEvent = "unexpectedBody"
  81. )
  82. type auditCounters struct {
  83. methodNotAllowed uint64
  84. unsupportedMediaType uint64
  85. bodyTooLarge uint64
  86. unexpectedBody uint64
  87. }
  88. const (
  89. maxConfigBodyBytes = 64 << 10 // 64 KiB
  90. configContentTypeHeader = "application/json"
  91. noBodyErrMsg = "request must not include a body"
  92. audioStreamContentTypeError = "Content-Type must be application/octet-stream or audio/L16"
  93. audioStreamBodyLimitDefault = 512 << 20 // 512 MiB
  94. )
  95. var audioStreamAllowedMediaTypes = []string{
  96. "application/octet-stream",
  97. "audio/l16",
  98. }
  99. var audioStreamBodyLimit = int64(audioStreamBodyLimitDefault) // bytes allowed per /audio/stream request; tests may override.
  100. func isJSONContentType(r *http.Request) bool {
  101. ct := strings.TrimSpace(r.Header.Get("Content-Type"))
  102. if ct == "" {
  103. return false
  104. }
  105. mediaType, _, err := mime.ParseMediaType(ct)
  106. if err != nil {
  107. return false
  108. }
  109. return strings.EqualFold(mediaType, configContentTypeHeader)
  110. }
  111. type ConfigPatch struct {
  112. FrequencyMHz *float64 `json:"frequencyMHz,omitempty"`
  113. OutputDrive *float64 `json:"outputDrive,omitempty"`
  114. StereoEnabled *bool `json:"stereoEnabled,omitempty"`
  115. StereoMode *string `json:"stereoMode,omitempty"`
  116. PilotLevel *float64 `json:"pilotLevel,omitempty"`
  117. RDSInjection *float64 `json:"rdsInjection,omitempty"`
  118. RDSEnabled *bool `json:"rdsEnabled,omitempty"`
  119. ToneLeftHz *float64 `json:"toneLeftHz,omitempty"`
  120. ToneRightHz *float64 `json:"toneRightHz,omitempty"`
  121. ToneAmplitude *float64 `json:"toneAmplitude,omitempty"`
  122. PS *string `json:"ps,omitempty"`
  123. RadioText *string `json:"radioText,omitempty"`
  124. PreEmphasisTauUS *float64 `json:"preEmphasisTauUS,omitempty"`
  125. LimiterEnabled *bool `json:"limiterEnabled,omitempty"`
  126. LimiterCeiling *float64 `json:"limiterCeiling,omitempty"`
  127. AudioGain *float64 `json:"audioGain,omitempty"`
  128. PI *string `json:"pi,omitempty"`
  129. PTY *int `json:"pty,omitempty"`
  130. TP *bool `json:"tp,omitempty"`
  131. TA *bool `json:"ta,omitempty"`
  132. MS *bool `json:"ms,omitempty"`
  133. CTEnabled *bool `json:"ctEnabled,omitempty"`
  134. RTPlusEnabled *bool `json:"rtPlusEnabled,omitempty"`
  135. RTPlusSeparator *string `json:"rtPlusSeparator,omitempty"`
  136. PTYN *string `json:"ptyn,omitempty"`
  137. LPS *string `json:"lps,omitempty"`
  138. ERTEnabled *bool `json:"ertEnabled,omitempty"`
  139. ERT *string `json:"ert,omitempty"`
  140. RDS2Enabled *bool `json:"rds2Enabled,omitempty"`
  141. StationLogoPath *string `json:"stationLogoPath,omitempty"`
  142. AF *[]float64 `json:"af,omitempty"`
  143. BS412Enabled *bool `json:"bs412Enabled,omitempty"`
  144. BS412ThresholdDBr *float64 `json:"bs412ThresholdDBr,omitempty"`
  145. MpxGain *float64 `json:"mpxGain,omitempty"`
  146. CompositeClipperEnabled *bool `json:"compositeClipperEnabled,omitempty"`
  147. CompositeClipperIterations *int `json:"compositeClipperIterations,omitempty"`
  148. CompositeClipperSoftKnee *float64 `json:"compositeClipperSoftKnee,omitempty"`
  149. CompositeClipperLookaheadMs *float64 `json:"compositeClipperLookaheadMs,omitempty"`
  150. }
  151. type IngestSaveRequest struct {
  152. Ingest config.IngestConfig `json:"ingest"`
  153. }
  154. func NewServer(cfg config.Config) *Server {
  155. return &Server{cfg: cfg}
  156. }
  157. func hasRequestBody(r *http.Request) bool {
  158. if r.ContentLength > 0 {
  159. return true
  160. }
  161. for _, te := range r.TransferEncoding {
  162. if strings.EqualFold(te, "chunked") {
  163. return true
  164. }
  165. }
  166. return false
  167. }
  168. func (s *Server) rejectBody(w http.ResponseWriter, r *http.Request) bool {
  169. // Returns true when the request has an unexpected body and the error response
  170. // has already been written — callers should return immediately in that case.
  171. // Returns false when there is no body (happy path — request should proceed).
  172. if !hasRequestBody(r) {
  173. return false
  174. }
  175. s.recordAudit(auditUnexpectedBody)
  176. http.Error(w, noBodyErrMsg, http.StatusBadRequest)
  177. return true
  178. }
  179. func (s *Server) recordAudit(evt auditEvent) {
  180. switch evt {
  181. case auditMethodNotAllowed:
  182. atomic.AddUint64(&s.audit.methodNotAllowed, 1)
  183. case auditUnsupportedMediaType:
  184. atomic.AddUint64(&s.audit.unsupportedMediaType, 1)
  185. case auditBodyTooLarge:
  186. atomic.AddUint64(&s.audit.bodyTooLarge, 1)
  187. case auditUnexpectedBody:
  188. atomic.AddUint64(&s.audit.unexpectedBody, 1)
  189. }
  190. }
  191. func (s *Server) auditSnapshot() map[string]uint64 {
  192. return map[string]uint64{
  193. "methodNotAllowed": atomic.LoadUint64(&s.audit.methodNotAllowed),
  194. "unsupportedMediaType": atomic.LoadUint64(&s.audit.unsupportedMediaType),
  195. "bodyTooLarge": atomic.LoadUint64(&s.audit.bodyTooLarge),
  196. "unexpectedBody": atomic.LoadUint64(&s.audit.unexpectedBody),
  197. }
  198. }
  199. func isAudioStreamContentType(r *http.Request) bool {
  200. ct := strings.TrimSpace(r.Header.Get("Content-Type"))
  201. if ct == "" {
  202. return false
  203. }
  204. mediaType, _, err := mime.ParseMediaType(ct)
  205. if err != nil {
  206. return false
  207. }
  208. for _, allowed := range audioStreamAllowedMediaTypes {
  209. if strings.EqualFold(mediaType, allowed) {
  210. return true
  211. }
  212. }
  213. return false
  214. }
  215. func (s *Server) SetTXController(tx TXController) {
  216. s.mu.Lock()
  217. s.tx = tx
  218. s.mu.Unlock()
  219. }
  220. func (s *Server) SetDriver(drv platform.SoapyDriver) {
  221. s.mu.Lock()
  222. s.drv = drv
  223. s.mu.Unlock()
  224. }
  225. func (s *Server) SetStreamSource(src *audio.StreamSource) {
  226. s.mu.Lock()
  227. s.streamSrc = src
  228. s.mu.Unlock()
  229. }
  230. func (s *Server) SetAudioIngress(ingress AudioIngress) {
  231. s.mu.Lock()
  232. s.audioIngress = ingress
  233. s.mu.Unlock()
  234. }
  235. func (s *Server) SetIngestRuntime(rt IngestRuntime) {
  236. s.mu.Lock()
  237. s.ingestRt = rt
  238. s.mu.Unlock()
  239. }
  240. func (s *Server) SetConfigSaver(save func(config.Config) error) {
  241. s.mu.Lock()
  242. s.saveConfig = save
  243. s.mu.Unlock()
  244. }
  245. func (s *Server) SetHardReload(fn func()) {
  246. s.mu.Lock()
  247. s.hardReload = fn
  248. s.mu.Unlock()
  249. }
  250. func (s *Server) Handler() http.Handler {
  251. mux := http.NewServeMux()
  252. mux.HandleFunc("/", s.handleUI)
  253. mux.HandleFunc("/logo", s.handleLogo)
  254. mux.HandleFunc("/healthz", s.handleHealth)
  255. mux.HandleFunc("/status", s.handleStatus)
  256. mux.HandleFunc("/dry-run", s.handleDryRun)
  257. mux.HandleFunc("/config", s.handleConfig)
  258. mux.HandleFunc("/config/ingest/save", s.handleIngestSave)
  259. mux.HandleFunc("/runtime", s.handleRuntime)
  260. mux.HandleFunc("/runtime/fault/reset", s.handleRuntimeFaultReset)
  261. mux.HandleFunc("/tx/start", s.handleTXStart)
  262. mux.HandleFunc("/tx/stop", s.handleTXStop)
  263. mux.HandleFunc("/audio/stream", s.handleAudioStream)
  264. return mux
  265. }
  266. func (s *Server) handleHealth(w http.ResponseWriter, _ *http.Request) {
  267. w.Header().Set("Content-Type", "application/json")
  268. _ = json.NewEncoder(w).Encode(map[string]any{"ok": true})
  269. }
  270. func (s *Server) handleUI(w http.ResponseWriter, r *http.Request) {
  271. if r.URL.Path != "/" {
  272. http.NotFound(w, r)
  273. return
  274. }
  275. w.Header().Set("Content-Type", "text/html; charset=utf-8")
  276. w.Header().Set("Cache-Control", "no-cache")
  277. w.Write(uiHTML)
  278. }
  279. func (s *Server) handleLogo(w http.ResponseWriter, r *http.Request) {
  280. w.Header().Set("Content-Type", "image/png")
  281. w.Header().Set("Cache-Control", "public, max-age=3600")
  282. w.Write(logoPNG)
  283. }
  284. func (s *Server) handleStatus(w http.ResponseWriter, _ *http.Request) {
  285. s.mu.RLock()
  286. cfg := s.cfg
  287. tx := s.tx
  288. s.mu.RUnlock()
  289. status := map[string]any{
  290. "service": "fm-rds-tx",
  291. "backend": cfg.Backend.Kind,
  292. "frequencyMHz": cfg.FM.FrequencyMHz,
  293. "stereoEnabled": cfg.FM.StereoEnabled,
  294. "stereoMode": cfg.FM.StereoMode,
  295. "rdsEnabled": cfg.RDS.Enabled,
  296. "preEmphasisTauUS": cfg.FM.PreEmphasisTauUS,
  297. "limiterEnabled": cfg.FM.LimiterEnabled,
  298. "fmModulationEnabled": cfg.FM.FMModulationEnabled,
  299. }
  300. if tx != nil {
  301. if stats := tx.TXStats(); stats != nil {
  302. if ri, ok := stats["runtimeIndicator"]; ok {
  303. status["runtimeIndicator"] = ri
  304. }
  305. if alert, ok := stats["runtimeAlert"]; ok {
  306. status["runtimeAlert"] = alert
  307. }
  308. if queue, ok := stats["queue"]; ok {
  309. status["queue"] = queue
  310. }
  311. if runtimeState, ok := stats["state"]; ok {
  312. status["runtimeState"] = runtimeState
  313. }
  314. }
  315. }
  316. w.Header().Set("Content-Type", "application/json")
  317. _ = json.NewEncoder(w).Encode(status)
  318. }
  319. func (s *Server) handleRuntime(w http.ResponseWriter, _ *http.Request) {
  320. s.mu.RLock()
  321. drv := s.drv
  322. tx := s.tx
  323. stream := s.streamSrc
  324. ingestRt := s.ingestRt
  325. s.mu.RUnlock()
  326. result := map[string]any{}
  327. if drv != nil {
  328. result["driver"] = drv.Stats()
  329. }
  330. if tx != nil {
  331. if stats := tx.TXStats(); stats != nil {
  332. result["engine"] = stats
  333. }
  334. }
  335. if stream != nil {
  336. result["audioStream"] = stream.Stats()
  337. }
  338. if ingestRt != nil {
  339. result["ingest"] = ingestRt.Stats()
  340. }
  341. result["controlAudit"] = s.auditSnapshot()
  342. w.Header().Set("Content-Type", "application/json")
  343. _ = json.NewEncoder(w).Encode(result)
  344. }
  345. func (s *Server) handleRuntimeFaultReset(w http.ResponseWriter, r *http.Request) {
  346. if r.Method != http.MethodPost {
  347. s.recordAudit(auditMethodNotAllowed)
  348. http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
  349. return
  350. }
  351. if s.rejectBody(w, r) { // BUG-01 fix: rejectBody returns true when rejected
  352. return
  353. }
  354. s.mu.RLock()
  355. tx := s.tx
  356. s.mu.RUnlock()
  357. if tx == nil {
  358. http.Error(w, "tx controller not available", http.StatusServiceUnavailable)
  359. return
  360. }
  361. if err := tx.ResetFault(); err != nil {
  362. http.Error(w, err.Error(), http.StatusConflict)
  363. return
  364. }
  365. w.Header().Set("Content-Type", "application/json")
  366. _ = json.NewEncoder(w).Encode(map[string]any{"ok": true})
  367. }
  368. // handleAudioStream accepts raw S16LE PCM via HTTP POST and pushes
  369. // it into the configured ingest http-raw source. Use with:
  370. //
  371. // curl -X POST --data-binary @- http://host:8088/audio/stream < audio.raw
  372. // ffmpeg ... -f s16le -ar 44100 -ac 2 - | curl -X POST --data-binary @- http://host:8088/audio/stream
  373. func (s *Server) handleAudioStream(w http.ResponseWriter, r *http.Request) {
  374. if r.Method != http.MethodPost {
  375. s.recordAudit(auditMethodNotAllowed)
  376. http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
  377. return
  378. }
  379. if !isAudioStreamContentType(r) {
  380. s.recordAudit(auditUnsupportedMediaType)
  381. http.Error(w, audioStreamContentTypeError, http.StatusUnsupportedMediaType)
  382. return
  383. }
  384. s.mu.RLock()
  385. ingress := s.audioIngress
  386. s.mu.RUnlock()
  387. if ingress == nil {
  388. http.Error(w, "audio ingest not configured (use --audio-http with ingest runtime)", http.StatusServiceUnavailable)
  389. return
  390. }
  391. // BUG-10 fix: /audio/stream is a long-lived streaming endpoint.
  392. // The global HTTP server ReadTimeout (5s) and WriteTimeout (10s) would
  393. // kill connections mid-stream. Disable them per-request via ResponseController
  394. // (requires Go 1.20+, confirmed Go 1.22).
  395. rc := http.NewResponseController(w)
  396. _ = rc.SetReadDeadline(time.Time{})
  397. _ = rc.SetWriteDeadline(time.Time{})
  398. r.Body = http.MaxBytesReader(w, r.Body, audioStreamBodyLimit)
  399. // Read body in chunks and push to ring buffer
  400. buf := make([]byte, 32768)
  401. totalFrames := 0
  402. for {
  403. n, err := r.Body.Read(buf)
  404. if n > 0 {
  405. written, writeErr := ingress.WritePCM16(buf[:n])
  406. totalFrames += written
  407. if writeErr != nil {
  408. http.Error(w, writeErr.Error(), http.StatusServiceUnavailable)
  409. return
  410. }
  411. }
  412. if err != nil {
  413. if err == io.EOF {
  414. break
  415. }
  416. var maxErr *http.MaxBytesError
  417. if errors.As(err, &maxErr) {
  418. s.recordAudit(auditBodyTooLarge)
  419. http.Error(w, maxErr.Error(), http.StatusRequestEntityTooLarge)
  420. return
  421. }
  422. http.Error(w, err.Error(), http.StatusInternalServerError)
  423. return
  424. }
  425. }
  426. w.Header().Set("Content-Type", "application/json")
  427. _ = json.NewEncoder(w).Encode(map[string]any{
  428. "ok": true,
  429. "frames": totalFrames,
  430. })
  431. }
  432. func (s *Server) handleTXStart(w http.ResponseWriter, r *http.Request) {
  433. if r.Method != http.MethodPost {
  434. s.recordAudit(auditMethodNotAllowed)
  435. http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
  436. return
  437. }
  438. if s.rejectBody(w, r) { // BUG-01 fix: rejectBody returns true when rejected
  439. return
  440. }
  441. s.mu.RLock()
  442. tx := s.tx
  443. s.mu.RUnlock()
  444. if tx == nil {
  445. http.Error(w, "tx controller not available", http.StatusServiceUnavailable)
  446. return
  447. }
  448. if err := tx.StartTX(); err != nil {
  449. http.Error(w, err.Error(), http.StatusConflict)
  450. return
  451. }
  452. w.Header().Set("Content-Type", "application/json")
  453. _ = json.NewEncoder(w).Encode(map[string]any{"ok": true, "action": "started"})
  454. }
  455. func (s *Server) handleTXStop(w http.ResponseWriter, r *http.Request) {
  456. if r.Method != http.MethodPost {
  457. s.recordAudit(auditMethodNotAllowed)
  458. http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
  459. return
  460. }
  461. if s.rejectBody(w, r) { // BUG-01 fix: rejectBody returns true when rejected
  462. return
  463. }
  464. s.mu.RLock()
  465. tx := s.tx
  466. s.mu.RUnlock()
  467. if tx == nil {
  468. http.Error(w, "tx controller not available", http.StatusServiceUnavailable)
  469. return
  470. }
  471. go func() {
  472. _ = tx.StopTX()
  473. }()
  474. w.Header().Set("Content-Type", "application/json")
  475. _ = json.NewEncoder(w).Encode(map[string]any{"ok": true, "action": "stop-requested"})
  476. }
  477. func (s *Server) handleDryRun(w http.ResponseWriter, _ *http.Request) {
  478. s.mu.RLock()
  479. cfg := s.cfg
  480. s.mu.RUnlock()
  481. w.Header().Set("Content-Type", "application/json")
  482. _ = json.NewEncoder(w).Encode(drypkg.Generate(cfg))
  483. }
  484. func (s *Server) handleConfig(w http.ResponseWriter, r *http.Request) {
  485. switch r.Method {
  486. case http.MethodGet:
  487. s.mu.RLock()
  488. cfg := s.cfg
  489. s.mu.RUnlock()
  490. w.Header().Set("Content-Type", "application/json")
  491. _ = json.NewEncoder(w).Encode(cfg)
  492. case http.MethodPost:
  493. if !isJSONContentType(r) {
  494. s.recordAudit(auditUnsupportedMediaType)
  495. http.Error(w, "Content-Type must be application/json", http.StatusUnsupportedMediaType)
  496. return
  497. }
  498. r.Body = http.MaxBytesReader(w, r.Body, maxConfigBodyBytes)
  499. var patch ConfigPatch
  500. // BUG-4 fix: reject unknown JSON fields (typos) with 400 rather than
  501. // silently ignoring them (e.g. "outputDrvie" would succeed and do nothing).
  502. dec := json.NewDecoder(r.Body)
  503. dec.DisallowUnknownFields()
  504. if err := dec.Decode(&patch); err != nil {
  505. statusCode := http.StatusBadRequest
  506. if strings.Contains(err.Error(), "http: request body too large") {
  507. statusCode = http.StatusRequestEntityTooLarge
  508. s.recordAudit(auditBodyTooLarge)
  509. }
  510. http.Error(w, err.Error(), statusCode)
  511. return
  512. }
  513. // Update the server's config snapshot (for GET /config and /status)
  514. s.mu.Lock()
  515. next := s.cfg
  516. if patch.FrequencyMHz != nil {
  517. next.FM.FrequencyMHz = *patch.FrequencyMHz
  518. }
  519. if patch.OutputDrive != nil {
  520. next.FM.OutputDrive = *patch.OutputDrive
  521. }
  522. if patch.ToneLeftHz != nil {
  523. next.Audio.ToneLeftHz = *patch.ToneLeftHz
  524. }
  525. if patch.ToneRightHz != nil {
  526. next.Audio.ToneRightHz = *patch.ToneRightHz
  527. }
  528. if patch.ToneAmplitude != nil {
  529. next.Audio.ToneAmplitude = *patch.ToneAmplitude
  530. }
  531. if patch.AudioGain != nil {
  532. next.Audio.Gain = *patch.AudioGain
  533. }
  534. if patch.PS != nil {
  535. next.RDS.PS = *patch.PS
  536. }
  537. if patch.RadioText != nil {
  538. next.RDS.RadioText = *patch.RadioText
  539. }
  540. if patch.PI != nil {
  541. next.RDS.PI = *patch.PI
  542. }
  543. if patch.PTY != nil {
  544. next.RDS.PTY = *patch.PTY
  545. }
  546. if patch.TP != nil {
  547. next.RDS.TP = *patch.TP
  548. }
  549. if patch.TA != nil {
  550. next.RDS.TA = *patch.TA
  551. }
  552. if patch.MS != nil {
  553. next.RDS.MS = *patch.MS
  554. }
  555. if patch.CTEnabled != nil {
  556. next.RDS.CTEnabled = *patch.CTEnabled
  557. }
  558. if patch.RTPlusEnabled != nil {
  559. next.RDS.RTPlusEnabled = *patch.RTPlusEnabled
  560. }
  561. if patch.RTPlusSeparator != nil {
  562. next.RDS.RTPlusSeparator = *patch.RTPlusSeparator
  563. }
  564. if patch.PTYN != nil {
  565. next.RDS.PTYN = *patch.PTYN
  566. }
  567. if patch.LPS != nil {
  568. next.RDS.LPS = *patch.LPS
  569. }
  570. if patch.ERTEnabled != nil {
  571. next.RDS.ERTEnabled = *patch.ERTEnabled
  572. }
  573. if patch.ERT != nil {
  574. next.RDS.ERT = *patch.ERT
  575. }
  576. if patch.RDS2Enabled != nil {
  577. next.RDS.RDS2Enabled = *patch.RDS2Enabled
  578. }
  579. if patch.StationLogoPath != nil {
  580. next.RDS.StationLogoPath = *patch.StationLogoPath
  581. }
  582. if patch.AF != nil {
  583. next.RDS.AF = *patch.AF
  584. }
  585. if patch.PreEmphasisTauUS != nil {
  586. next.FM.PreEmphasisTauUS = *patch.PreEmphasisTauUS
  587. }
  588. if patch.StereoEnabled != nil {
  589. next.FM.StereoEnabled = *patch.StereoEnabled
  590. }
  591. if patch.StereoMode != nil {
  592. next.FM.StereoMode = *patch.StereoMode
  593. }
  594. if patch.LimiterEnabled != nil {
  595. next.FM.LimiterEnabled = *patch.LimiterEnabled
  596. }
  597. if patch.LimiterCeiling != nil {
  598. next.FM.LimiterCeiling = *patch.LimiterCeiling
  599. }
  600. if patch.RDSEnabled != nil {
  601. next.RDS.Enabled = *patch.RDSEnabled
  602. }
  603. if patch.PilotLevel != nil {
  604. next.FM.PilotLevel = *patch.PilotLevel
  605. }
  606. if patch.RDSInjection != nil {
  607. next.FM.RDSInjection = *patch.RDSInjection
  608. }
  609. if patch.BS412Enabled != nil {
  610. next.FM.BS412Enabled = *patch.BS412Enabled
  611. }
  612. if patch.BS412ThresholdDBr != nil {
  613. next.FM.BS412ThresholdDBr = *patch.BS412ThresholdDBr
  614. }
  615. if patch.MpxGain != nil {
  616. next.FM.MpxGain = *patch.MpxGain
  617. }
  618. if patch.CompositeClipperEnabled != nil {
  619. next.FM.CompositeClipper.Enabled = *patch.CompositeClipperEnabled
  620. }
  621. if patch.CompositeClipperIterations != nil {
  622. next.FM.CompositeClipper.Iterations = *patch.CompositeClipperIterations
  623. }
  624. if patch.CompositeClipperSoftKnee != nil {
  625. next.FM.CompositeClipper.SoftKnee = *patch.CompositeClipperSoftKnee
  626. }
  627. if patch.CompositeClipperLookaheadMs != nil {
  628. next.FM.CompositeClipper.LookaheadMs = *patch.CompositeClipperLookaheadMs
  629. }
  630. if err := next.Validate(); err != nil {
  631. s.mu.Unlock()
  632. http.Error(w, err.Error(), http.StatusBadRequest)
  633. return
  634. }
  635. lp := LivePatch{
  636. FrequencyMHz: patch.FrequencyMHz,
  637. OutputDrive: patch.OutputDrive,
  638. StereoEnabled: patch.StereoEnabled,
  639. StereoMode: patch.StereoMode,
  640. PilotLevel: patch.PilotLevel,
  641. RDSInjection: patch.RDSInjection,
  642. RDSEnabled: patch.RDSEnabled,
  643. LimiterEnabled: patch.LimiterEnabled,
  644. LimiterCeiling: patch.LimiterCeiling,
  645. PS: patch.PS,
  646. RadioText: patch.RadioText,
  647. TA: patch.TA,
  648. TP: patch.TP,
  649. ToneLeftHz: patch.ToneLeftHz,
  650. ToneRightHz: patch.ToneRightHz,
  651. ToneAmplitude: patch.ToneAmplitude,
  652. AudioGain: patch.AudioGain,
  653. CompositeClipperEnabled: patch.CompositeClipperEnabled,
  654. }
  655. // NEU-02 fix: determine whether any live-patchable fields are present,
  656. // then release the lock before calling UpdateConfig to avoid holding
  657. // s.mu across a potentially blocking engine call.
  658. tx := s.tx
  659. hasLiveFields := patch.FrequencyMHz != nil || patch.OutputDrive != nil ||
  660. patch.StereoEnabled != nil || patch.StereoMode != nil || patch.PilotLevel != nil ||
  661. patch.RDSInjection != nil || patch.RDSEnabled != nil ||
  662. patch.LimiterEnabled != nil || patch.LimiterCeiling != nil ||
  663. patch.PS != nil || patch.RadioText != nil || patch.TA != nil || patch.TP != nil ||
  664. patch.ToneLeftHz != nil || patch.ToneRightHz != nil ||
  665. patch.ToneAmplitude != nil || patch.AudioGain != nil ||
  666. patch.CompositeClipperEnabled != nil
  667. s.mu.Unlock()
  668. // Apply live fields to running engine outside the lock.
  669. if tx != nil && hasLiveFields {
  670. if err := tx.UpdateConfig(lp); err != nil {
  671. http.Error(w, err.Error(), http.StatusBadRequest)
  672. return
  673. }
  674. }
  675. // Persist the validated config snapshot when a config saver is available.
  676. // This ensures restart-required UI changes survive process restarts instead
  677. // of only updating the in-memory snapshot.
  678. s.mu.RLock()
  679. save := s.saveConfig
  680. s.mu.RUnlock()
  681. if save != nil {
  682. if err := save(next); err != nil {
  683. http.Error(w, err.Error(), http.StatusInternalServerError)
  684. return
  685. }
  686. }
  687. // Commit the server snapshot only after validation, optional persistence,
  688. // and any required live update succeeded.
  689. s.mu.Lock()
  690. s.cfg = next
  691. s.mu.Unlock()
  692. // NEU-03 fix: report live=true only when live-patchable fields were applied.
  693. live := tx != nil && hasLiveFields
  694. w.Header().Set("Content-Type", "application/json")
  695. _ = json.NewEncoder(w).Encode(map[string]any{"ok": true, "live": live})
  696. default:
  697. http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
  698. }
  699. }
  700. func (s *Server) handleIngestSave(w http.ResponseWriter, r *http.Request) {
  701. if r.Method != http.MethodPost {
  702. s.recordAudit(auditMethodNotAllowed)
  703. http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
  704. return
  705. }
  706. if !isJSONContentType(r) {
  707. s.recordAudit(auditUnsupportedMediaType)
  708. http.Error(w, "Content-Type must be application/json", http.StatusUnsupportedMediaType)
  709. return
  710. }
  711. r.Body = http.MaxBytesReader(w, r.Body, maxConfigBodyBytes)
  712. var req IngestSaveRequest
  713. if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
  714. statusCode := http.StatusBadRequest
  715. if strings.Contains(err.Error(), "http: request body too large") {
  716. statusCode = http.StatusRequestEntityTooLarge
  717. s.recordAudit(auditBodyTooLarge)
  718. }
  719. http.Error(w, err.Error(), statusCode)
  720. return
  721. }
  722. s.mu.Lock()
  723. next := s.cfg
  724. next.Ingest = req.Ingest
  725. if err := next.Validate(); err != nil {
  726. s.mu.Unlock()
  727. http.Error(w, err.Error(), http.StatusBadRequest)
  728. return
  729. }
  730. save := s.saveConfig
  731. reload := s.hardReload
  732. if save == nil {
  733. s.mu.Unlock()
  734. http.Error(w, "config save is not configured (start with --config <path>)", http.StatusServiceUnavailable)
  735. return
  736. }
  737. if err := save(next); err != nil {
  738. s.mu.Unlock()
  739. http.Error(w, err.Error(), http.StatusInternalServerError)
  740. return
  741. }
  742. s.cfg = next
  743. s.mu.Unlock()
  744. w.Header().Set("Content-Type", "application/json")
  745. reloadScheduled := reload != nil
  746. _ = json.NewEncoder(w).Encode(map[string]any{
  747. "ok": true,
  748. "saved": true,
  749. "reloadScheduled": reloadScheduled,
  750. })
  751. if reloadScheduled && s.reloadPending.CompareAndSwap(false, true) {
  752. go func(fn func()) {
  753. time.Sleep(250 * time.Millisecond)
  754. s.reloadPending.Store(false)
  755. fn()
  756. }(reload)
  757. }
  758. }