add stats api

README.md

@@ -228,36 +228,31 @@ Base path: `/api/v1`
 
 ### Implemented
 
-| Method | Path                      | Description                                                                                        |
-| ------ | ------------------------- | -------------------------------------------------------------------------------------------------- |
-| `GET`  | `/brokers`                | List MQTT brokers and connection status                                                            |
-| `GET`  | `/iatas`                  | List all known IATA codes                                                                          |
-| `GET`  | `/iatas/{iata}`           | Get a single IATA code                                                                             |
-| `GET`  | `/regions`                | List all regions (summary)                                                                         |
-| `GET`  | `/regions/{id}`           | Get a single region with IATA list                                                                 |
-| `GET`  | `/channels`               | List channels (optional: `?hash=<hex>&iata=<code>&limit=50`)                                       |
-| `GET`  | `/channels/{id}`          | Get channel detail by integer ID                                                                   |
-| `GET`  | `/channels/{id}/messages` | List messages for a channel (optional: `?since=<ms>&iata=<code>&limit=50`)                         |
-| `GET`  | `/messages`               | List all messages (optional: `?channelId=<int>&channelHash=<hex>&iata=<code>&since=<ms>&limit=50`) |
-| `GET`  | `/observers`              | List observers (optional: `?iata=<code>&type=<str>&broker=<name>&status=online\|offline`)          |
-| `GET`  | `/observers/{observerId}` | Get observer detail including broker last-seen timestamps                                          |
-
-### Stubbed (501 Not Implemented)
-
-| Method | Path                                | Description                        |
-| ------ | ----------------------------------- | ---------------------------------- |
-| `GET`  | `/packets`                          | List packets with filters          |
-| `GET`  | `/packets/{packetHash}`             | Get packet with all observations   |
-| `GET`  | `/nodes`                            | List nodes                         |
-| `GET`  | `/nodes/{nodeId}`                   | Get node detail                    |
-| `GET`  | `/nodes/{nodeId}/observations`      | List observations for a node       |
-| `GET`  | `/observers/{observerId}/telemetry` | Observer telemetry history         |
-| `GET`  | `/observers/{observerId}/adverts`   | Adverts heard by observer          |
-| `GET`  | `/stats/overview`                   | Network overview stats             |
-| `GET`  | `/stats/observations`               | Observation time series            |
-| `GET`  | `/stats/payloadBreakdown`           | Observations by payload type       |
-| `GET`  | `/stats/topNodes`                   | Top nodes by observation count     |
-| `GET`  | `/stats/topObservers`               | Top observers by observation count |
+| Method | Path                                | Description                                                                                        |
+| ------ | ----------------------------------- | -------------------------------------------------------------------------------------------------- |
+| `GET`  | `/brokers`                          | List MQTT brokers and connection status                                                            |
+| `GET`  | `/iatas`                            | List all known IATA codes                                                                          |
+| `GET`  | `/iatas/{iata}`                     | Get a single IATA code                                                                             |
+| `GET`  | `/regions`                          | List all regions (summary)                                                                         |
+| `GET`  | `/regions/{id}`                     | Get a single region with IATA list                                                                 |
+| `GET`  | `/channels`                         | List channels (optional: `?hash=<hex>&iata=<code>&limit=50`)                                       |
+| `GET`  | `/channels/{id}`                    | Get channel detail by integer ID                                                                   |
+| `GET`  | `/channels/{id}/messages`           | List messages for a channel (optional: `?since=<ms>&iata=<code>&limit=50`)                         |
+| `GET`  | `/messages`                         | List all messages (optional: `?channelId=<int>&channelHash=<hex>&iata=<code>&since=<ms>&limit=50`) |
+| `GET`  | `/observers`                        | List observers (optional: `?iata=<code>&type=<str>&broker=<name>&status=online\|offline`)          |
+| `GET`  | `/observers/{observerId}`           | Get observer detail including broker last-seen timestamps                                          |
+| `GET`  | `/packets`                          | List packets with filters                                                                          |
+| `GET`  | `/packets/{packetHash}`             | Get packet with all observations                                                                   |
+| `GET`  | `/nodes`                            | List nodes                                                                                         |
+| `GET`  | `/nodes/{nodeId}`                   | Get node detail                                                                                    |
+| `GET`  | `/nodes/{nodeId}/observations`      | List observations for a node                                                                       |
+| `GET`  | `/observers/{observerId}/telemetry` | Observer telemetry history                                                                         |
+| `GET`  | `/observers/{observerId}/adverts`   | Adverts heard by observer                                                                          |
+| `GET`  | `/stats/overview`                   | Network overview stats                                                                             |
+| `GET`  | `/stats/observations`               | Hourly observation time series (last 7 days by default)                                            |
+| `GET`  | `/stats/payload-breakdown`          | Observation counts by payload type (last 24h by default)                                           |
+| `GET`  | `/stats/top-nodes`                  | Top N nodes by observation count (from materialized view)                                          |
+| `GET`  | `/stats/top-observers`              | Top N observers by observation count (last 24h by default)                                         |
 
 ---
 
