add observer tele api

README.md

@@ -304,14 +304,14 @@ Tower server.
 - [x] REST API: IATAs, Regions
 - [x] REST API: Channels (list + detail + messages) with IATA filter
 - [x] REST API: Messages (cross-channel) with IATA filter
-- [x] REST API: Observers (list + detail with broker last-seen)
+- [x] REST API: Observers (telemetry, list + detail with broker last-seen)
 - [x] REST API: Brokers (list with connection status)
 
 ### In progress / next
 
 - [ ] REST API: Nodes (list + detail + observations)
 - [ ] REST API: Packets (list + detail)
-- [ ] REST API: Observer telemetry + adverts sub-endpoints
+- [ ] REST API: Observer + adverts sub-endpoints
 - [ ] REST API: Stats
 - [ ] REST API: Pagination
 - [ ] Path resolution (node short ID lookup)

db/queries/queries.sql

@@ -109,6 +109,16 @@ SELECT radio_freq_mhz, radio_bw_khz, radio_sf, radio_cr
 FROM observers
 WHERE id = $1;
 
+-- name: GetObserverTelemetry :many
+SELECT id, reported_at, battery_voltage_mv, airtime_tx_pct, airtime_rx_pct,
+       noise_floor_db, uptime_seconds, queue_length, debug_flags, receive_errors
+FROM observer_telemetry
+WHERE observer_id = $1
+  AND ($2::timestamptz IS NULL OR reported_at >= $2)
+  AND ($3::timestamptz IS NULL OR reported_at <= $3)
+  AND ($4 = 0 OR id > $4)
+ORDER BY reported_at ASC;
+
 -- ============================================================
 -- OBSERVER BROKERS
 -- ============================================================

db/sqlc/queries.sql.go

@@ -366,6 +366,73 @@ func (q *Queries) GetObserverRadio(ctx context.Context, id uuid.UUID) (GetObserv
 	return i, err
 }
 
