track payload_type on observations and serve the breakdown from a matview

db/migrations/015_observations_payload_type.sql

@@ -0,0 +1,17 @@
+-- Copy payload_type onto the observation (like source_broker) so the breakdown
+-- drops its join back to packets, which was overrunning /dev/shm and 500ing.
+
+ALTER TABLE packet_observations ADD COLUMN payload_type SMALLINT;
+
+-- Backfill the last 7 days only (all the breakdown reads); parallelism off so
+-- the join spills to disk, not /dev/shm.
+SET max_parallel_workers_per_gather = 0;
+
+UPDATE packet_observations po
+SET payload_type = p.payload_type
+FROM packets p
+WHERE p.packet_hash = po.packet_hash
+  AND po.heard_at > NOW() - INTERVAL '7 days'
+  AND po.payload_type IS NULL;
+
+RESET max_parallel_workers_per_gather;

db/migrations/016_mv_payload_breakdown.sql

@@ -0,0 +1,15 @@
+-- Precomputed payload breakdown so the request reads a small table instead of
+-- scanning a week of observations.
+
+CREATE MATERIALIZED VIEW mv_payload_breakdown_by_iata AS
+SELECT
+  iata,
+  payload_type,
+  COUNT(*) AS count
+FROM packet_observations
+WHERE heard_at > NOW() - INTERVAL '7 days'
+  AND payload_type IS NOT NULL
+GROUP BY iata, payload_type;
+
+CREATE UNIQUE INDEX idx_mv_payload_breakdown
+  ON mv_payload_breakdown_by_iata(iata, payload_type);

db/packets.go

@@ -505,6 +505,7 @@ func (s *Store) InsertObservation(ctx context.Context, o ingest.InsertObservatio
 		BandwidthKhz:      &o.BandwidthKHz,
 		CodingRate:        &o.CodingRate,
 		SourceBroker:      &o.SourceBroker,
+		PayloadType:       &o.PayloadType,
 	}
 	row, err := s.q.InsertObservation(ctx, params)
 	if errors.Is(err, pgx.ErrNoRows) {

db/queries/queries.sql

@@ -523,9 +523,10 @@ INSERT INTO packet_observations (
   spread_factor,
   bandwidth_khz,
   coding_rate,
-  source_broker
+  source_broker,
+  payload_type
 ) VALUES (
-  $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16
+  $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17
 )
 ON CONFLICT (packet_hash, observer_id) DO NOTHING
 RETURNING *;
@@ -821,15 +822,13 @@ ORDER BY observation_count DESC
 LIMIT $2;
 
 -- name: GetStatsPayloadBreakdown :many
--- Returns observation counts grouped by payload type for the given window and IATA.
+-- Payload-type counts for the IATA, from the precomputed view.
 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 (COALESCE(cardinality($2::bpchar[]), 0) = 0 OR po.iata = ANY($2::bpchar[]))
-GROUP BY p.payload_type
+  payload_type,
+  SUM(count)::bigint AS count
+FROM mv_payload_breakdown_by_iata
+WHERE (COALESCE(cardinality($1::bpchar[]), 0) = 0 OR iata = ANY($1::bpchar[]))
+GROUP BY payload_type
 ORDER BY count DESC;
 
 -- name: GetStatsNodeTypes :many
@@ -1141,6 +1140,9 @@ REFRESH MATERIALIZED VIEW CONCURRENTLY mv_hourly_iata_stats;
 -- name: RefreshTopNodes :exec
 REFRESH MATERIALIZED VIEW CONCURRENTLY mv_top_nodes_by_iata;
 
+-- name: RefreshPayloadBreakdown :exec
+REFRESH MATERIALIZED VIEW CONCURRENTLY mv_payload_breakdown_by_iata;
+
 -- name: RefreshRadioPresets :exec
 REFRESH MATERIALIZED VIEW CONCURRENTLY mv_radio_presets;
 

db/sqlc/mock/querier.go

@@ -550,18 +550,18 @@ func (mr *MockQuerierMockRecorder) GetStatsOverview(ctx, dollar_1 any) *gomock.C
 }
 
 // GetStatsPayloadBreakdown mocks base method.
