perf: per-IATA activity tables + analytics query/matview tuning

db/channels.go

@@ -47,16 +47,28 @@ func (s *Store) UpsertChannelHashOnly(ctx context.Context, channelHash []byte) (
 	return int(rowID), nil
 }
 
-func (s *Store) ListChannels(ctx context.Context, limit int32, hash []byte, iata string, cursor int64) (api.Page[api.ChannelSummary], error) {
+func (s *Store) UpsertChannelIATA(ctx context.Context, channelHash []byte, iata string, heardAt time.Time) error {
+	return s.q.UpsertChannelIATA(ctx, sqlc.UpsertChannelIATAParams{
+		ChannelHash: channelHash,
+		Iata:        iata,
+		LastHeard:   pgtype.Timestamptz{Time: heardAt, Valid: true},
+	})
+}
+
+func (s *Store) DeleteOldChannelIATAs(ctx context.Context, cutoff time.Time) error {
+	return s.q.DeleteOldChannelIATAs(ctx, pgtype.Timestamptz{Time: cutoff, Valid: true})
+}
+
+func (s *Store) ListChannels(ctx context.Context, limit int32, hash []byte, iatas []string, cursor int64) (api.Page[api.ChannelSummary], error) {
 	var cursorTS pgtype.Timestamptz
 	if cursor > 0 {
 		cursorTS = pgtype.Timestamptz{Time: time.UnixMilli(cursor), Valid: true}
 	}
 	rows, err := s.q.ListChannels(ctx, sqlc.ListChannelsParams{
-		Column1: hash,
-		Column2: iata,
-		Column3: cursorTS,
-		Limit:   limit + 1,
+		ChannelHash: hash,
+		Iatas:       iatas,
+		CursorTs:    cursorTS,
+		PageLimit:   limit + 1,
 	})
 	if err != nil {
 		return api.Page[api.ChannelSummary]{}, err

db/channels_test.go

@@ -21,15 +21,15 @@ func TestListChannels_Empty(t *testing.T) {
 
 	mock.EXPECT().
 		ListChannels(gomock.Any(), sqlc.ListChannelsParams{
-			Column1: nil,
-			Column2: "",
-			Column3: pgtype.Timestamptz{},
-			Limit:   11,
+			ChannelHash: nil,
+			Iatas:       nil,
+			CursorTs:    pgtype.Timestamptz{},
+			PageLimit:   11,
 		}).
 		Return([]sqlc.Channel{}, nil)
 
 	store := &Store{q: mock}
-	page, err := store.ListChannels(context.Background(), 10, nil, "", 0)
+	page, err := store.ListChannels(context.Background(), 10, nil, nil, 0)
 	if err != nil {
 		t.Fatalf("unexpected error: %v", err)
 	}
@@ -63,15 +63,15 @@ func TestListChannels_Pagination(t *testing.T) {
 
 	mock.EXPECT().
 		ListChannels(gomock.Any(), sqlc.ListChannelsParams{
-			Column1: nil,
-			Column2: "",
-			Column3: pgtype.Timestamptz{},
-			Limit:   3, // limit+1
+			ChannelHash: nil,
+			Iatas:       nil,
+			CursorTs:    pgtype.Timestamptz{},
+			PageLimit:   3, // limit+1
 		}).
 		Return(rows, nil)
 
 	store := &Store{q: mock}
-	page, err := store.ListChannels(context.Background(), 2, nil, "", 0)
+	page, err := store.ListChannels(context.Background(), 2, nil, nil, 0)
 	if err != nil {
 		t.Fatalf("unexpected error: %v", err)
 	}
@@ -95,12 +95,32 @@ func TestListChannels_DBError(t *testing.T) {
 		Return(nil, errors.New("db error"))
 
 	store := &Store{q: mock}
-	_, err := store.ListChannels(context.Background(), 10, nil, "", 0)
+	_, err := store.ListChannels(context.Background(), 10, nil, nil, 0)
 	if err == nil {
 		t.Fatal("expected error, got nil")
 	}
 }
 
+func TestListChannels_IATAFilter(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	mock := mockdb.NewMockQuerier(ctrl)
+
+	mock.EXPECT().
+		ListChannels(gomock.Any(), sqlc.ListChannelsParams{
+			ChannelHash: nil,
+			Iatas:       []string{"YOW", "YYZ"},
+			CursorTs:    pgtype.Timestamptz{},
+			PageLimit:   11,
+		}).
+		Return([]sqlc.Channel{}, nil)
+
+	store := &Store{q: mock}
+	_, err := store.ListChannels(context.Background(), 10, nil, []string{"YOW", "YYZ"}, 0)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+}
+
 func TestGetChannel_Basic(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	mock := mockdb.NewMockQuerier(ctrl)

db/migrations/011_channel_iatas.sql

@@ -0,0 +1,23 @@
+-- Per-IATA channel activity so the IATA filter skips the ~7s EXISTS over packets.
+-- Keyed by raw hash (channels can share one), so no FK to channels.
+
+CREATE TABLE channel_iatas (
+  channel_hash BYTEA NOT NULL,
+  iata         CHAR(3) NOT NULL REFERENCES iata_codes(iata) ON DELETE CASCADE,
+  last_heard   TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  PRIMARY KEY (channel_hash, iata)
+);
+
+CREATE INDEX idx_channel_iatas_iata ON channel_iatas(iata);
+
+-- Seed from retained packets; parallelism off so the join spills to disk, not /dev/shm.
+SET max_parallel_workers_per_gather = 0;
+
+INSERT INTO channel_iatas (channel_hash, iata, last_heard)
+SELECT p.channel_hash, po.iata, MAX(po.heard_at)
+FROM packets p
+JOIN packet_observations po ON po.packet_hash = p.packet_hash
+WHERE p.channel_hash IS NOT NULL
+GROUP BY p.channel_hash, po.iata;
+
+RESET max_parallel_workers_per_gather;

db/migrations/012_trace_iatas.sql

@@ -0,0 +1,23 @@
+-- Per-IATA trace activity so the trace filter stops joining every trace packet
+-- to observations (the hash join was overrunning /dev/shm).
+
+CREATE TABLE trace_iatas (
+  trace_tag  BYTEA NOT NULL,
+  iata       CHAR(3) NOT NULL REFERENCES iata_codes(iata) ON DELETE CASCADE,
+  last_heard TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  PRIMARY KEY (trace_tag, iata)
+);
+
+CREATE INDEX idx_trace_iatas_iata ON trace_iatas(iata);
+
+-- Seed from retained packets; parallelism off so the join spills to disk, not /dev/shm.
+SET max_parallel_workers_per_gather = 0;
+
+INSERT INTO trace_iatas (trace_tag, iata, last_heard)
+SELECT p.trace_tag, po.iata, MAX(po.heard_at)
+FROM packets p
+JOIN packet_observations po ON po.packet_hash = p.packet_hash
+WHERE p.trace_tag IS NOT NULL
+GROUP BY p.trace_tag, po.iata;
+
+RESET max_parallel_workers_per_gather;

db/migrations/013_packets_scope_index.sql

@@ -0,0 +1,3 @@
+-- Scope stats count off scope_id, which had no index.
+
+CREATE INDEX idx_packets_scope ON packets(scope_id) WHERE scope_id IS NOT NULL;

db/migrations/014_known_routes_last_seen_index.sql

@@ -0,0 +1,3 @@
+-- Routes list orders by last_seen; index it (was seq-scanning ~150k rows per page).
+
+CREATE INDEX idx_known_routes_last_seen ON known_routes(last_seen DESC);

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,17 @@
+-- Precomputed payload breakdown, bucketed by hour so the stats endpoint can
+-- serve any window (24h/7d/30d) by summing the buckets in range instead of
+-- scanning a month of observations per request.
+
+CREATE MATERIALIZED VIEW mv_payload_breakdown_by_iata AS
+SELECT
+  iata,
+  payload_type,
+  date_trunc('hour', heard_at)::timestamptz AS bucket,
+  COUNT(*) AS count
+FROM packet_observations
+WHERE heard_at > NOW() - INTERVAL '30 days'
+  AND payload_type IS NOT NULL
+GROUP BY iata, payload_type, date_trunc('hour', heard_at);
+
+CREATE UNIQUE INDEX idx_mv_payload_breakdown
+  ON mv_payload_breakdown_by_iata(iata, payload_type, bucket);

db/migrations/017_mv_top_observers.sql

@@ -0,0 +1,19 @@
+-- Precomputed observer activity, bucketed by hour so top-observers can be
+-- served for any window (24h/7d/30d) by summing the buckets in range. Was a
+-- ~2s per-request scan of a week of observations.
+
+CREATE MATERIALIZED VIEW mv_top_observers_by_iata AS
+SELECT
+  po.iata,
+  po.observer_id,
+  o.display_name,
+  o.observer_type,
+  date_trunc('hour', po.heard_at)::timestamptz AS bucket,
+  COUNT(*) AS observation_count
+FROM packet_observations po
+JOIN observers o ON o.id = po.observer_id
+WHERE po.heard_at > NOW() - INTERVAL '30 days'
+GROUP BY po.iata, po.observer_id, o.display_name, o.observer_type, date_trunc('hour', po.heard_at);
+
+CREATE UNIQUE INDEX idx_mv_top_observers
+  ON mv_top_observers_by_iata(iata, observer_id, bucket);

db/migrations/018_mv_top_talkers.sql

@@ -0,0 +1,19 @@
+-- Hourly-bucketed talker counts so top-talkers serves any window from a small
+-- table instead of the live channel_messages/observations join (millions of rows
+-- per request). Per-IATA counts; cross-IATA senders are approximate all-regions.
+
+CREATE MATERIALIZED VIEW mv_top_talkers_by_iata AS
+SELECT
+  po.iata,
+  cm.sender_name,
+  date_trunc('hour', cm.sent_at)::timestamptz AS bucket,
+  COUNT(DISTINCT cm.id) AS message_count,
+  MAX(cm.sent_at) AS last_sent
+FROM channel_messages cm
+JOIN packet_observations po ON po.packet_hash = cm.packet_hash
+WHERE cm.sender_name IS NOT NULL
+  AND cm.sent_at > NOW() - INTERVAL '30 days'
+GROUP BY po.iata, cm.sender_name, date_trunc('hour', cm.sent_at);
+
+CREATE UNIQUE INDEX idx_mv_top_talkers
+  ON mv_top_talkers_by_iata(iata, sender_name, bucket);

db/migrations/019_mv_top_advertisers.sql

@@ -0,0 +1,22 @@
+-- Hourly-bucketed advert counts (distinct ADVERT packets per node = adverts sent,
+-- not heard) so top-advertisers serves any window from a small table, not the live
+-- scan. Per-IATA counts; cross-IATA adverts are approximate all-regions.
+
+CREATE MATERIALIZED VIEW mv_top_advertisers_by_iata AS
+SELECT
+  po.iata,
+  n.id AS node_id,
+  n.name,
+  n.node_type,
+  date_trunc('hour', po.heard_at)::timestamptz AS bucket,
+  COUNT(DISTINCT p.packet_hash) AS advert_count,
+  MAX(po.heard_at) AS last_heard
+FROM packets p
+JOIN packet_observations po ON po.packet_hash = p.packet_hash
+JOIN nodes n ON n.public_key = p.origin_pubkey
+WHERE p.payload_type = 4 -- ADVERT
+  AND po.heard_at > NOW() - INTERVAL '30 days'
+GROUP BY po.iata, n.id, n.name, n.node_type, date_trunc('hour', po.heard_at);
+
+CREATE UNIQUE INDEX idx_mv_top_advertisers
+  ON mv_top_advertisers_by_iata(iata, node_id, bucket);

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) {