// Package api exposes the REST endpoints and the SSE stream. package api import ( "encoding/json" "fmt" "net/http" "strconv" "time" "dash/internal/engine" "dash/internal/store" ) type API struct { engine *engine.Engine store *store.Store hub *engine.Hub } func New(e *engine.Engine, s *store.Store, h *engine.Hub) *API { return &API{engine: e, store: s, hub: h} } func (a *API) Register(mux *http.ServeMux) { mux.HandleFunc("GET /api/targets", a.handleTargets) mux.HandleFunc("GET /api/targets/{id}/history", a.handleHistory) mux.HandleFunc("GET /api/incidents", a.handleIncidents) mux.HandleFunc("GET /api/stream", a.handleStream) } func writeJSON(w http.ResponseWriter, v any) { w.Header().Set("Content-Type", "application/json") json.NewEncoder(w).Encode(v) } // parseRange turns "30m", "24h", "7d" into a duration. Default 24h. func parseRange(s string) (time.Duration, error) { if s == "" { return 24 * time.Hour, nil } if n, err := strconv.Atoi(s[:len(s)-1]); err == nil && n > 0 { switch s[len(s)-1] { case 'm': return time.Duration(n) * time.Minute, nil case 'h': return time.Duration(n) * time.Hour, nil case 'd': return time.Duration(n) * 24 * time.Hour, nil } } return 0, fmt.Errorf("invalid range %q (use e.g. 30m, 24h, 7d)", s) } // TargetView is the API shape for one target's current state. type TargetView struct { ID string `json:"id"` Name string `json:"name"` Type string `json:"type"` Endpoint string `json:"endpoint"` IntervalSec int `json:"interval_sec"` State engine.State `json:"state"` Fails int `json:"fails"` Last *store.Check `json:"last,omitempty"` Uptime24h *float64 `json:"uptime_24h,omitempty"` Uptime7d *float64 `json:"uptime_7d,omitempty"` Uptime30d *float64 `json:"uptime_30d,omitempty"` Latency store.LatencyStats `json:"latency"` CertDays *int `json:"cert_days,omitempty"` Incident *store.Incident `json:"incident,omitempty"` } func (a *API) targetViews() []TargetView { now := time.Now().UnixMilli() day := now - 24*time.Hour.Milliseconds() week := now - 7*24*time.Hour.Milliseconds() month := now - 30*24*time.Hour.Milliseconds() snap := a.engine.Snapshot() out := make([]TargetView, 0, len(snap)) for _, ts := range snap { t := ts.Target v := TargetView{ ID: t.ID, Name: t.Name, Type: string(t.Type), Endpoint: endpointOf(ts), IntervalSec: int(t.Interval.Duration.Seconds()), State: ts.State, Fails: ts.Fails, Incident: ts.Incident, } if ts.Last != nil { v.Last = &store.Check{ TS: ts.Last.TS.UnixMilli(), OK: ts.Last.OK, LatencyMS: ts.Last.LatencyMS, Detail: ts.Last.Detail, Err: ts.Last.Err, } if t.Type == "tls" && ts.Last.Detail != "" { if d, err := strconv.Atoi(ts.Last.Detail); err == nil { v.CertDays = &d } } } if ok, total, err := a.store.Uptime(t.ID, day); err == nil && total > 0 { pct := 100 * float64(ok) / float64(total) v.Uptime24h = &pct } if ok, total, err := a.store.Uptime(t.ID, week); err == nil && total > 0 { pct := 100 * float64(ok) / float64(total) v.Uptime7d = &pct } if ok, total, err := a.store.Uptime(t.ID, month); err == nil && total > 0 { pct := 100 * float64(ok) / float64(total) v.Uptime30d = &pct } if stats, err := a.store.Latency(t.ID, day); err == nil { v.Latency = stats } out = append(out, v) } return out } func endpointOf(ts engine.TargetStatus) string { t := ts.Target switch { case t.URL != "": return t.URL case t.Address != "": return t.Address case t.Container != "": return t.Container default: return t.Host } } func (a *API) handleTargets(w http.ResponseWriter, r *http.Request) { writeJSON(w, a.targetViews()) } type historyResponse struct { TargetID string `json:"target_id"` Range string `json:"range"` From int64 `json:"from"` To int64 `json:"to"` Points []store.Check `json:"points"` Buckets []store.Bucket `json:"buckets"` BucketMS int64 `json:"bucket_ms"` } func (a *API) handleHistory(w http.ResponseWriter, r *http.Request) { id := r.PathValue("id") rangeStr := r.URL.Query().Get("range") dur, err := parseRange(rangeStr) if err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } if rangeStr == "" { rangeStr = "24h" } to := time.Now().UnixMilli() from := to - dur.Milliseconds() points, err := a.store.History(id, from, to, 500) if err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) return } // 24 buckets across the window: 1h each for 24h, scaling with range. bucketMS := dur.Milliseconds() / 24 buckets, err := a.store.Buckets(id, from, to, bucketMS) if err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) return } if points == nil { points = []store.Check{} } if buckets == nil { buckets = []store.Bucket{} } writeJSON(w, historyResponse{ TargetID: id, Range: rangeStr, From: from, To: to, Points: points, Buckets: buckets, BucketMS: bucketMS, }) } func (a *API) handleIncidents(w http.ResponseWriter, r *http.Request) { dur, err := parseRange(r.URL.Query().Get("range")) if err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } from := time.Now().Add(-dur).UnixMilli() incidents, err := a.store.Incidents(from) if err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) return } writeJSON(w, incidents) } // handleStream is the SSE endpoint. Events: // // snapshot — full target state, sent once on connect // check — every completed probe // state — confirmed up/degraded/down transitions func (a *API) handleStream(w http.ResponseWriter, r *http.Request) { flusher, ok := w.(http.Flusher) if !ok { http.Error(w, "streaming 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.Header().Set("X-Accel-Buffering", "no") send := func(name string, v any) bool { data, err := json.Marshal(v) if err != nil { return true } if _, err := fmt.Fprintf(w, "event: %s\ndata: %s\n\n", name, data); err != nil { return false } flusher.Flush() return true } if !send("snapshot", a.targetViews()) { return } sub := a.hub.Subscribe() defer a.hub.Unsubscribe(sub) keepalive := time.NewTicker(25 * time.Second) defer keepalive.Stop() for { select { case <-r.Context().Done(): return case ev := <-sub: if !send(ev.Name, ev.Data) { return } case <-keepalive.C: if _, err := fmt.Fprint(w, ": keepalive\n\n"); err != nil { return } flusher.Flush() } } }