diff --git a/internal/api/api.go b/internal/api/api.go new file mode 100644 index 0000000..f318b4b --- /dev/null +++ b/internal/api/api.go @@ -0,0 +1,259 @@ +// 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() + } + } +}