nginx-stream-server/cmd/dashboard/main.go

package main

import (
	"context"
	"fmt"
	"html/template"
	"log"
	"mime"
	"net/http"
	"net/http/httputil"
	"net/url"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"github.com/go-chi/chi/v5"
	"github.com/go-chi/httprate"
	"github.com/goccy/go-json"

	"git.send.nrw/sendnrw/nginx-stream-server/internal/mtx"
)

var (
	listen     = env("LISTEN", ":8080")
	mtxAPI     = env("MTX_API", "http://127.0.0.1:9997")
	mtxHLS     = env("MTX_HLS", "http://127.0.0.1:8888")
	streamsCSV = os.Getenv("STREAMS")
	basicUser  = os.Getenv("BASIC_AUTH_USER")
	basicPass  = os.Getenv("BASIC_AUTH_PASS")

	// Root für Webassets (per ENV überschreibbar)
	webRoot = env("WEB_ROOT", "web")
	// Templates werden beim Start geladen
	tpls *template.Template
)

// ----- SSE Broker -----
type sseClient chan []byte

type sseBroker struct {
	mu      sync.Mutex
	clients map[sseClient]struct{}
}

func newSSEBroker() *sseBroker {
	return &sseBroker{clients: make(map[sseClient]struct{})}
}
func (b *sseBroker) add(c sseClient) {
	b.mu.Lock()
	b.clients[c] = struct{}{}
	b.mu.Unlock()
}
func (b *sseBroker) del(c sseClient) {
	b.mu.Lock()
	delete(b.clients, c)
	b.mu.Unlock()
}
func (b *sseBroker) broadcast(msg []byte) {
	b.mu.Lock()
	for c := range b.clients {
		select {
		case c <- msg:
		default: /* slow client, drop */
		}
	}
	b.mu.Unlock()
}

var (
	streamsJSONCache []byte
	streamsJSONETag  string
	streamsCacheMu   sync.RWMutex
	sseB             = newSSEBroker()
)

func eq(a, b []byte) bool {
	if len(a) != len(b) {
		return false
	}
	for i := range a {
		if a[i] != b[i] {
			return false
		}
	}
	return true
}

func init() {
	_ = mime.AddExtensionType(".css", "text/css")
	_ = mime.AddExtensionType(".js", "application/javascript")
	_ = mime.AddExtensionType(".mjs", "application/javascript")
	_ = mime.AddExtensionType(".woff2", "font/woff2")
	_ = mime.AddExtensionType(".woff", "font/woff")
}

func env(k, def string) string {
	if v := os.Getenv(k); v != "" {
		return v
	}
	return def
}

func mustLoadTemplates() *template.Template {
	pattern := filepath.Join(webRoot, "templates", "*.html")
	t, err := template.ParseGlob(pattern)
	if err != nil {
		log.Fatalf("templates laden fehlgeschlagen (%s): %v", pattern, err)
	}
	return t
}

