mqtt-5-test/internal/web/server.go

// Package web serves an optional loopback-only, read-only event inspector.
package web

import (
	"context"
	"embed"
	"encoding/json"
	"errors"
	"fmt"
	"io/fs"
	"net"
	"net/http"
	"net/url"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"robot-market/test-mqtt-5/internal/config"
	"robot-market/test-mqtt-5/internal/events"
)

const maxReplay = 500

//go:embed assets
var assets embed.FS

type Compatibility struct {
	Feature  string `json:"feature"`
	Status   string `json:"status"`
	Evidence string `json:"evidence"`
}
type Snapshot struct {
	Events        []events.Event         `json:"events"`
	Counts        map[events.Kind]uint64 `json:"counts"`
	Gap           *events.Notice         `json:"gap,omitempty"`
	Connection    string                 `json:"connection"`
	LastEventID   uint64                 `json:"last_event_id"`
	Compatibility []Compatibility        `json:"compatibility"`
}
type Viewer struct {
	ready    func() bool
	cfg      config.Web
	store    *events.Store
	shutdown time.Duration
	handler  http.Handler
	server   *http.Server
	listener net.Listener
	done     chan error
	slots    chan struct{}
	active   atomic.Int32
	once     sync.Once
}

func New(cfg config.Web, store *events.Store, shutdown time.Duration) *Viewer {
	v := &Viewer{cfg: cfg, store: store, shutdown: shutdown, done: make(chan error, 1), slots: make(chan struct{}, cfg.MaxViewers)}
	mux := http.NewServeMux()
	mux.HandleFunc("GET /", v.index)
	sub, _ := fs.Sub(assets, "assets")
	mux.Handle("GET /assets/", http.StripPrefix("/assets/", http.FileServer(http.FS(sub))))
	mux.HandleFunc("GET /api/snapshot", v.snapshot)
	mux.HandleFunc("GET /api/events", v.stream)
	mux.HandleFunc("GET /healthz", v.health)
	v.handler = v.guard(mux)
	v.server = &http.Server{Handler: v.handler, ReadHeaderTimeout: 3 * time.Second, IdleTimeout: 20 * time.Second, MaxHeaderBytes: 8 << 10}
	return v
}

// SetReadiness must be called before Start. Readiness reflects live MQTT state,
// independent of connection events rolling out of the bounded viewer history.
func (v *Viewer) SetReadiness(ready func() bool) { v.ready = ready }

func (v *Viewer) health(w http.ResponseWriter, r *http.Request) {
	w.Header().Set("Cache-Control", "no-store")
	if v.ready == nil || !v.ready() {
		http.Error(w, "MQTT not ready", http.StatusServiceUnavailable)
		return
	}
	w.Header().Set("Content-Type", "application/json")
	_, _ = w.Write([]byte(`{"status":"ready"}`))
}
func (v *Viewer) Address() string {
	if v.listener == nil {
		return v.cfg.Address
	}
	return v.listener.Addr().String()
}
func (v *Viewer) Handler() http.Handler { return v.handler }
func (v *Viewer) Start() error {
	listener, err := net.Listen("tcp", v.cfg.Address)
	if err != nil {
		return errors.New("viewer could not bind configured loopback address")
	}
	v.listener = listener
	go func() {
		err := v.server.Serve(listener)
		if errors.Is(err, http.ErrServerClosed) {
			err = nil
		}
		v.done <- err
	}()
	return nil
}
func (v *Viewer) Close(ctx context.Context) error {
	var result error
	v.once.Do(func() {
		if v.listener == nil {
			return
		}
		result = v.server.Shutdown(ctx)
		if result != nil {
			_ = v.server.Close()
		}
		select {
		case err := <-v.done:
			if result == nil {
				result = err
			}
		case <-ctx.Done():
			if result == nil {
				result = errors.New("viewer shutdown deadline exceeded")
			}
		}
	})
	return result
}
func (v *Viewer) guard(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Security-Policy", "default-src 'none'; script-src 'self'; style-src 'self'; connect-src 'self'; img-src 'self' data:; base-uri 'none'; form-action 'none'; frame-ancestors 'none'")
		w.Header().Set("X-Content-Type-Options", "nosniff")
		w.Header().Set("X-Frame-Options", "DENY")
		w.Header().Set("Referrer-Policy", "no-referrer")
		w.Header().Set("Permissions-Policy", "camera=(), microphone=(), geolocation=()")
		if !allowedHost(r.Host) {
			http.Error(w, "local viewer only", http.StatusForbidden)
			return
		}
		if origin := r.Header.Get("Origin"); origin != "" {
			parsed, err := url.Parse(origin)
			if err != nil || parsed.User != nil || parsed.Path != "" || parsed.RawQuery != "" || parsed.Fragment != "" || parsed.Scheme != "http" || !strings.EqualFold(parsed.Host, r.Host) {
				http.Error(w, "same-origin viewer only", http.StatusForbidden)
				return
			}
		}
		next.ServeHTTP(w, r)
	})
}
func allowedHost(hostport string) bool {
	host, _, err := net.SplitHostPort(hostport)
	if err != nil {
		return false
	}
	return strings.EqualFold(host, "127.0.0.1") || strings.EqualFold(host, "localhost")
}
func (v *Viewer) index(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-store")
	_, _ = w.Write(indexHTML)
}