-func (m *MockQuerier) GetStatsPayloadBreakdown(ctx context.Context, arg db.GetStatsPayloadBreakdownParams) ([]db.GetStatsPayloadBreakdownRow, error) {
+func (m *MockQuerier) GetStatsPayloadBreakdown(ctx context.Context, dollar_1 []string) ([]db.GetStatsPayloadBreakdownRow, error) {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "GetStatsPayloadBreakdown", ctx, arg)
+	ret := m.ctrl.Call(m, "GetStatsPayloadBreakdown", ctx, dollar_1)
 	ret0, _ := ret[0].([]db.GetStatsPayloadBreakdownRow)
 	ret1, _ := ret[1].(error)
 	return ret0, ret1
 }
 
 // GetStatsPayloadBreakdown indicates an expected call of GetStatsPayloadBreakdown.
-func (mr *MockQuerierMockRecorder) GetStatsPayloadBreakdown(ctx, arg any) *gomock.Call {
+func (mr *MockQuerierMockRecorder) GetStatsPayloadBreakdown(ctx, dollar_1 any) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetStatsPayloadBreakdown", reflect.TypeOf((*MockQuerier)(nil).GetStatsPayloadBreakdown), ctx, arg)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetStatsPayloadBreakdown", reflect.TypeOf((*MockQuerier)(nil).GetStatsPayloadBreakdown), ctx, dollar_1)
 }
 
 // GetStatsTopAdvertisers mocks base method.
@@ -995,6 +995,20 @@ func (mr *MockQuerierMockRecorder) RefreshHourlyStats(ctx any) *gomock.Call {
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RefreshHourlyStats", reflect.TypeOf((*MockQuerier)(nil).RefreshHourlyStats), ctx)
 }
 