func main() {
	log.SetFlags(log.LstdFlags | log.Lshortfile)
	tpls = mustLoadTemplates()

	r := chi.NewRouter()
	// Strenge CSP (alles lokal)
	r.Use(func(next http.Handler) http.Handler {
		return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
			w.Header().Set("Content-Security-Policy",
				strings.Join([]string{
					"default-src 'self'",
					"img-src 'self' data:",
					"style-src 'self'",
					"font-src 'self'",
					"script-src 'self'",
					"connect-src 'self'",
					"media-src 'self' blob:", // <— HIER blob: zulassen
					"worker-src 'self' blob:",
				}, "; "),
			)
			w.Header().Set("X-Content-Type-Options", "nosniff")
			w.Header().Set("X-Frame-Options", "SAMEORIGIN")
			w.Header().Set("Referrer-Policy", "no-referrer")
			next.ServeHTTP(w, req)
		})
	})

	// Rate Limit nur auf REST, nicht auf SSE:
	r.Group(func(r chi.Router) {
		r.Use(httprate.LimitByIP(240, time.Minute)) // fertig
		r.Get("/api/streams", apiStreams)
	})

	// SSE ohne Rate-Limit (1 Verbindung pro Browser):
	r.Get("/api/streams/events", sseHandler)

	// Optional Basic Auth für Seiten
	if basicUser != "" {
		creds := basicUser + ":" + basicPass
		r.Group(func(p chi.Router) {
			p.Use(func(next http.Handler) http.Handler {
				return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
					a := strings.SplitN(r.Header.Get("Authorization"), " ", 2)
					if len(a) == 2 && a[0] == "Basic" && a[1] == basic(creds) {
						next.ServeHTTP(w, r)
						return
					}
					w.Header().Set("WWW-Authenticate", "Basic realm=Restricted")
					http.Error(w, "auth required", http.StatusUnauthorized)
				})
			})
			p.Get("/", pageIndex)
			p.Get("/{name}", pageStream)
		})
	} else {
		r.Get("/", pageIndex)
		r.Get("/{name}", pageStream)
	}

	// /static → echtes Dateisystem
	staticDir := http.Dir(filepath.Join(webRoot, "static"))
	r.Handle("/static/*",
		http.StripPrefix("/static",
			http.FileServer(staticDir),
		),
	)

	// HLS-Reverse-Proxy
	up, _ := url.Parse(mtxHLS)
	proxy := httputil.NewSingleHostReverseProxy(up)
	proxy.ErrorHandler = func(w http.ResponseWriter, r *http.Request, e error) {
		http.Error(w, "upstream error", http.StatusBadGateway)
	}
	r.Handle("/hls/*", http.StripPrefix("/hls", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if strings.Contains(r.URL.Path, "..") {
			http.NotFound(w, r)
			return
		}
		proxy.ServeHTTP(w, r)
	})))

	// Health
	r.Get("/health", func(w http.ResponseWriter, r *http.Request) { w.Write([]byte("ok")) })

	go func() {
		t := time.NewTicker(1 * time.Second)
		defer t.Stop()
		for range t.C {
			// gleiche Logik wie in apiStreams, nur ohne Response
			ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
			c := &mtx.Client{BaseURL: mtxAPI, User: os.Getenv("MTX_API_USER"), Pass: os.Getenv("MTX_API_PASS")}
			pl, err := c.Paths(ctx)
			cancel()
			if err != nil {
				// optional: bei Fehler nichts senden
				continue
			}

			allowed := map[string]bool{}
			if streamsCSV != "" {
				for _, s := range strings.Split(streamsCSV, ",") {
					allowed[strings.TrimSpace(s)] = true
				}
			}
			type item struct {
				Name    string `json:"name"`
				Live    bool   `json:"live"`
				Viewers int    `json:"viewers"`
			}
			out := struct {
				Items []item `json:"items"`
			}{}
			for _, p := range pl.Items {
				if len(allowed) > 0 && !allowed[p.Name] {
					continue
				}
				out.Items = append(out.Items, item{Name: p.Name, Live: p.Live(), Viewers: p.Viewers()})
			}

			buf, _ := json.Marshal(out)

			streamsCacheMu.RLock()
			same := eq(buf, streamsJSONCache)
			streamsCacheMu.RUnlock()
			if same {
				continue
			}

			streamsCacheMu.Lock()
			streamsJSONCache = buf
			streamsJSONETag = fmt.Sprintf("%x", len(buf))
			streamsCacheMu.Unlock()

			// an alle SSE-Clients senden
			sseB.broadcast(buf)
		}
	}()

	log.Printf("Dashboard listening on %s (API=%s HLS=%s, WEB_ROOT=%s)\n", listen, mtxAPI, mtxHLS, webRoot)
	if err := http.ListenAndServe(listen, r); err != nil {
		log.Fatal(err)
	}
}

