atc_server.go

package netpipe

import (
	"encoding/json"
	"fmt"
	"net/http"
)

// ATCServer is the HTTP server that serves the ATC dashboard and streams
// telemetry events via Server-Sent Events (SSE). No external dependencies.
//
// Endpoints:
//   GET /             - serves the ATC dashboard HTML
//   GET /events       - SSE stream of real-time events
//   GET /snapshot     - JSON snapshot of current state
type ATCServer struct {
	bus          *ATCBus
	port         int
	snapshotFunc func() ATCSnapshot // callback to get current state
}

// newATCServer creates the HTTP server. Call start() to begin serving.
func newATCServer(bus *ATCBus, port int, snapshotFunc func() ATCSnapshot) *ATCServer {
	return &ATCServer{
		bus:          bus,
		port:         port,
		snapshotFunc: snapshotFunc,
	}
}

// start begins serving the ATC dashboard. Runs in a goroutine.
func (a *ATCServer) start() {
	mux := http.NewServeMux()
	mux.HandleFunc("/", a.handleDashboard)
	mux.HandleFunc("/events", a.handleSSE)
	mux.HandleFunc("/snapshot", a.handleSnapshot)

	addr := fmt.Sprintf(":%d", a.port)
	fmt.Printf("netpipe ATC dashboard: http://localhost%s\n", addr)

	go http.ListenAndServe(addr, mux)
}

// handleDashboard serves the embedded HTML dashboard.
func (a *ATCServer) handleDashboard(w http.ResponseWriter, r *http.Request) {
	w.Header().Set("Content-Type", "text/html; charset=utf-8")
	w.Write([]byte(atcDashboardHTML))
}

// handleSSE streams events to the dashboard via Server-Sent Events.
func (a *ATCServer) handleSSE(w http.ResponseWriter, r *http.Request) {
	w.Header().Set("Content-Type", "text/event-stream")
	w.Header().Set("Cache-Control", "no-cache")
	w.Header().Set("Connection", "keep-alive")
	w.Header().Set("Access-Control-Allow-Origin", "*")

	flusher, ok := w.(http.Flusher)
	if !ok {
		http.Error(w, "SSE not supported", http.StatusInternalServerError)
		return
	}

	// send recent log as initial batch
	for _, evt := range a.bus.recentLog() {
		data := marshalEvent(evt)
		fmt.Fprintf(w, "data: %s\n\n", data)
	}
	flusher.Flush()

	// subscribe to live events
	ch := a.bus.subscribe()
	defer a.bus.unsubscribe(ch)

	ctx := r.Context()
	for {
		select {
		case <-ctx.Done():
			return
		case evt, ok := <-ch:
			if !ok {
				return
			}
			data := marshalEvent(evt)
			fmt.Fprintf(w, "data: %s\n\n", data)
			flusher.Flush()
		}
	}
}

// handleSnapshot returns a JSON snapshot of the current state.
func (a *ATCServer) handleSnapshot(w http.ResponseWriter, r *http.Request) {
	w.Header().Set("Content-Type", "application/json")
	w.Header().Set("Access-Control-Allow-Origin", "*")
	snapshot := a.snapshotFunc()
	json.NewEncoder(w).Encode(snapshot)
}