8c5e75b06b
- Add formatTime template func for human-readable timestamps (just now, 2m ago, etc.) - Add client-side formatTimeStr for SSE live-update timestamps - Fix SSE body rendering: buildRequestBody() populates headers/query/body for live-added requests - Fix SSE body encoding: use string in sseRequest to avoid base64 encoding of []byte - Add inspect handler tests: renders HTML, empty state, auto-creates endpoint - Timestamps include title attr with full ISO timestamp for hover detail
126 lines
2.8 KiB
Go
126 lines
2.8 KiB
Go
package handler
|
|
|
|
import (
|
|
"encoding/json"
|
|
"fmt"
|
|
"net/http"
|
|
"sync"
|
|
|
|
"github.com/elfoundation/hatch/internal/store"
|
|
)
|
|
|
|
// sseHub manages SSE subscribers per endpoint.
|
|
type sseHub struct {
|
|
mu sync.RWMutex
|
|
subs map[string]map[chan []byte]struct{}
|
|
}
|
|
|
|
var hub = &sseHub{
|
|
subs: make(map[string]map[chan []byte]struct{}),
|
|
}
|
|
|
|
func (h *sseHub) subscribe(endpointID string) chan []byte {
|
|
ch := make(chan []byte, 64)
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
if h.subs[endpointID] == nil {
|
|
h.subs[endpointID] = make(map[chan []byte]struct{})
|
|
}
|
|
h.subs[endpointID][ch] = struct{}{}
|
|
return ch
|
|
}
|
|
|
|
func (h *sseHub) unsubscribe(endpointID string, ch chan []byte) {
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
delete(h.subs[endpointID], ch)
|
|
if len(h.subs[endpointID]) == 0 {
|
|
delete(h.subs, endpointID)
|
|
}
|
|
}
|
|
|
|
// sseRequest is the JSON payload sent over SSE.
|
|
// It uses string for Body to avoid base64 encoding.
|
|
type sseRequest struct {
|
|
ID string `json:"id"`
|
|
EndpointID string `json:"endpoint_id"`
|
|
Method string `json:"method"`
|
|
Path string `json:"path"`
|
|
Headers string `json:"headers"`
|
|
Query string `json:"query"`
|
|
Body string `json:"body"`
|
|
CreatedAt string `json:"created_at"`
|
|
}
|
|
|
|
func (h *sseHub) broadcast(endpointID string, req *store.Request) {
|
|
h.mu.RLock()
|
|
defer h.mu.RUnlock()
|
|
subs := h.subs[endpointID]
|
|
if subs == nil {
|
|
return
|
|
}
|
|
sseReq := sseRequest{
|
|
ID: req.ID,
|
|
EndpointID: req.EndpointID,
|
|
Method: req.Method,
|
|
Path: req.Path,
|
|
Headers: req.Headers,
|
|
Query: req.Query,
|
|
Body: string(req.Body),
|
|
CreatedAt: req.CreatedAt,
|
|
}
|
|
data, err := json.Marshal(sseReq)
|
|
if err != nil {
|
|
return
|
|
}
|
|
for ch := range subs {
|
|
select {
|
|
case ch <- data:
|
|
default:
|
|
}
|
|
}
|
|
}
|
|
|
|
// broadcastRequest publishes req to SSE subscribers for its endpoint.
|
|
func broadcastRequest(endpointID string, req *store.Request) {
|
|
hub.broadcast(endpointID, req)
|
|
}
|
|
|
|
// HandleSSE serves an SSE stream for GET /e/{endpointID}/events.
|
|
func HandleSSE(repo store.Repository) http.HandlerFunc {
|
|
return func(w http.ResponseWriter, r *http.Request) {
|
|
endpointID := r.PathValue("endpointID")
|
|
if endpointID == "" {
|
|
http.Error(w, "missing endpoint ID", 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-cache")
|
|
w.Header().Set("Connection", "keep-alive")
|
|
w.WriteHeader(http.StatusOK)
|
|
flusher.Flush()
|
|
|
|
ch := hub.subscribe(endpointID)
|
|
defer hub.unsubscribe(endpointID, ch)
|
|
|
|
ctx := r.Context()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case data, ok := <-ch:
|
|
if !ok {
|
|
return
|
|
}
|
|
fmt.Fprintf(w, "data: %s\n\n", data)
|
|
flusher.Flush()
|
|
}
|
|
}
|
|
}
|
|
}
|