+// RefreshPayloadBreakdown mocks base method.
+func (m *MockQuerier) RefreshPayloadBreakdown(ctx context.Context) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "RefreshPayloadBreakdown", ctx)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// RefreshPayloadBreakdown indicates an expected call of RefreshPayloadBreakdown.
+func (mr *MockQuerierMockRecorder) RefreshPayloadBreakdown(ctx any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RefreshPayloadBreakdown", reflect.TypeOf((*MockQuerier)(nil).RefreshPayloadBreakdown), ctx)
+}
+
 // RefreshRadioPresets mocks base method.
 func (m *MockQuerier) RefreshRadioPresets(ctx context.Context) error {
 	m.ctrl.T.Helper()

db/sqlc/models.go

@@ -74,6 +74,12 @@ type MvHourlyIataStat struct {
 	ActiveObservers  int64              `json:"active_observers"`
 }
 
+type MvPayloadBreakdownByIatum struct {
+	Iata        string `json:"iata"`
+	PayloadType *int16 `json:"payload_type"`
+	Count       int64  `json:"count"`
+}
+
 type MvRadioPreset struct {
 	Preset     string `json:"preset"`
 	Iata       string `json:"iata"`
@@ -250,6 +256,7 @@ type PacketObservation struct {
 	BandwidthKhz      *float32           `json:"bandwidth_khz"`
 	CodingRate        *int16             `json:"coding_rate"`
 	SourceBroker      *string            `json:"source_broker"`
+	PayloadType       *int16             `json:"payload_type"`
 }
 
 type Region struct {

db/sqlc/querier.go

@@ -60,8 +60,8 @@ type Querier interface {
 	// STATS
 	// ============================================================
 	GetStatsOverview(ctx context.Context, dollar_1 []string) (GetStatsOverviewRow, error)
-	// Returns observation counts grouped by payload type for the given window and IATA.
-	GetStatsPayloadBreakdown(ctx context.Context, arg GetStatsPayloadBreakdownParams) ([]GetStatsPayloadBreakdownRow, error)
+	// Payload-type counts for the IATA, from the precomputed view.
+	GetStatsPayloadBreakdown(ctx context.Context, dollar_1 []string) ([]GetStatsPayloadBreakdownRow, error)
 	// Returns the top N nodes by distinct ADVERT packet count for the given window and IATA.
 	// COUNT(DISTINCT p.packet_hash) rather than COUNT(*): the same advert broadcast is commonly
 	// heard by more than one observer, and each hearing is its own packet_observations row --
@@ -156,6 +156,7 @@ type Querier interface {
 	// that IATA, or where any hop's prefix_4 is now ambiguous (matches >1 node).
 	ReconfirmRoutes(ctx context.Context) error
 	RefreshHourlyStats(ctx context.Context) error
+	RefreshPayloadBreakdown(ctx context.Context) error
 	RefreshRadioPresets(ctx context.Context) error
 	RefreshTopNodes(ctx context.Context) error
 	// ============================================================

db/sqlc/queries.sql.go

@@ -1197,29 +1197,22 @@ func (q *Queries) GetStatsOverview(ctx context.Context, dollar_1 []string) (GetS
 
 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 (COALESCE(cardinality($2::bpchar[]), 0) = 0 OR po.iata = ANY($2::bpchar[]))
-GROUP BY p.payload_type
+  payload_type,
+  SUM(count)::bigint AS count
+FROM mv_payload_breakdown_by_iata
+WHERE (COALESCE(cardinality($1::bpchar[]), 0) = 0 OR iata = ANY($1::bpchar[]))
+GROUP BY payload_type
 ORDER BY count DESC
 `
 
-type GetStatsPayloadBreakdownParams struct {
-	HeardAt pgtype.Timestamptz `json:"heard_at"`
-	Column2 []string           `json:"column_2"`
-}
-
 type GetStatsPayloadBreakdownRow struct {
-	PayloadType int16 `json:"payload_type"`
-	Count       int64 `json:"count"`
+	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)
+// Payload-type counts for the IATA, from the precomputed view.
+func (q *Queries) GetStatsPayloadBreakdown(ctx context.Context, dollar_1 []string) ([]GetStatsPayloadBreakdownRow, error) {
+	rows, err := q.db.Query(ctx, getStatsPayloadBreakdown, dollar_1)
 	if err != nil {
 		return nil, err
 	}
@@ -1551,12 +1544,13 @@ INSERT INTO packet_observations (
   spread_factor,
   bandwidth_khz,
   coding_rate,
-  source_broker
+  source_broker,
+  payload_type
 ) VALUES (
-  $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16
+  $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17
 )
 ON CONFLICT (packet_hash, observer_id) DO NOTHING
-RETURNING id, packet_hash, observer_id, iata, heard_at, path_length_byte, hash_size, hop_count, path_bytes, rssi, snr, propagation_time_ms, radio_freq_mhz, spread_factor, bandwidth_khz, coding_rate, source_broker
+RETURNING id, packet_hash, observer_id, iata, heard_at, path_length_byte, hash_size, hop_count, path_bytes, rssi, snr, propagation_time_ms, radio_freq_mhz, spread_factor, bandwidth_khz, coding_rate, source_broker, payload_type
 `
 
 type InsertObservationParams struct {
@@ -1576,6 +1570,7 @@ type InsertObservationParams struct {
 	BandwidthKhz      *float32           `json:"bandwidth_khz"`
 	CodingRate        *int16             `json:"coding_rate"`
 	SourceBroker      *string            `json:"source_broker"`
+	PayloadType       *int16             `json:"payload_type"`
 }
 
 // ============================================================
@@ -1599,6 +1594,7 @@ func (q *Queries) InsertObservation(ctx context.Context, arg InsertObservationPa
 		arg.BandwidthKhz,
 		arg.CodingRate,
 		arg.SourceBroker,
+		arg.PayloadType,
 	)
 	var i PacketObservation
 	err := row.Scan(
@@ -1619,6 +1615,7 @@ func (q *Queries) InsertObservation(ctx context.Context, arg InsertObservationPa
 		&i.BandwidthKhz,
 		&i.CodingRate,
 		&i.SourceBroker,
+		&i.PayloadType,
 	)
 	return i, err
 }
@@ -2293,7 +2290,7 @@ func (q *Queries) ListNodes(ctx context.Context, arg ListNodesParams) ([]ListNod
 }
 
 const listObservationsForPacket = `-- name: ListObservationsForPacket :many
-SELECT po.id, po.packet_hash, po.observer_id, po.iata, po.heard_at, po.path_length_byte, po.hash_size, po.hop_count, po.path_bytes, po.rssi, po.snr, po.propagation_time_ms, po.radio_freq_mhz, po.spread_factor, po.bandwidth_khz, po.coding_rate, po.source_broker, o.display_name AS observer_name
+SELECT po.id, po.packet_hash, po.observer_id, po.iata, po.heard_at, po.path_length_byte, po.hash_size, po.hop_count, po.path_bytes, po.rssi, po.snr, po.propagation_time_ms, po.radio_freq_mhz, po.spread_factor, po.bandwidth_khz, po.coding_rate, po.source_broker, po.payload_type, o.display_name AS observer_name
 FROM packet_observations po
 LEFT JOIN observers o ON o.id = po.observer_id
 WHERE po.packet_hash = $1
@@ -2318,6 +2315,7 @@ type ListObservationsForPacketRow struct {
 	BandwidthKhz      *float32           `json:"bandwidth_khz"`
 	CodingRate        *int16             `json:"coding_rate"`
 	SourceBroker      *string            `json:"source_broker"`
+	PayloadType       *int16             `json:"payload_type"`
 	ObserverName      *string            `json:"observer_name"`
 }
 
@@ -2348,6 +2346,7 @@ func (q *Queries) ListObservationsForPacket(ctx context.Context, packetHash []by
 			&i.BandwidthKhz,
 			&i.CodingRate,
 			&i.SourceBroker,
+			&i.PayloadType,
 			&i.ObserverName,
 		); err != nil {
 			return nil, err
@@ -3073,6 +3072,15 @@ func (q *Queries) RefreshHourlyStats(ctx context.Context) error {
 	return err
 }
 
+const refreshPayloadBreakdown = `-- name: RefreshPayloadBreakdown :exec
+REFRESH MATERIALIZED VIEW CONCURRENTLY mv_payload_breakdown_by_iata
+`
+
+func (q *Queries) RefreshPayloadBreakdown(ctx context.Context) error {
+	_, err := q.db.Exec(ctx, refreshPayloadBreakdown)
+	return err
+}
+
 const refreshRadioPresets = `-- name: RefreshRadioPresets :exec
 REFRESH MATERIALIZED VIEW CONCURRENTLY mv_radio_presets
 `

db/stats.go

@@ -51,22 +51,20 @@ func (s *Store) GetStatsObservations(ctx context.Context, iatas []string, since
 	return points, nil
 }
 
-func (s *Store) GetStatsPayloadBreakdown(ctx context.Context, iatas []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: iatas,
-	})
+// since is unused: the view is a rolling 7-day window refreshed on a timer.
+func (s *Store) GetStatsPayloadBreakdown(ctx context.Context, iatas []string, _ time.Time) ([]api.PayloadBreakdownItem, error) {
+	rows, err := s.q.GetStatsPayloadBreakdown(ctx, iatas)
 	if err != nil {
 		return nil, err
 	}
 	items := make([]api.PayloadBreakdownItem, 0, len(rows))
 	for _, v := range rows {
+		if v.PayloadType == nil {
+			continue
+		}
 		items = append(items, api.PayloadBreakdownItem{
-			PayloadType:     v.PayloadType,
-			PayloadTypeName: api.PayloadTypeName(v.PayloadType),
+			PayloadType:     *v.PayloadType,
+			PayloadTypeName: api.PayloadTypeName(*v.PayloadType),
 			Count:           v.Count,
 		})
 	}
@@ -244,6 +242,10 @@ func (s *Store) RefreshTopNodes(ctx context.Context) error {
 	return s.q.RefreshTopNodes(ctx)
 }
 
+func (s *Store) RefreshPayloadBreakdown(ctx context.Context) error {
+	return s.q.RefreshPayloadBreakdown(ctx)
+}
+
 func (s *Store) RefreshRadioPresets(ctx context.Context) error {
 	return s.q.RefreshRadioPresets(ctx)
 }

internal/background/tasks.go

@@ -24,6 +24,9 @@ func ViewRefreshTask(store *db.Store, interval time.Duration) Task {
 			if err := store.RefreshTopNodes(ctx); err != nil {
 				log.Printf("background[view_refresh]: top nodes: %v", err)
 			}
+			if err := store.RefreshPayloadBreakdown(ctx); err != nil {
+				log.Printf("background[view_refresh]: payload breakdown: %v", err)
+			}
 			if err := store.RefreshRadioPresets(ctx); err != nil {
 				log.Printf("background[view_refresh]: radio presets: %v", err)
 			}

internal/ingest/packet.go

@@ -54,6 +54,7 @@ type InsertObservationParams struct {
 	BandwidthKHz      float32
 	CodingRate        int16
 	SourceBroker      string
+	PayloadType       int16
 }
 
 // RadioSettings holds the radio configuration for an observer, populated from
@@ -770,6 +771,7 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [
 		BandwidthKHz:      radio.BWKHz,
 		CodingRate:        radio.CR,
 		SourceBroker:      w.cfg.BrokerName,
+		PayloadType:       int16(packet.PayloadType()),
 	}
 	inserted, err := w.db.InsertObservation(ctx, oParams)
 	if err != nil {