package server

import (
	"encoding/json"
	"fmt"
	"net/http"
	"strings"
	"time"

	"github.com/operator/command-center/internal/eventbus"
)

// handleSSE implements GET /sse/events?topic=<topic>[,<topic>...]
//
// One subscription is opened per topic listed in the query string. Events
// are emitted as `event: <topic>\ndata: <json>\n\n` frames. A periodic ping
// comment keeps proxies and browsers from idling the connection out.
//
// This endpoint is mounted under /api (so auth middleware applies). Phase 3
// shipped a topic-locked WebSocket at /ws; Phase 4 adds this SSE multiplex
// for the dashboard widgets that need multiple feeds.
func (s *Server) handleSSE(w http.ResponseWriter, r *http.Request) {
	if s.bus == nil {
		http.Error(w, "eventbus not configured", http.StatusServiceUnavailable)
		return
	}
	topics := parseSSETopics(r.URL.Query().Get("topic"))
	if len(topics) == 0 {
		http.Error(w, "topic query parameter required", http.StatusBadRequest)
		return
	}

	flusher, ok := w.(http.Flusher)
	if !ok {
		http.Error(w, "streaming not supported", http.StatusInternalServerError)
		return
	}
	w.Header().Set("Content-Type", "text/event-stream")
	w.Header().Set("Cache-Control", "no-store")
	w.Header().Set("Connection", "keep-alive")
	w.Header().Set("X-Accel-Buffering", "no")
	w.WriteHeader(http.StatusOK)
	flusher.Flush()

	subs := make([]*eventbus.Subscription, 0, len(topics))
	for _, t := range topics {
		subs = append(subs, s.bus.Subscribe(t))
	}
	defer func() {
		for _, sub := range subs {
			sub.Close()
		}
	}()

	cases := make([]selectCase, 0, len(subs))
	for _, sub := range subs {
		cases = append(cases, selectCase{ch: sub.Chan()})
	}

	pingTimer := time.NewTicker(20 * time.Second)
	defer pingTimer.Stop()

	for {
		// Manual select across N subscription channels via fan-in. Avoiding
		// reflect.Select for ~5 topics keeps the hot path simple.
		select {
		case <-r.Context().Done():
			return
		case <-pingTimer.C:
			if _, err := fmt.Fprintf(w, ": keep-alive\n\n"); err != nil {
				return
			}
			flusher.Flush()
		default:
			// Try each subscription in turn without blocking. The 50 ms
			// timeout puts a ceiling on per-iteration latency.
			delivered := false
			for i, c := range cases {
				select {
				case ev, ok := <-c.ch:
					if !ok {
						continue
					}
					payload, _ := json.Marshal(ev.Payload)
					if _, err := fmt.Fprintf(w, "event: %s\ndata: %s\n\n", ev.Topic, payload); err != nil {
						return
					}
					flusher.Flush()
					delivered = true
					_ = i
				default:
				}
			}
			if !delivered {
				// Short blocking wait so we don't busy-loop. The ping ticker
				// or the request context cancellation will preempt us.
				time.Sleep(50 * time.Millisecond)
			}
		}
	}
}

type selectCase struct{ ch <-chan eventbus.Event }

func parseSSETopics(s string) []eventbus.Topic {
	if s == "" {
		return nil
	}
	parts := strings.Split(s, ",")
	out := make([]eventbus.Topic, 0, len(parts))
	seen := map[string]struct{}{}
	for _, p := range parts {
		p = strings.TrimSpace(p)
		if p == "" {
			continue
		}
		if _, dup := seen[p]; dup {
			continue
		}
		seen[p] = struct{}{}
		out = append(out, eventbus.Topic(p))
	}
	return out
}