navi/internal/apphost/events.go

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 invalidates the idle rev cache and pushes a small SSE payload.
func (a *App) NotifyState(kind string) {
	if a == nil {
		return
	}
	a.invalidateRevCache()
	if 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()
		}
	}
}