@@ -310,10 +305,11 @@ Tower server.
 - [x] REST API: Pagination
 - [x] REST API: Nodes (list + detail + observations)
 - [x] REST API: Packets (list + detail)
+- [x] REST API: Stats
 
 ### In progress / next
 
-- [ ] REST API: Stats
+- [ ] Materialized view refresh (mv_hourly_iata_stats, mv_top_nodes_by_iata)
 - [ ] Path resolution (node short ID lookup)
 - [ ] Propagation time calculation
 - [ ] Routes and traces endpoints
@@ -328,4 +324,4 @@ Tower server.
 - [ ] Region management via API (currently config-file only)
 - [ ] WebSocket subscription unsubscribe (scaffolded, not implemented)
 - [ ] Observer owner tracking (schema exists, API excluded by design)
-- [ ] Swagger/OpenAPI documentation
+- [ ] Swagger/OpenAPI documentation via swaggo/swag

db/migrations/001_schema.sql

@@ -312,7 +312,7 @@ CREATE INDEX idx_channel_messages_sent_brin ON channel_messages USING BRIN (sent
 CREATE MATERIALIZED VIEW mv_hourly_iata_stats AS
 SELECT
   iata,
-  date_trunc('hour', heard_at) AS hour,
+  date_trunc('hour', heard_at)::timestamptz AS hour,
   COUNT(*) AS observation_count,
   COUNT(DISTINCT packet_hash) AS unique_packets,
   COUNT(DISTINCT observer_id) AS active_observers
@@ -320,9 +320,6 @@ FROM packet_observations
 WHERE heard_at > NOW() - INTERVAL '7 days'
 GROUP BY iata, date_trunc('hour', heard_at);
 
-CREATE UNIQUE INDEX idx_mv_hourly_iata
-  ON mv_hourly_iata_stats(iata, hour);
-
 CREATE MATERIALIZED VIEW mv_top_nodes_by_iata AS
 SELECT
   ni.iata,

db/queries/queries.sql

@@ -488,11 +488,12 @@ SELECT
   COUNT(DISTINCT po.iata)         AS active_iatas
 FROM packet_observations po
 WHERE po.heard_at > NOW() - INTERVAL '24 hours'
-  AND ($1::char(3) IS NULL OR po.iata = $1);
+  AND ($1 = '' OR po.iata ILIKE $1);
 
 -- name: GetHourlyStats :many
-SELECT * FROM mv_hourly_iata_stats
-WHERE ($1::char(3) IS NULL OR iata = $1)
+SELECT iata, hour, observation_count, unique_packets, active_observers
+FROM mv_hourly_iata_stats
+WHERE ($1 = '' OR iata ILIKE $1)
   AND hour >= NOW() - $2::interval
 ORDER BY iata, hour;
 
@@ -502,6 +503,38 @@ WHERE ($1::char(3) IS NULL OR iata = $1)
 ORDER BY observation_count DESC
 LIMIT $2;
 
+-- name: GetStatsPayloadBreakdown :many
+-- Returns observation counts grouped by payload type for the given window and IATA.
+SELECT
+  p.payload_type,
+  COUNT(*) AS count
+FROM packet_observations po
+JOIN packets p ON p.packet_hash = po.packet_hash
+WHERE po.heard_at > $1
+  AND ($2 = '' OR po.iata ILIKE $2)
+GROUP BY p.payload_type
+ORDER BY count DESC;
+
+-- name: GetStatsTopObservers :many
+-- Returns the top N observers by observation count for the given window and IATA.
+SELECT
+  o.id,
+  o.display_name,
+  o.observer_type,
+  COUNT(*) AS observation_count,
+  COALESCE((
+    SELECT po2.iata FROM packet_observations po2
+    WHERE po2.observer_id = o.id
+    ORDER BY po2.heard_at DESC LIMIT 1
+  ), '') AS iata
+FROM packet_observations po
+JOIN observers o ON o.id = po.observer_id
+WHERE po.heard_at > $1
+  AND ($2 = '' OR po.iata ILIKE $2)
+GROUP BY o.id
+ORDER BY observation_count DESC
+LIMIT $3;
+
 -- ============================================================
 -- REGIONS
 -- ============================================================

db/sqlc/models.go

@@ -50,11 +50,11 @@ type IataCode struct {
 }
 
 type MvHourlyIataStat struct {
-	Iata             string          `json:"iata"`
-	Hour             pgtype.Interval `json:"hour"`
-	ObservationCount int64           `json:"observation_count"`
-	UniquePackets    int64           `json:"unique_packets"`
-	ActiveObservers  int64           `json:"active_observers"`
+	Iata             string             `json:"iata"`
+	Hour             pgtype.Timestamptz `json:"hour"`
+	ObservationCount int64              `json:"observation_count"`
+	UniquePackets    int64              `json:"unique_packets"`
+	ActiveObservers  int64              `json:"active_observers"`
 }
 
 type MvTopNodesByIatum struct {

db/sqlc/queries.sql.go

@@ -150,14 +150,15 @@ func (q *Queries) GetChannelsByHash(ctx context.Context, arg GetChannelsByHashPa
 }
 
 const getHourlyStats = `-- name: GetHourlyStats :many
-SELECT iata, hour, observation_count, unique_packets, active_observers FROM mv_hourly_iata_stats
-WHERE ($1::char(3) IS NULL OR iata = $1)
+SELECT iata, hour, observation_count, unique_packets, active_observers
+FROM mv_hourly_iata_stats
+WHERE ($1 = '' OR iata ILIKE $1)
   AND hour >= NOW() - $2::interval
 ORDER BY iata, hour
 `
 
 type GetHourlyStatsParams struct {
-	Column1 string          `json:"column_1"`
+	Column1 interface{}     `json:"column_1"`
 	Column2 pgtype.Interval `json:"column_2"`
 }
 
@@ -552,7 +553,7 @@ SELECT
   COUNT(DISTINCT po.iata)         AS active_iatas
 FROM packet_observations po
 WHERE po.heard_at > NOW() - INTERVAL '24 hours'
-  AND ($1::char(3) IS NULL OR po.iata = $1)
+  AND ($1 = '' OR po.iata ILIKE $1)
 `
 
 type GetStatsOverviewRow struct {
@@ -565,7 +566,7 @@ type GetStatsOverviewRow struct {
 // ============================================================
 // STATS
 // ============================================================
-func (q *Queries) GetStatsOverview(ctx context.Context, dollar_1 string) (GetStatsOverviewRow, error) {
+func (q *Queries) GetStatsOverview(ctx context.Context, dollar_1 interface{}) (GetStatsOverviewRow, error) {
 	row := q.db.QueryRow(ctx, getStatsOverview, dollar_1)
 	var i GetStatsOverviewRow
 	err := row.Scan(
@@ -577,6 +578,110 @@ func (q *Queries) GetStatsOverview(ctx context.Context, dollar_1 string) (GetSta
 	return i, err
 }
 
+const getStatsPayloadBreakdown = `-- name: GetStatsPayloadBreakdown :many
+SELECT
+  p.payload_type,
+  COUNT(*) AS count
+FROM packet_observations po
+JOIN packets p ON p.packet_hash = po.packet_hash
+WHERE po.heard_at > $1
+  AND ($2 = '' OR po.iata ILIKE $2)
+GROUP BY p.payload_type
+ORDER BY count DESC
+`
+
+type GetStatsPayloadBreakdownParams struct {
+	HeardAt pgtype.Timestamptz `json:"heard_at"`
+	Column2 interface{}        `json:"column_2"`
+}
+
+type GetStatsPayloadBreakdownRow struct {
+	PayloadType int16 `json:"payload_type"`
+	Count       int64 `json:"count"`
+}
+
+// Returns observation counts grouped by payload type for the given window and IATA.
+func (q *Queries) GetStatsPayloadBreakdown(ctx context.Context, arg GetStatsPayloadBreakdownParams) ([]GetStatsPayloadBreakdownRow, error) {
+	rows, err := q.db.Query(ctx, getStatsPayloadBreakdown, arg.HeardAt, arg.Column2)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	items := []GetStatsPayloadBreakdownRow{}
+	for rows.Next() {
+		var i GetStatsPayloadBreakdownRow
+		if err := rows.Scan(&i.PayloadType, &i.Count); err != nil {
+			return nil, err
+		}
+		items = append(items, i)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return items, nil
+}
+
+const getStatsTopObservers = `-- name: GetStatsTopObservers :many
+SELECT
+  o.id,
+  o.display_name,
+  o.observer_type,
+  COUNT(*) AS observation_count,
+  COALESCE((
+    SELECT po2.iata FROM packet_observations po2
+    WHERE po2.observer_id = o.id
+    ORDER BY po2.heard_at DESC LIMIT 1
+  ), '') AS iata
+FROM packet_observations po
+JOIN observers o ON o.id = po.observer_id
+WHERE po.heard_at > $1
+  AND ($2 = '' OR po.iata ILIKE $2)
+GROUP BY o.id
+ORDER BY observation_count DESC
+LIMIT $3
+`
+
+type GetStatsTopObserversParams struct {
+	HeardAt pgtype.Timestamptz `json:"heard_at"`
+	Column2 interface{}        `json:"column_2"`
+	Limit   int32              `json:"limit"`
+}
+
+type GetStatsTopObserversRow struct {
+	ID               uuid.UUID   `json:"id"`
+	DisplayName      *string     `json:"display_name"`
+	ObserverType     *string     `json:"observer_type"`
+	ObservationCount int64       `json:"observation_count"`
+	Iata             interface{} `json:"iata"`
+}
+
+// Returns the top N observers by observation count for the given window and IATA.
+func (q *Queries) GetStatsTopObservers(ctx context.Context, arg GetStatsTopObserversParams) ([]GetStatsTopObserversRow, error) {
+	rows, err := q.db.Query(ctx, getStatsTopObservers, arg.HeardAt, arg.Column2, arg.Limit)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	items := []GetStatsTopObserversRow{}
+	for rows.Next() {
+		var i GetStatsTopObserversRow
+		if err := rows.Scan(
+			&i.ID,
+			&i.DisplayName,
+			&i.ObserverType,
+			&i.ObservationCount,
+			&i.Iata,
+		); err != nil {
+			return nil, err
+		}
+		items = append(items, i)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return items, nil
+}
+
 const getTopNodes = `-- name: GetTopNodes :many
 SELECT iata, node_id, name, node_type, observation_count, last_heard FROM mv_top_nodes_by_iata
 WHERE ($1::char(3) IS NULL OR iata = $1)

db/store.go

@@ -386,14 +386,14 @@ func (s *Store) UpsertRegionIATA(ctx context.Context, regionID int32, iata strin
 // cursor is last_seen epoch ms of the last item; pass 0 to start from the beginning.
 // Note: after sqlc generate, verify Column param names match generated types.
 func (s *Store) ListChannels(ctx context.Context, limit int32, hash []byte, iata string, cursor int64) (api.Page[api.ChannelSummary], error) {
-	var cursorTs pgtype.Timestamptz
+	var cursorTS pgtype.Timestamptz
 	if cursor > 0 {
-		cursorTs = pgtype.Timestamptz{Time: time.UnixMilli(cursor), Valid: true}
+		cursorTS = pgtype.Timestamptz{Time: time.UnixMilli(cursor), Valid: true}
 	}
 	rows, err := s.q.ListChannels(ctx, sqlc.ListChannelsParams{
 		Column1: hash,
 		Column2: iata,
-		Column3: cursorTs,
+		Column3: cursorTS,
 		Limit:   limit + 1,
 	})
 	if err != nil {
@@ -1048,3 +1048,119 @@ func (s *Store) GetPacket(ctx context.Context, packetHash []byte) (*api.Packet,
 	}
 	return p, nil
 }
+
+// GetStatsOverview returns top-line network figures for the last 24 hours.
+func (s *Store) GetStatsOverview(ctx context.Context, iata string) (*api.StatsOverview, error) {
+	row, err := s.q.GetStatsOverview(ctx, iata)
+	if err != nil {
+		return nil, err
+	}
+	return &api.StatsOverview{
+		TotalPackets:      row.TotalPackets,
+		TotalObservations: row.TotalObservations,
+		ActiveObservers:   row.ActiveObservers,
+		ActiveIATAs:       row.ActiveIatas,
+		WindowHours:       24,
+	}, nil
+}
+
+// GetStatsObservations returns hourly observation counts for charting.
+func (s *Store) GetStatsObservations(ctx context.Context, iata string, since time.Time) ([]api.ObservationPoint, error) {
+	if since.IsZero() {
+		since = time.Now().Add(-7 * 24 * time.Hour)
+	}
+	interval := time.Since(since)
+	rows, err := s.q.GetHourlyStats(ctx, sqlc.GetHourlyStatsParams{
+		Column1: iata,
+		Column2: pgtype.Interval{Microseconds: int64(interval.Hours()) * 3600 * 1e6, Valid: true},
+	})
+	if err != nil {
+		return nil, err
+	}
+	points := make([]api.ObservationPoint, 0, len(rows))
+	for _, v := range rows {
+		points = append(points, api.ObservationPoint{
+			Hour:             v.Hour.Time.UnixMilli(),
+			IATA:             v.Iata,
+			ObservationCount: v.ObservationCount,
+			UniquePackets:    v.UniquePackets,
+			ActiveObservers:  v.ActiveObservers,
+		})
+	}
+	return points, nil
+}
+
+func (s *Store) GetStatsPayloadBreakdown(ctx context.Context, iata string, since time.Time) ([]api.PayloadBreakdownItem, error) {
+	if since.IsZero() {
+		since = time.Now().Add(-24 * time.Hour)
+	}
+	rows, err := s.q.GetStatsPayloadBreakdown(ctx, sqlc.GetStatsPayloadBreakdownParams{
+		HeardAt: pgtype.Timestamptz{Time: since, Valid: true},
+		Column2: iata,
+	})
+	if err != nil {
+		return nil, err
+	}
+	items := make([]api.PayloadBreakdownItem, 0, len(rows))
+	for _, v := range rows {
+		items = append(items, api.PayloadBreakdownItem{
+			PayloadType:     v.PayloadType,
+			PayloadTypeName: api.PayloadTypeName(v.PayloadType),
+			Count:           v.Count,
+		})
+	}
+	return items, nil
+}
+
+func (s *Store) GetStatsTopNodes(ctx context.Context, iata string, limit int32) ([]api.TopNode, error) {
+	rows, err := s.q.GetTopNodes(ctx, sqlc.GetTopNodesParams{
+		Column1: iata,
+		Limit:   limit,
+	})
+	if err != nil {
+		return nil, err
+	}
+	items := make([]api.TopNode, 0, len(rows))
+	for _, v := range rows {
+		var count int64
+		if v.ObservationCount != nil {
+			count = *v.ObservationCount
+		}
+		items = append(items, api.TopNode{
+			NodeID:           v.NodeID,
+			NodeName:         v.Name,
+			NodeType:         v.NodeType,
+			NodeTypeName:     api.NodeTypeName(v.NodeType),
+			IATA:             v.Iata,
+			ObservationCount: count,
+			LastHeard:        v.LastHeard.Time.UnixMilli(),
+		})
+	}
+	return items, nil
+}
+
+func (s *Store) GetStatsTopObservers(ctx context.Context, iata string, since time.Time, limit int32) ([]api.TopObserver, error) {
+	if since.IsZero() {
+		since = time.Now().Add(-24 * time.Hour)
+	}
+	rows, err := s.q.GetStatsTopObservers(ctx, sqlc.GetStatsTopObserversParams{
+		HeardAt: pgtype.Timestamptz{Time: since, Valid: true},
+		Column2: iata,
+		Limit:   limit,
+	})
+	if err != nil {
+		return nil, err
+	}
+	items := make([]api.TopObserver, 0, len(rows))
+	for _, v := range rows {
+		iata, _ := v.Iata.(string)
+		items = append(items, api.TopObserver{
+			ObserverID:       v.ID,
+			DisplayName:      v.DisplayName,
+			ObserverType:     v.ObserverType,
+			IATA:             iata,
+			ObservationCount: v.ObservationCount,
+		})
+	}
+	return items, nil
+}

internal/api/handlers/stats.go

@@ -1,99 +1,157 @@
 package handlers
 
 import (
+	"log"
 	"net/http"
+	"strconv"
+	"time"
 
+	"github.com/MeshCore-Tower/tower-server/internal/api"
 	"github.com/go-chi/chi/v5"
 )
 
 // StatsRouter mounts all /stats routes onto a subrouter.
 //
-// GET  /stats/overview                       → GetStatsOverview
-// GET  /stats/observations                   → GetStatsObservations
-// GET  /stats/payloadBreakdown               → GetStatsPayloadBreakdown
-// GET  /stats/topNodes                       → GetStatsTopNodes
-// GET  /stats/topObservers                   → GetStatsTopObservers
+// GET  /stats/overview         → GetStatsOverview
+// GET  /stats/observations     → GetStatsObservations
+// GET  /stats/payload-breakdown → GetStatsPayloadBreakdown
+// GET  /stats/top-nodes         → GetStatsTopNodes
+// GET  /stats/top-observers     → GetStatsTopObservers
 //
-// All endpoints accept either iata= (one or comma-separated) or regionId=
-// (expands to all IATAs in that super-region via region_iatas).
-func StatsRouter() http.Handler {
+// All endpoints accept an optional iata= filter (case-insensitive).
+// regionId= expansion and comma-separated IATAs are not yet implemented.
+func StatsRouter(reader api.Reader) http.Handler {
 	r := chi.NewRouter()
 
-	r.Get("/overview", GetStatsOverview)
-	r.Get("/observations", GetStatsObservations)
-	r.Get("/payloadBreakdown", GetStatsPayloadBreakdown)
-	r.Get("/topNodes", GetStatsTopNodes)
-	r.Get("/topObservers", GetStatsTopObservers)
+	// GET /api/v1/stats/overview
+	//
+	// Query params (all optional):
+	//
+	//	iata=<code>   filter to a single IATA (case-insensitive)
+	r.Get("/overview", func(w http.ResponseWriter, r *http.Request) {
+		iata := r.URL.Query().Get("iata")
+		overview, err := reader.GetStatsOverview(r.Context(), iata)
+		if err != nil {
+			log.Printf("api: GetStatsOverview failed: %v", err)
+			respondError(w, http.StatusInternalServerError, "internal server error")
+			return
+		}
+		respond(w, http.StatusOK, overview)
+	})
+
+	// GET /api/v1/stats/observations
+	//
+	// Query params (all optional):
+	//
+	//	iata=<code>        filter to a single IATA (case-insensitive)
+	//	since=<epoch ms>   start of window; defaults to 7 days ago
+	r.Get("/observations", func(w http.ResponseWriter, r *http.Request) {
+		iata := r.URL.Query().Get("iata")
+		var since time.Time
+		if p := r.URL.Query().Get("since"); p != "" {
+			ms, err := strconv.ParseInt(p, 10, 64)
+			if err != nil {
+				respondError(w, http.StatusBadRequest, "since must be epoch milliseconds")
+				return
+			}
+			since = time.UnixMilli(ms)
+		}
+		points, err := reader.GetStatsObservations(r.Context(), iata, since)
+		if err != nil {
+			log.Printf("api: GetStatsObservations failed: %v", err)
+			respondError(w, http.StatusInternalServerError, "internal server error")
+			return
+		}
+		respond(w, http.StatusOK, points)
+	})
+
+	// GET /api/v1/stats/payload-breakdown
+	//
+	// Query params (all optional):
+	//
+	//	iata=<code>        filter to a single IATA (case-insensitive)
+	//	since=<epoch ms>   start of window; defaults to last 24h
+	r.Get("/payload-breakdown", func(w http.ResponseWriter, r *http.Request) {
+		iata := r.URL.Query().Get("iata")
+		var since time.Time
+		if p := r.URL.Query().Get("since"); p != "" {
+			ms, err := strconv.ParseInt(p, 10, 64)
+			if err != nil {
+				respondError(w, http.StatusBadRequest, "since must be epoch milliseconds")
+				return
+			}
+			since = time.UnixMilli(ms)
+		}
+		breakdown, err := reader.GetStatsPayloadBreakdown(r.Context(), iata, since)
+		if err != nil {
+			log.Printf("api: GetStatsPayloadBreakdown failed: %v", err)
+			respondError(w, http.StatusInternalServerError, "internal server error")
+			return
+		}
+		respond(w, http.StatusOK, breakdown)
+	})
+
+	// GET /api/v1/stats/top-nodes
+	//
+	// Query params (all optional):
+	//
+	//	iata=<code>   filter to a single IATA (case-insensitive)
+	//	limit=10
+	r.Get("/top-nodes", func(w http.ResponseWriter, r *http.Request) {
+		iata := r.URL.Query().Get("iata")
+		var limit int32 = 10
+		if p := r.URL.Query().Get("limit"); p != "" {
+			l, err := strconv.ParseInt(p, 10, 32)
+			if err != nil {
+				respondError(w, http.StatusBadRequest, "limit must be an integer")
+				return
+			}
+			limit = int32(l)
+		}
+		nodes, err := reader.GetStatsTopNodes(r.Context(), iata, limit)
+		if err != nil {
+			log.Printf("api: GetStatsTopNodes failed: %v", err)
+			respondError(w, http.StatusInternalServerError, "internal server error")
+			return
+		}
+		respond(w, http.StatusOK, nodes)
+	})
+
+	// GET /api/v1/stats/top-observers
+	//
+	// Query params (all optional):
+	//
+	//	iata=<code>        filter to a single IATA (case-insensitive)
+	//	since=<epoch ms>   start of window; defaults to last 24h
+	//	limit=10
+	r.Get("/top-observers", func(w http.ResponseWriter, r *http.Request) {
+		iata := r.URL.Query().Get("iata")
+		var since time.Time
+		if p := r.URL.Query().Get("since"); p != "" {
+			ms, err := strconv.ParseInt(p, 10, 64)
+			if err != nil {
+				respondError(w, http.StatusBadRequest, "since must be epoch milliseconds")
+				return
+			}
+			since = time.UnixMilli(ms)
+		}
+		var limit int32 = 10
+		if p := r.URL.Query().Get("limit"); p != "" {
+			l, err := strconv.ParseInt(p, 10, 32)
+			if err != nil {
+				respondError(w, http.StatusBadRequest, "limit must be an integer")
+				return
+			}
+			limit = int32(l)
+		}
+		observers, err := reader.GetStatsTopObservers(r.Context(), iata, since, limit)
+		if err != nil {
+			log.Printf("api: GetStatsTopObservers failed: %v", err)
+			respondError(w, http.StatusInternalServerError, "internal server error")
+			return
+		}
+		respond(w, http.StatusOK, observers)
+	})
 
 	return r
 }
-
-// GetStatsOverview handles GET /api/v1/stats/overview
-//
-// Query params (all optional):
-//   iata=YOW          (one or comma-separated)
-//   regionId=<id>
-//
-// Returns top-line figures: total packets and observations last 24h,
-// active observers, active IATAs, unique nodes seen.
-func GetStatsOverview(w http.ResponseWriter, r *http.Request) {
-	// TODO: query mv_hourly_iata_stats + live aggregates, write JSON response.
-	w.WriteHeader(http.StatusNotImplemented)
-}
-
-// GetStatsObservations handles GET /api/v1/stats/observations
-//
-// Query params (all optional):
-//   iata=YOW
-//   regionId=<id>
-//   range=24h          (duration string: 24h, 7d, 30d)
-//   interval=1h        (bucket size: 5m, 1h, 1d)
-//
-// Returns a time series of observation counts bucketed by interval,
-// suitable for charting.
-func GetStatsObservations(w http.ResponseWriter, r *http.Request) {
-	// TODO: query mv_hourly_iata_stats, group by bucket, write JSON response.
-	w.WriteHeader(http.StatusNotImplemented)
-}
-
-// GetStatsPayloadBreakdown handles GET /api/v1/stats/payloadBreakdown
-//
-// Query params (all optional):
-//   iata=YOW
-//   regionId=<id>
-//   range=24h
-//
-// Returns observation counts grouped by payload_type for the given window.
-func GetStatsPayloadBreakdown(w http.ResponseWriter, r *http.Request) {
-	// TODO: query packets + observations filtered by time/iata, group by payload_type.
-	w.WriteHeader(http.StatusNotImplemented)
-}
-
-// GetStatsTopNodes handles GET /api/v1/stats/topNodes
-//
-// Query params (all optional):
-//   iata=YOW
-//   regionId=<id>
-//   range=24h
-//   limit=10
-//
-// Returns the top N nodes by observation contribution count using
-// mv_top_nodes_by_iata.
-func GetStatsTopNodes(w http.ResponseWriter, r *http.Request) {
-	// TODO: query mv_top_nodes_by_iata, apply filters, write JSON response.
-	w.WriteHeader(http.StatusNotImplemented)
-}
-
-// GetStatsTopObservers handles GET /api/v1/stats/topObservers
-//
-// Query params (all optional):
-//   iata=YOW
-//   regionId=<id>
-//   range=24h
-//   limit=10
-//
-// Returns the top N observers by observation count for the given window.
-func GetStatsTopObservers(w http.ResponseWriter, r *http.Request) {
-	// TODO: query packet_observations grouped by observer_id, apply filters.
-	w.WriteHeader(http.StatusNotImplemented)
-}

internal/api/reader.go

@@ -9,6 +9,51 @@ import (
 	"github.com/google/uuid"
 )
 
+// StatsOverview is the top-level network summary for the overview endpoint.
+type StatsOverview struct {
+	TotalPackets      int64 `json:"totalPackets"`
+	TotalObservations int64 `json:"totalObservations"`
+	ActiveObservers   int64 `json:"activeObservers"`
+	ActiveIATAs       int64 `json:"activeIatas"`
+	WindowHours       int   `json:"windowHours"` // always 24 for now
+}
+
+// ObservationPoint is a single time-bucketed observation count.
+type ObservationPoint struct {
+	Hour             int64  `json:"hour"` // epoch ms, start of bucket
+	IATA             string `json:"iata"`
+	ObservationCount int64  `json:"observationCount"`
+	UniquePackets    int64  `json:"uniquePackets"`
+	ActiveObservers  int64  `json:"activeObservers"`
+}
+
+// PayloadBreakdownItem is a single payload type with its observation count.
+type PayloadBreakdownItem struct {
+	PayloadType     int16  `json:"payloadType"`
+	PayloadTypeName string `json:"payloadTypeName"`
+	Count           int64  `json:"count"`
+}
+
+// TopNode is a node ranked by observation count.
+type TopNode struct {
+	NodeID           uuid.UUID `json:"nodeId"`
+	NodeName         *string   `json:"nodeName,omitempty"`
+	NodeType         int16     `json:"nodeType"`
+	NodeTypeName     string    `json:"nodeTypeName"`
+	IATA             string    `json:"iata"`
+	ObservationCount int64     `json:"observationCount"`
+	LastHeard        int64     `json:"lastHeard"` // epoch ms
+}
+
+// TopObserver is an observer ranked by observation count.
+type TopObserver struct {
+	ObserverID       uuid.UUID `json:"observerId"`
+	DisplayName      *string   `json:"displayName,omitempty"`
+	ObserverType     *string   `json:"observerType,omitempty"`
+	IATA             string    `json:"iata"`
+	ObservationCount int64     `json:"observationCount"`
+}
+
 // PacketLatestObserver is the most recent observer summary rolled into a packet list item.
 type PacketLatestObserver struct {
 	ID          uuid.UUID `json:"id"`
@@ -339,4 +384,22 @@ type Reader interface {
 	// GetPacket returns full packet detail including all observations with radio settings.
 	// Returns nil, pgx.ErrNoRows if not found.
 	GetPacket(ctx context.Context, packetHash []byte) (*Packet, error)
+	// GetStatsOverview returns top-line network figures for the last 24 hours.
+	// Pass empty string iata to return stats across all IATAs.
+	GetStatsOverview(ctx context.Context, iata string) (*StatsOverview, error)
+	// GetStatsObservations returns hourly observation counts for charting.
+	// Pass empty string iata to return stats across all IATAs.
+	// since defines the start of the window; pass zero time for default (last 7 days).
+	GetStatsObservations(ctx context.Context, iata string, since time.Time) ([]ObservationPoint, error)
+	// GetStatsPayloadBreakdown returns observation counts grouped by payload type.
+	// Pass empty string iata to return stats across all IATAs.
+	// since defines the start of the window; pass zero time for default (last 24h).
+	GetStatsPayloadBreakdown(ctx context.Context, iata string, since time.Time) ([]PayloadBreakdownItem, error)
+	// GetStatsTopNodes returns the top N nodes by observation count.
+	// Pass empty string iata to return stats across all IATAs.
+	GetStatsTopNodes(ctx context.Context, iata string, limit int32) ([]TopNode, error)
+	// GetStatsTopObservers returns the top N observers by observation count.
+	// Pass empty string iata to return stats across all IATAs.
+	// since defines the start of the window; pass zero time for default (last 24h).
+	GetStatsTopObservers(ctx context.Context, iata string, since time.Time, limit int32) ([]TopObserver, error)
 }

internal/api/router/router.go

@@ -1,3 +1,7 @@
+// Package router wires all HTTP routes onto the Chi router and injects
+// dependencies (hub, reader, ingest workers) into the handler closures.
+// All routes are mounted under /api/v1 with public and private groups
+// stubbed for future auth middleware.
 package router
 
 import (
@@ -57,7 +61,7 @@ func New(h *hub.Hub, reader api.Reader, workers []*ingest.Worker) http.Handler {
 			r.Mount("/messages", handlers.MessagesRouter(reader))
 			r.Mount("/iatas", handlers.IATAsRouter(reader))
 			r.Mount("/regions", handlers.RegionsRouter(reader))
-			r.Mount("/stats", handlers.StatsRouter())
+			r.Mount("/stats", handlers.StatsRouter(reader))
 		})
 
 		// Private group — auth middleware applied.