var indexHTML = []byte(`<!doctype html>
<html lang="en"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1"><title>MQTT test viewer</title><link rel="stylesheet" href="/assets/style.css"></head>
<body><header><div><p class="eyebrow">ROBOT MARKET · TEST LAB</p><h1>MQTT message viewer</h1><p class="sub">Live, read-only view of the Go observer event stream.</p></div><span id="connection" class="badge">starting</span></header>
<main><section class="stats" id="stats" aria-live="polite"></section><section class="compat"><h2>Compatibility evidence</h2><p class="hint">A message arriving does not establish the device’s MQTT version or native MQTT 5 feature support.</p><div class="table-scroll"><table><thead><tr><th>Feature</th><th>Status</th><th>Evidence</th></tr></thead><tbody id="compatibility"></tbody></table></div></section>
<section class="events"><div class="section-heading"><div><h2>Event timeline</h2><p class="hint">Incoming message previews. Pause affects the display, not MQTT processing.</p></div><div class="actions"><span id="stream" class="stream">connecting</span><button id="pause">Pause view</button><button id="clear">Clear view</button></div></div>
<form id="filters" class="filters"><label>Direction<input name="direction" placeholder="in / out"></label><label>Topic contains<input name="topic"></label><label>Device ID<input name="device"></label><label>Request ID<input name="request"></label><button type="reset">Clear filters</button></form>
<p id="notice" class="notice" aria-live="polite"></p><div id="timeline" class="timeline" aria-live="polite"></div></section></main><script src="/assets/app.js" defer></script></body></html>`)