+const getObserverTelemetry = `-- name: GetObserverTelemetry :many
+SELECT id, reported_at, battery_voltage_mv, airtime_tx_pct, airtime_rx_pct,
+       noise_floor_db, uptime_seconds, queue_length, debug_flags, receive_errors
+FROM observer_telemetry
+WHERE observer_id = $1
+  AND ($2::timestamptz IS NULL OR reported_at >= $2)
+  AND ($3::timestamptz IS NULL OR reported_at <= $3)
+  AND ($4 = 0 OR id > $4)
+ORDER BY reported_at ASC
+`
+
+type GetObserverTelemetryParams struct {
+	ObserverID uuid.UUID          `json:"observer_id"`
+	Column2    pgtype.Timestamptz `json:"column_2"`
+	Column3    pgtype.Timestamptz `json:"column_3"`
+	Column4    interface{}        `json:"column_4"`
+}
+
+type GetObserverTelemetryRow struct {
+	ID               int64              `json:"id"`
+	ReportedAt       pgtype.Timestamptz `json:"reported_at"`
+	BatteryVoltageMv *int32             `json:"battery_voltage_mv"`
+	AirtimeTxPct     *float32           `json:"airtime_tx_pct"`
+	AirtimeRxPct     *float32           `json:"airtime_rx_pct"`
+	NoiseFloorDb     *float32           `json:"noise_floor_db"`
+	UptimeSeconds    *int64             `json:"uptime_seconds"`
+	QueueLength      *int32             `json:"queue_length"`
+	DebugFlags       *int32             `json:"debug_flags"`
+	ReceiveErrors    *int32             `json:"receive_errors"`
+}
+
+func (q *Queries) GetObserverTelemetry(ctx context.Context, arg GetObserverTelemetryParams) ([]GetObserverTelemetryRow, error) {
+	rows, err := q.db.Query(ctx, getObserverTelemetry,
+		arg.ObserverID,
+		arg.Column2,
+		arg.Column3,
+		arg.Column4,
+	)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	items := []GetObserverTelemetryRow{}
+	for rows.Next() {
+		var i GetObserverTelemetryRow
+		if err := rows.Scan(
+			&i.ID,
+			&i.ReportedAt,
+			&i.BatteryVoltageMv,
+			&i.AirtimeTxPct,
+			&i.AirtimeRxPct,
+			&i.NoiseFloorDb,
+			&i.UptimeSeconds,
+			&i.QueueLength,
+			&i.DebugFlags,
+			&i.ReceiveErrors,
+		); err != nil {
+			return nil, err
+		}
+		items = append(items, i)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return items, nil
+}
+
 const getPacket = `-- name: GetPacket :one
 SELECT packet_hash, payload_type, payload_version, route_type, transport_codes_present, region_code, sub_region_code, origin_pubkey, raw_payload, parsed_payload, decrypted, channel_hash, first_heard_at, last_heard_at, observation_count FROM packets WHERE packet_hash = $1
 `

db/store.go

@@ -638,3 +638,33 @@ func (s *Store) DeleteOldTelemetry(ctx context.Context, cutoff time.Time) error
 func (s *Store) DeleteOldPackets(ctx context.Context, cutoff time.Time) error {
 	return s.q.DeleteOldPackets(ctx, pgtype.Timestamptz{Time: cutoff, Valid: true})
 }
+
+// GetObserverTelemetry returns telemetry points for an observer within the given time range.
+// since and until define the window; pass zero times to use defaults (last 24h).
+// TODO: implement server-side bucketing by interval when needed.
+// Currently returns all points in the range at stored resolution.
+func (s *Store) GetObserverTelemetry(ctx context.Context, observerID uuid.UUID, since, until time.Time, afterID int64) (*api.ObserverTelemetry, error) {
+	rows, err := s.q.GetObserverTelemetry(ctx, sqlc.GetObserverTelemetryParams{
+		ObserverID: observerID,
+		Column2:    pgtype.Timestamptz{Time: since, Valid: !since.IsZero()},
+		Column3:    pgtype.Timestamptz{Time: until, Valid: !until.IsZero()},
+		Column4:    afterID,
+	})
+	if err != nil {
+		return nil, err
+	}
+	points := make([]api.ObserverTelemetryPoint, 0, len(rows))
+	for _, v := range rows {
+		points = append(points, api.ObserverTelemetryPoint{
+			T:             v.ReportedAt.Time.Unix(),
+			BatteryMV:     v.BatteryVoltageMv,
+			AirtimeTxPct:  v.AirtimeTxPct,
+			AirtimeRxPct:  v.AirtimeRxPct,
+			NoiseFloorDB:  v.NoiseFloorDb,
+			UptimeSeconds: v.UptimeSeconds,
+			QueueLength:   v.QueueLength,
+			ReceiveErrors: v.ReceiveErrors,
+		})
+	}
+	return &api.ObserverTelemetry{Points: points}, nil
+}

internal/api/handlers/observers.go

@@ -2,6 +2,8 @@ package handlers
 
 import (
 	"net/http"
+	"strconv"
+	"time"
 
 	"github.com/MeshCore-Tower/tower-server/internal/api"
 	"github.com/go-chi/chi/v5"
@@ -58,30 +60,62 @@ func ObserversRouter(reader api.Reader) http.Handler {
 			}
 			respond(w, http.StatusOK, obs)
 		})
-		r.Get("/telemetry", GetObserverTelemetry)
 		r.Get("/adverts", ListObserverAdverts)
+		// GET /api/v1/observers/{observerId}/telemetry
+		//
+		// Query params (all optional):
+		//
+		//	range=24h              duration string: 24h, 7d, 30d
+		//	afterId=<status id>    for deterministic WS reconnection backfill
+		//
+		// Returns a time-bucketed array of telemetry points suitable for charting
+		// (battery, airtime, noise floor, uptime, queue depth, receive errors).
+		r.Get("/telemetry", 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
+			}
+
+			since := time.Now().Add(-duration)
+			until := time.Time{} // no upper bound
+
+			telemetry, err := reader.GetObserverTelemetry(r.Context(), observerID, since, until, afterID)
+			if err != nil {
+				respondError(w, http.StatusInternalServerError, "internal server error")
+				return
+			}
+
+			telemetry.Range = rangeParam
+			telemetry.Interval = r.URL.Query().Get("interval") // echoed back, not used server-side yet
+			respond(w, http.StatusOK, telemetry)
+		})
 	})
 
 	return r
 }
 
