beacon-server/internal/api/handlers/observers.go

// 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)
	}
}