diff --git a/README.md b/README.md index ce9c413..98b1637 100644 --- a/README.md +++ b/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) diff --git a/db/queries/queries.sql b/db/queries/queries.sql index 5fc81c7..27ee902 100644 --- a/db/queries/queries.sql +++ b/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 -- ============================================================ diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index d911b18..a2965a0 100644 --- a/db/sqlc/queries.sql.go +++ b/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 ` diff --git a/db/store.go b/db/store.go index 64e50ab..135929f 100644 --- a/db/store.go +++ b/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 +} diff --git a/internal/api/handlers/observers.go b/internal/api/handlers/observers.go index 34b52cb..620f067 100644 --- a/internal/api/handlers/observers.go +++ b/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= 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= 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): diff --git a/internal/api/reader.go b/internal/api/reader.go index 424fdc3..42d107d 100644 --- a/internal/api/reader.go +++ b/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 }