-// GetObserverTelemetry handles GET /api/v1/observers/{observerId}/telemetry
-//
-// Query params (all optional):
-//
-//	range=24h              duration string: 24h, 7d, 30d
-//	afterId=<status id>    for deterministic WS reconnection backfill
-//	limit=100
-//
-// Returns a time-bucketed array of telemetry points suitable for charting
-// (battery, airtime, noise floor, uptime, queue depth, receive errors).
-func GetObserverTelemetry(w http.ResponseWriter, r *http.Request) {
-	// observerId := chi.URLParam(r, "observerId")
-	// TODO: query status_metadata history, bucket by interval, write JSON response.
-	// afterId (int64): WHERE id > afterId ORDER BY id ASC LIMIT limit
-	w.WriteHeader(http.StatusNotImplemented)
-}
-
 // ListObserverAdverts handles GET /api/v1/observers/{observerId}/adverts
 //
 // Query params (all optional):

internal/api/reader.go

@@ -9,6 +9,26 @@ import (
 	"github.com/google/uuid"
 )
 
+// ObserverTelemetryPoint is a single telemetry snapshot for an observer.
+type ObserverTelemetryPoint struct {
+	T             int64    `json:"t"` // epoch ms
+	BatteryMV     *int32   `json:"batteryMv,omitempty"`
+	AirtimeTxPct  *float32 `json:"airtimeTxPct,omitempty"`
+	AirtimeRxPct  *float32 `json:"airtimeRxPct,omitempty"`
+	NoiseFloorDB  *float32 `json:"noiseFloorDb,omitempty"`
+	UptimeSeconds *int64   `json:"uptimeSeconds,omitempty"`
+	QueueLength   *int32   `json:"queueLength,omitempty"`
+	ReceiveErrors *int32   `json:"receiveErrors,omitempty"`
+}
+
+// ObserverTelemetry is the full telemetry response for an observer.
+// Range and interval reflect the query parameters used.
+type ObserverTelemetry struct {
+	Range    string                   `json:"range"`
+	Interval string                   `json:"interval"`
+	Points   []ObserverTelemetryPoint `json:"points"`
+}
+
 // ObserverSummary is the minimal observer representation used in list responses.
 type ObserverSummary struct {
 	ID           uuid.UUID `json:"id"`
@@ -109,6 +129,16 @@ type IATA struct {
 	Lng         *float64 `json:"lon"`
 }
 
+// Page is a generic paginated response envelope used by all list endpoints
+// that support cursor-based pagination. NextCursor is the ID of the last item
+// returned and should be passed as the cursor param in the next request.
+// HasMore is true when additional results exist beyond the current page.
+type Page[T any] struct {
+	Items      []T    `json:"items"`
+	NextCursor *int64 `json:"nextCursor,omitempty"`
+	HasMore    bool   `json:"hasMore"`
+}
+
 type Reader interface {
 	// ListIATAs returns all known IATA codes with display name and coordinates.
 	// IATAs are auto-created on first packet arrival from that location.
@@ -148,4 +178,8 @@ type Reader interface {
 	// GetObserver returns full detail for a single observer by UUID.
 	// Returns nil, pgx.ErrNoRows if the observer is not found.
 	GetObserver(ctx context.Context, observerID uuid.UUID) (*Observer, error)
+	// GetObserverTelemetry returns telemetry points for an observer within the given time range.
+	// since and until define the window; pass zero times to use defaults (last 24h).
+	GetObserverTelemetry(ctx context.Context, observerID uuid.UUID, since, until time.Time, afterID int64) (*ObserverTelemetry, error)
+	// TODO: add interval time.Duration param for server-side bucketing
 }