func sseHandler(w http.ResponseWriter, r *http.Request) {
	// SSE-Header
	w.Header().Set("Content-Type", "text/event-stream")
	w.Header().Set("Cache-Control", "no-cache")
	w.Header().Set("Connection", "keep-alive")
	// CORS brauchst du nicht, gleiche Origin

	// Flush unterstützen
	flusher, ok := w.(http.Flusher)
	if !ok {
		http.Error(w, "stream unsupported", http.StatusInternalServerError)
		return
	}

	// Client-Kanal registrieren
	ch := make(sseClient, 8)
	sseB.add(ch)
	defer sseB.del(ch)

	// Beim Connect sofort den aktuellen Snapshot schicken (falls vorhanden)
	streamsCacheMu.RLock()
	snap := streamsJSONCache
	streamsCacheMu.RUnlock()
	if len(snap) > 0 {
		fmt.Fprintf(w, "event: update\ndata: %s\n\n", string(snap))
		flusher.Flush()
	}

	// Heartbeat (hält Proxies freundlich)
	hb := time.NewTicker(15 * time.Second)
	defer hb.Stop()

	// Abbruch, wenn Client trennt
	notify := r.Context().Done()

	for {
		select {
		case <-notify:
			return
		case msg := <-ch:
			// JSON als "update" senden
			fmt.Fprintf(w, "event: update\ndata: %s\n\n", string(msg))
			flusher.Flush()
		case <-hb.C:
			// Kommentar als Ping
			fmt.Fprintf(w, ": ping\n\n")
			flusher.Flush()
		}
	}
}

func apiStreams(w http.ResponseWriter, r *http.Request) {
	ctx, cancel := context.WithTimeout(r.Context(), 3*time.Second)
	defer cancel()

	c := &mtx.Client{BaseURL: mtxAPI, User: os.Getenv("MTX_API_USER"), Pass: os.Getenv("MTX_API_PASS")}
	pl, err := c.Paths(ctx)
	if err != nil {
		w.Header().Set("Content-Type", "application/json")
		w.WriteHeader(http.StatusBadGateway)
		_ = json.NewEncoder(w).Encode(map[string]any{"error": err.Error()})
		return
	}

	allowed := map[string]bool{}
	if streamsCSV != "" {
		for _, s := range strings.Split(streamsCSV, ",") {
			allowed[strings.TrimSpace(s)] = true
		}
	}
	type item struct {
		Name    string `json:"name"`
		Live    bool   `json:"live"`
		Viewers int    `json:"viewers"`
	}
	out := struct {
		Items []item `json:"items"`
	}{}
	for _, p := range pl.Items {
		if len(allowed) > 0 && !allowed[p.Name] {
			continue
		}
		out.Items = append(out.Items, item{Name: p.Name, Live: p.Live(), Viewers: p.Viewers()})
	}
	w.Header().Set("Content-Type", "application/json")
	_ = json.NewEncoder(w).Encode(out)
}

func pageIndex(w http.ResponseWriter, r *http.Request) {
	w.Header().Set("Content-Type", "text/html; charset=utf-8")
	if err := tpls.ExecuteTemplate(w, "index.html", nil); err != nil {
		http.Error(w, err.Error(), 500)
	}
}

func pageStream(w http.ResponseWriter, r *http.Request) {
	name := chi.URLParam(r, "name")
	w.Header().Set("Content-Type", "text/html; charset=utf-8")
	if err := tpls.ExecuteTemplate(w, "stream.html", map[string]any{
		"Name":     name,
		"JSONName": fmt.Sprintf("%q", name),
	}); err != nil {
		http.Error(w, err.Error(), 500)
	}
}

// Basic helper bleibt, falls du Basic Auth nutzt
func basic(creds string) string {
	const tbl = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"
	b := []byte(creds)
	var out []byte
	for i := 0; i < len(b); i += 3 {
		var v uint32
		n := 0
		for j := 0; j < 3; j++ {
			v <<= 8
			if i+j < len(b) {
				v |= uint32(b[i+j])
				n++
			}
		}
		for j := 0; j < 4; j++ {
			if j <= n {
				out = append(out, tbl[(v>>(18-6*uint(j)))&0x3f])
			} else {
				out = append(out, '=')
			}
		}
	}
	return string(out)
}