func (v *Viewer) snapshot(w http.ResponseWriter, r *http.Request) {
	w.Header().Set("Content-Type", "application/json; charset=utf-8")
	w.Header().Set("Cache-Control", "no-store")
	eventsList, counts := v.store.Snapshot()
	var gap *events.Notice
	if len(eventsList) > maxReplay {
		reset := eventsList[len(eventsList)-maxReplay].ID - 1
		gap = &events.Notice{Gap: uint64(len(eventsList) - maxReplay), ResetID: reset}
		eventsList = eventsList[len(eventsList)-maxReplay:]
	} else if len(eventsList) > 0 && eventsList[0].ID > 1 {
		gap = &events.Notice{Gap: eventsList[0].ID - 1, ResetID: eventsList[0].ID - 1}
	}
	snap := makeSnapshot(eventsList, counts)
	snap.Gap = gap
	if err := json.NewEncoder(w).Encode(snap); err != nil {
		return
	}
}
func makeSnapshot(list []events.Event, counts map[events.Kind]uint64) Snapshot {
	connection := "starting"
	var last uint64
	for _, e := range list {
		if e.ID > last {
			last = e.ID
		}
		switch e.Kind {
		case events.ConnectionUp:
			connection = "connected"
		case events.ConnectionDown:
			connection = "disconnected"
		}
	}
	compatible := []Compatibility{
		{Feature: "Go backend MQTT 5 session", Status: "not_tested", Evidence: "No connection event recorded in the retained viewer history."},
		{Feature: "Device JSON telemetry/status contract", Status: "not_tested", Evidence: "No device payload has been validated in this observer session."},
		{Feature: "ESP32 native response/correlation round trip", Status: "not_tested", Evidence: "The viewer is read-only and cannot issue a correlated command. Payload arrival alone is not proof."},
	}
	for _, e := range list {
		if e.Kind == "scenario_result" && e.Scenario != nil {
			r := e.Scenario
			compatible = append(compatible, Compatibility{Feature: r.Feature, Status: r.Status, Evidence: "Reported " + r.Environment + " · " + r.Endpoint + " · " + r.Version + " · " + r.Time.Format(time.RFC3339) + " · " + r.Observation})
		}
		if e.Kind == events.ConnectionUp && strings.Contains(e.Outcome, "protocol=5") {
			compatible[0].Status = "passed"
			compatible[0].Evidence = "Go MQTT 5 observer connection established (client " + safeEvidence(e.ClientID) + "). This says nothing about an ESP32 session."
		}
		if e.Kind == events.MessageReceived && !strings.Contains(e.Outcome, "contract validation disabled") && strings.Contains(e.Topic, "/telemetry") && e.Payload.Text != "" {
			compatible[1].Status = "passed"
			compatible[1].Evidence = "A telemetry JSON payload matched the test contract for device " + safeEvidence(e.DeviceID) + ". This does not identify the device implementation or prove ESP32 compatibility."
		}
	}
	return Snapshot{Events: list, Counts: counts, Connection: connection, LastEventID: last, Compatibility: compatible}
}
func safeEvidence(s string) string {
	if len(s) > 64 {
		return s[:64]
	}
	return s
}
func (v *Viewer) stream(w http.ResponseWriter, r *http.Request) {
	after, err := cursor(r)
	if err != nil {
		http.Error(w, "invalid event cursor", http.StatusBadRequest)
		return
	}
	select {
	case v.slots <- struct{}{}:
		defer func() { <-v.slots }()
	case <-r.Context().Done():
		return
	default:
		http.Error(w, "viewer connection limit reached", http.StatusServiceUnavailable)
		return
	}
	flusher, ok := w.(http.Flusher)
	if !ok {
		http.Error(w, "stream unsupported", http.StatusInternalServerError)
		return
	}
	w.Header().Set("Content-Type", "text/event-stream; charset=utf-8")
	w.Header().Set("Cache-Control", "no-cache, no-store")
	w.Header().Set("X-Accel-Buffering", "no")
	w.WriteHeader(http.StatusOK)
	flusher.Flush()
	v.active.Add(1)
	defer v.active.Add(-1)
	replay, gap, stream, unsubscribe := v.store.Subscribe(after, v.cfg.ViewerQueue, maxReplay)
	defer unsubscribe()
	if gap != nil {
		if !writeSSE(w, flusher, "gap", gap.ResetID, *gap) {
			return
		}
	}
	for _, event := range replay {
		if !writeSSE(w, flusher, "mqtt-event", event.ID, events.Notice{Event: &event}) {
			return
		}
	}
	ticker := time.NewTicker(15 * time.Second)
	defer ticker.Stop()
	for {
		select {
		case <-r.Context().Done():
			return
		case notice, open := <-stream:
			if !open {
				return
			}
			if notice.Gap > 0 {
				if !writeSSE(w, flusher, "gap", notice.ResetID, notice) {
					return
				}
			}
			if notice.Event != nil {
				if !writeSSE(w, flusher, "mqtt-event", notice.Event.ID, notice) {
					return
				}
			}
		case <-ticker.C:
			controller := http.NewResponseController(w)
			_ = controller.SetWriteDeadline(time.Now().Add(3 * time.Second))
			if _, err := fmt.Fprint(w, ": keepalive\n\n"); err != nil {
				return
			}
			flusher.Flush()
		}
	}
}
func cursor(r *http.Request) (uint64, error) {
	value := r.Header.Get("Last-Event-ID")
	if value == "" {
		value = r.URL.Query().Get("after")
	}
	if value == "" {
		return 0, nil
	}
	n, err := strconv.ParseUint(value, 10, 64)
	if err != nil {
		return 0, errors.New("bad event cursor")
	}
	return n, nil
}
func writeSSE(w http.ResponseWriter, f http.Flusher, name string, id uint64, value any) bool {
	data, err := json.Marshal(value)
	if err != nil {
		return false
	}
	controller := http.NewResponseController(w)
	if err = controller.SetWriteDeadline(time.Now().Add(3 * time.Second)); err != nil && !errors.Is(err, http.ErrNotSupported) {
		return false
	}
	if id > 0 {
		if _, err = fmt.Fprintf(w, "id: %d\n", id); err != nil {
			return false
		}
	}
	if _, err = fmt.Fprintf(w, "event: %s\ndata: %s\n\n", name, data); err != nil {
		return false
	}
	f.Flush()
	return true
}
func (v *Viewer) ActiveViewers() int { return int(v.active.Load()) }