// Copyright 2026 Beacon Contributors // SPDX-License-Identifier: AGPL-3.0-or-later package handlers import ( "errors" "net/http" "strconv" "time" "github.com/MeshCore-Beacon/beacon-server/internal/api" "github.com/go-chi/chi/v5" "github.com/google/uuid" "github.com/jackc/pgx/v5" ) // ObserversRouter mounts all /observers routes onto a subrouter. // // GET /observers → listObservers // GET /observers/{observerId} → getObserver // GET /observers/{observerId}/telemetry → getObserverTelemetry // GET /observers/{observerId}/activity → getObserverActivity // GET /observers/{observerId}/adverts → listObserverAdverts func ObserversRouter(reader api.Reader) http.Handler { r := chi.NewRouter() r.Get("/", listObservers(reader)) r.Get("/directory", listObserverDirectory(reader)) r.Route("/{observerId}", func(r chi.Router) { r.Get("/", getObserver(reader)) r.Get("/adverts", listObserverAdverts(reader)) r.Get("/telemetry", getObserverTelemetry(reader)) r.Get("/activity", getObserverActivity(reader)) }) return r } // listObservers godoc // // @Summary List observers // @Tags Observers // @Produce json // @Param iata query string false "Filter by single IATA code (case-insensitive)" // @Param iatas query string false "Filter by multiple IATA codes, comma-separated e.g. YVR,YYJ" // @Param regionId query int false "Filter by region ID, expands to member IATAs" // @Param region query string false "Filter by region slug, expands to member IATAs" // @Param type query string false "Filter by observer type (e.g. meshcoretomqtt, meshcore-ha)" // @Param broker query string false "Filter by broker name" // @Param status query string false "Filter by status (online or offline)" // @Param name query string false "Partial case-insensitive display name match" // @Param scope query string false "Filter by transport scope name e.g. %23bc (URL-encoded #bc)" // @Param cursor query int false "last_seen epoch ms of last item for pagination" // @Param limit query int false "Max results (default 50); must be positive, values above 200 are clamped" minimum(1) maximum(200) // @Success 200 {object} api.Page[api.ObserverSummary] // @Failure 400 {object} handlers.APIError // @Failure 500 {object} handlers.APIError // @Router /observers [get] func listObservers(reader api.Reader) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { observerType := r.URL.Query().Get("type") broker := r.URL.Query().Get("broker") name := r.URL.Query().Get("name") status := r.URL.Query().Get("status") scope := r.URL.Query().Get("scope") var cursor int64 if cursorParam := r.URL.Query().Get("cursor"); cursorParam != "" { c, err := strconv.ParseInt(cursorParam, 10, 64) if err != nil { respondError(w, http.StatusBadRequest, "cursor must be an integer") return } cursor = c } limit, err := parseLimit(r, 50) if err != nil { respondError(w, http.StatusBadRequest, err.Error()) return } iatas := parseIATAs(r) if regionIDStr := r.URL.Query().Get("regionId"); regionIDStr != "" || r.URL.Query().Get("region") != "" { regionIATAs, err := resolveRegionIATAs(r.Context(), regionIDStr, r.URL.Query().Get("region"), reader) if err != nil { respondRegionError(w, err) return } iatas = append(iatas, regionIATAs...) } observers, err := reader.ListObservers(r.Context(), iatas, observerType, broker, status, name, scope, cursor, limit) if err != nil { respondError(w, http.StatusInternalServerError, "failed to get list of observers") return } respond(w, http.StatusOK, observers) } } // getObserver godoc // // @Summary Get observer detail // @Tags Observers // @Produce json // @Param observerId path string true "Observer UUID" // @Success 200 {object} api.Observer // @Failure 400 {object} handlers.APIError // @Failure 404 {object} handlers.APIError // @Failure 500 {object} handlers.APIError // @Router /observers/{observerId} [get] func getObserver(reader api.Reader) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { observerID := chi.URLParam(r, "observerId") id, err := uuid.Parse(observerID) if err != nil { respondError(w, http.StatusBadRequest, "failed to parse observer UUID") return } obs, err := reader.GetObserver(r.Context(), id) switch { case errors.Is(err, pgx.ErrNoRows), err == nil && obs == nil: respondError(w, http.StatusNotFound, "observer not found") return case err != nil: respondError(w, http.StatusInternalServerError, "internal server error") return } respond(w, http.StatusOK, obs) } } // listObserverAdverts godoc // // @Summary List advert packets heard by an observer // @Tags Observers // @Produce json // @Param observerId path string true "Observer UUID" // @Param cursor query int false "Observation ID of last item for pagination" // @Param limit query int false "Max results (default 50); must be positive, values above 200 are clamped" minimum(1) maximum(200) // @Success 200 {object} api.Page[api.AdvertObservation] // @Failure 400 {object} handlers.APIError // @Failure 500 {object} handlers.APIError // @Router /observers/{observerId}/adverts [get] func listObserverAdverts(reader api.Reader) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { observerID, err := uuid.Parse(chi.URLParam(r, "observerId")) if err != nil { respondError(w, http.StatusBadRequest, "invalid observer ID") return } var cursor int64 if cursorParam := r.URL.Query().Get("cursor"); cursorParam != "" { c, err := strconv.ParseInt(cursorParam, 10, 64) if err != nil { respondError(w, http.StatusBadRequest, "cursor must be an integer") return } cursor = c } limit, err := parseLimit(r, 50) if err != nil { respondError(w, http.StatusBadRequest, err.Error()) return } adverts, err := reader.ListObserverAdverts(r.Context(), observerID, cursor, limit) if err != nil { respondError(w, http.StatusInternalServerError, "internal server error") return } respond(w, http.StatusOK, adverts) } } // getObserverTelemetry godoc // // @Summary Get observer telemetry history // @Tags Observers // @Produce json // @Param observerId path string true "Observer UUID" // @Param range query string false "Duration window e.g. 24h, 48h, 168h (default 24h)" // @Param afterId query int false "Return points after this telemetry ID for WS reconnection backfill" // @Param interval query string false "Bucketing interval: 1h (default), 6h, or 24h" // @Success 200 {object} api.ObserverTelemetry // @Failure 400 {object} handlers.APIError // @Failure 500 {object} handlers.APIError // @Router /observers/{observerId}/telemetry [get] func getObserverTelemetry(reader api.Reader) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { observerID, err := uuid.Parse(chi.URLParam(r, "observerId")) if err != nil { respondError(w, http.StatusBadRequest, "invalid observer ID") return } rangeParam := r.URL.Query().Get("range") if rangeParam == "" { rangeParam = "24h" } duration, err := time.ParseDuration(rangeParam) if err != nil { respondError(w, http.StatusBadRequest, "invalid range, use e.g. 24h, 48h, 168h") return } afterID := int64(0) if afterIDParam := r.URL.Query().Get("afterId"); afterIDParam != "" { id, err := strconv.ParseInt(afterIDParam, 10, 64) if err != nil { respondError(w, http.StatusBadRequest, "afterId must be an integer") return } afterID = id } intervalParam := r.URL.Query().Get("interval") if intervalParam == "" { intervalParam = "1h" } var bucketHours int32 switch intervalParam { case "1h": bucketHours = 0 // use raw query case "6h": bucketHours = 6 case "24h": bucketHours = 24 default: respondError(w, http.StatusBadRequest, "invalid interval, use 1h, 6h or 24h") return } since := time.Now().Add(-duration) until := time.Time{} // no upper bound var telemetry *api.ObserverTelemetry if bucketHours == 0 { telemetry, err = reader.GetObserverTelemetry(r.Context(), observerID, since, until, afterID) } else { var points []api.ObserverTelemetryPoint points, err = reader.GetObserverTelemetryBucketed(r.Context(), observerID, since, until, bucketHours) if err == nil { telemetry = &api.ObserverTelemetry{Points: points} } } if err != nil { respondError(w, http.StatusInternalServerError, "internal server error") return } telemetry.Range = rangeParam telemetry.Interval = intervalParam respond(w, http.StatusOK, telemetry) } } // activityIntervals is the fixed bucket set; every value divides 24h so bucket // starts stay aligned to the clock regardless of when the window began. var activityIntervals = map[string]time.Duration{ "5m": 5 * time.Minute, "15m": 15 * time.Minute, "1h": time.Hour, "6h": 6 * time.Hour, "24h": 24 * time.Hour, } // getObserverActivity godoc // // @Summary Get observer heard-activity history // @Tags Observers // @Produce json // @Param observerId path string true "Observer UUID" // @Param range query string false "Trailing window as a Go duration, max 720h (default 24h); max 48h when interval is under 1h" // @Param interval query string false "Bucket size: 5m, 15m, 1h, 6h or 24h (default 15m)" // @Param until query int false "Optional exclusive end in epoch milliseconds; aligned down to a complete bucket, at most 30 days old. Activity uses this window; summary freshness/latestRecordedAt and the last complete hour are measured at response generation time, independently of until." // @Success 200 {object} api.ObserverActivity // @Failure 400 {object} handlers.APIError // @Failure 404 {object} handlers.APIError // @Failure 500 {object} handlers.APIError // @Router /observers/{observerId}/activity [get] func getObserverActivity(reader api.Reader) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { observerID, err := uuid.Parse(chi.URLParam(r, "observerId")) if err != nil { respondError(w, http.StatusBadRequest, "invalid observer ID") return } rangeParam := r.URL.Query().Get("range") if rangeParam == "" { rangeParam = "24h" } window, err := time.ParseDuration(rangeParam) if err != nil || window <= 0 || window > 720*time.Hour { respondError(w, http.StatusBadRequest, "invalid range, use a duration up to 720h e.g. 24h, 168h, 720h") return } intervalParam := r.URL.Query().Get("interval") if intervalParam == "" { intervalParam = "15m" } interval, ok := activityIntervals[intervalParam] if !ok { respondError(w, http.StatusBadRequest, "invalid interval, use 5m, 15m, 1h, 6h or 24h") return } // the raw path scans live observations, so keep sub-hour windows short if interval < time.Hour && window > 48*time.Hour { respondError(w, http.StatusBadRequest, "range must be 48h or less for intervals under 1h") return } if window/interval > 1000 { respondError(w, http.StatusBadRequest, "range/interval exceeds 1000 buckets") return } until := time.Time{} if values, ok := r.URL.Query()["until"]; ok { n, err := strconv.ParseInt(values[0], 10, 64) now := time.Now() if len(values) != 1 || err != nil || n < 0 || n > now.Add(clockSkewTolerance).UnixMilli() || n < now.Add(-720*time.Hour).UnixMilli() { respondError(w, http.StatusBadRequest, "until must be one past epoch-millisecond timestamp within 30 days") return } until = time.UnixMilli(n) } activity, err := reader.GetObserverActivity(r.Context(), observerID, window, interval, until) if err != nil { if errors.Is(err, pgx.ErrNoRows) { respondError(w, http.StatusNotFound, "observer not found") return } respondError(w, http.StatusInternalServerError, "internal server error") return } activity.Range = rangeParam activity.Interval = intervalParam respond(w, http.StatusOK, activity) } }