package apphost import ( "encoding/json" "fmt" "net/http" "sync" "time" ) // eventBroker fans out lightweight UI notifications (SSE). type eventBroker struct { mu sync.Mutex subs map[chan []byte]struct{} } func newEventBroker() *eventBroker { return &eventBroker{subs: make(map[chan []byte]struct{})} } func (b *eventBroker) subscribe() chan []byte { ch := make(chan []byte, 8) b.mu.Lock() b.subs[ch] = struct{}{} b.mu.Unlock() return ch } func (b *eventBroker) unsubscribe(ch chan []byte) { b.mu.Lock() if _, ok := b.subs[ch]; ok { delete(b.subs, ch) close(ch) } b.mu.Unlock() } func (b *eventBroker) publish(payload []byte) { b.mu.Lock() defer b.mu.Unlock() for ch := range b.subs { select { case ch <- payload: default: // Drop if a slow client is behind; safety-net poll will catch up. } } } // NotifyState pushes a small SSE payload so the UI can refresh immediately. func (a *App) NotifyState(kind string) { if a == nil || a.events == nil { return } if kind == "" { kind = "state" } b, err := json.Marshal(map[string]string{"type": kind}) if err != nil { return } a.events.publish(b) } func (a *App) handleEvents(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet { http.Error(w, "GET only", http.StatusMethodNotAllowed) return } if a.APIToken != "" { tok := r.Header.Get("X-Navis-Token") if tok == "" { tok = r.URL.Query().Get("t") } if tok != a.APIToken { http.Error(w, "unauthorized", http.StatusUnauthorized) return } } flusher, ok := w.(http.Flusher) if !ok { http.Error(w, "stream unsupported", http.StatusInternalServerError) return } w.Header().Set("Content-Type", "text/event-stream") w.Header().Set("Cache-Control", "no-cache") w.Header().Set("Connection", "keep-alive") w.WriteHeader(http.StatusOK) flusher.Flush() ch := a.events.subscribe() defer a.events.unsubscribe(ch) ping := time.NewTicker(20 * time.Second) defer ping.Stop() for { select { case <-r.Context().Done(): return case <-a.ctx.Done(): return case <-ping.C: if _, err := fmt.Fprintf(w, ": ping\n\n"); err != nil { return } flusher.Flush() case msg, ok := <-ch: if !ok { return } if _, err := fmt.Fprintf(w, "data: %s\n\n", msg); err != nil { return } flusher.Flush() } } }