dandri/beacon-server
1
0
Fork 0
mirror of https://github.com/MeshCore-Beacon/beacon-server.git synced 2026-10-20 16:52:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge upstream 2.0 baseline into isolated topology integration

Browse Source
This commit is contained in:
n30nex
2026-09-30 21:54:09 -04:00
parent a5c8d2839b 3ac4035942
commit be6360305c
60 changed files with 1555 additions and 2000 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/ci.yml
+1 -1
View File
@@ -58,7 +58,7 @@ jobs:
- name: Verify stats and endpoint queries against PostgreSQL 16
env:
BEACON_TEST_POSTGRES_DSN: postgres://postgres:backup-ci-only@127.0.0.1:5432/postgres?sslmode=disable
run: go test ./db -run '^Test(Signal|Paths|PacketSummaries|PacketEndpointsResolveLive|ObserverMetrics|RouteEvidence|RouteEvidenceIndex|AnalyticsRetention|AnalyticsRetentionConcurrent|DeleteOldPacketsBatches|MeshMapperCatalogue|ChannelMessageScopes|NodeLocationReset|Reconfirm[A-Za-z]*)Postgres$' -count=1 -v
run: go test ./db -run '^Test(Signal|Paths|PacketSummaries|PacketEndpointsResolveLive|ObserverMetrics|RouteEvidence|RouteEvidenceIndex|AnalyticsRetention|AnalyticsRetentionConcurrent|DeleteOldPacketsBatches|MeshMapperCatalogue|ChannelMessageScopes|NodeLocationReset|RunMigrations[A-Za-z]*|Reconfirm[A-Za-z]*)Postgres$' -count=1 -v
- name: Verify backup command against PostgreSQL 16
run: |
CONTRIBUTING.md
+3 -3
View File
@@ -79,8 +79,8 @@ swag init # if you changed any handler or api type (see below)
All schema changes must include a proper migration path:
- Add a new migration file to `db/migrations/` following the existing naming
convention (e.g. `002_add_observation_count.sql`). Do not modify existing
migration files — append only via new files.
convention (e.g. `002_add_observation_count.sql`). `001_baseline.sql` is the
2.0.0 schema; do not modify existing migration files — append only via new files.
- Update `db/queries/queries.sql` with any new or modified queries
- Re-run `sqlc generate` to regenerate `db/sqlc/`
- Update the store layer in `db/` to expose the new functionality
@@ -239,7 +239,7 @@ Scopes are optional but helpful for larger codebases. Common scopes: `api`,
```
cmd/beacon/ — main entry point, wiring, startup
db/ — store layer: sqlc-generated code + thin mapping layer
migrations/ — SQL schema (single file, append only)
migrations/ — SQL schema (2.0.0 baseline + numbered files, append only)
queries/ — SQL queries (input to sqlc)
sqlc/ — generated Go code (do not edit by hand)
internal/
db/accounts_integration_test.go
+1 -7
View File
@@ -50,15 +50,9 @@ func TestAccountsPostgres(t *testing.T) {
t.Fatal(err)
}
defer pool.Close()
sql, err := migrationFiles.ReadFile("migrations/034_accounts.sql")
if err != nil {
if err := RunMigrations(ctx, pool); err != nil {
t.Fatal(err)
}
for i := 0; i < 2; i++ {
if err := applyMigration(ctx, pool, string(sql)); err != nil {
t.Fatal(err)
}
}
store := New(pool, 0, 0)
items, err := store.ListAccounts(ctx)
if err != nil || items == nil || len(items) != 0 {
db/analytics_retention_integration_test.go
+23 -45
View File
@@ -16,13 +16,20 @@ import (
var retainedViews = []string{"mv_hourly_iata_stats", "mv_payload_breakdown_by_iata", "mv_top_observers_by_iata", "mv_top_talkers_by_iata", "mv_top_advertisers_by_iata", "mv_observer_activity_hourly", "mv_signal_stats_hourly", "mv_path_stats_hourly"}
func analyticsSnapshot(t *testing.T, ctx context.Context, tx pgx.Tx) map[string]string {
func refreshRetainedViews(t *testing.T, ctx context.Context, tx pgx.Tx) {
t.Helper()
result := map[string]string{}
for _, view := range retainedViews {
if _, err := tx.Exec(ctx, "REFRESH MATERIALIZED VIEW "+view); err != nil {
t.Fatal(err)
}
}
}
func analyticsSnapshot(t *testing.T, ctx context.Context, tx pgx.Tx) map[string]string {
t.Helper()
refreshRetainedViews(t, ctx, tx)
result := map[string]string{}
for _, view := range retainedViews {
var rows string
if err := tx.QueryRow(ctx, "SELECT COALESCE(jsonb_agg(r ORDER BY r::text),'[]')::text FROM (SELECT to_jsonb(v) r FROM "+view+" v) s").Scan(&rows); err != nil {
t.Fatal(err)
@@ -34,9 +41,8 @@ func analyticsSnapshot(t *testing.T, ctx context.Context, tx pgx.Tx) map[string]
func TestAnalyticsRetentionConcurrentPostgres(t *testing.T) {
ctx, tx := retentionTx(t)
analyticsTables(t, ctx, tx)
applyStatsMigration(t, ctx, tx, "039_analytics_retention.sql")
if _, err := tx.Exec(ctx, `INSERT INTO packets (packet_hash,last_heard_at) VALUES ('\x01',NOW()-interval '5 days'); INSERT INTO observers(id) VALUES ('00000000-0000-0000-0000-000000000001')`); err != nil {
applyBaseline(t, ctx, tx)
if _, err := tx.Exec(ctx, `INSERT INTO packets (packet_hash,payload_type,payload_version,route_type,raw_payload,raw_header,first_heard_at,last_heard_at) VALUES ('\x01',4,0,1,'\x00','\x00',NOW()-interval '5 days',NOW()-interval '5 days'); INSERT INTO observers(id,public_key) VALUES ('00000000-0000-0000-0000-000000000001','\x01')`); err != nil {
t.Fatal(err)
}
var schema string
@@ -70,7 +76,7 @@ func TestAnalyticsRetentionConcurrentPostgres(t *testing.T) {
}
defer ingest.Rollback(context.Background())
// An in-flight FK insert holds KEY SHARE on the packet until it commits.
if _, err = ingest.Exec(ctx, `INSERT INTO packet_observations(packet_hash,observer_id,iata,heard_at) VALUES ('\x01','00000000-0000-0000-0000-000000000001','YVR',NOW()-interval '5 days')`); err != nil {
if _, err = ingest.Exec(ctx, `INSERT INTO packet_observations(packet_hash,observer_id,iata,heard_at,path_length_byte,hash_size,hop_count) VALUES ('\x01','00000000-0000-0000-0000-000000000001','YVR',NOW()-interval '5 days',0,1,0)`); err != nil {
t.Fatal(err)
}
q := sqlc.New(conn)
@@ -114,33 +120,11 @@ func TestAnalyticsRetentionConcurrentPostgres(t *testing.T) {
}
}
// Minimal real tables keep this regression runnable against an empty CI database.
func analyticsTables(t *testing.T, ctx context.Context, tx pgx.Tx) {
t.Helper()
isolateStatsSchema(t, ctx, tx)
_, err := tx.Exec(ctx, `
CREATE TABLE packets (packet_hash bytea PRIMARY KEY, payload_type smallint, payload_version smallint,
route_type smallint, raw_payload bytea, raw_header bytea, origin_pubkey bytea,
first_heard_at timestamptz, last_heard_at timestamptz);
CREATE INDEX ON packets(last_heard_at);
CREATE TABLE packet_observations (id bigserial PRIMARY KEY, packet_hash bytea REFERENCES packets ON DELETE CASCADE,
observer_id uuid NOT NULL, iata char(3) NOT NULL, heard_at timestamptz NOT NULL,
path_length_byte smallint, hash_size smallint, hop_count smallint, path_bytes bytea,
snr real, rssi smallint, airtime_ms real, payload_type smallint, UNIQUE(packet_hash,observer_id));
CREATE TABLE observers (id uuid PRIMARY KEY, public_key bytea, display_name text, observer_type text);
CREATE TABLE nodes (id uuid PRIMARY KEY, public_key bytea UNIQUE, node_type smallint, name text);
CREATE TABLE channel_messages (id bigserial PRIMARY KEY, channel_id integer, packet_hash bytea UNIQUE REFERENCES packets ON DELETE CASCADE,
sender_name text, content text, sent_at timestamptz);
`)
if err != nil {
t.Fatal(err)
}
}
func TestAnalyticsRetentionPostgres(t *testing.T) {
ctx, tx := retentionTx(t)
analyticsTables(t, ctx, tx)
applyBaseline(t, ctx, tx)
_, err := tx.Exec(ctx, `
INSERT INTO channels (id,channel_hash) VALUES (1,'\x01');
INSERT INTO observers (id,public_key,display_name,observer_type) SELECT ('00000000-0000-0000-0000-00000000000'||i)::uuid,int4send(i),CASE WHEN i=1 THEN NULL ELSE 'Observer '||i END,CASE WHEN i=1 THEN NULL ELSE 'test' END FROM generate_series(1,3) i;
INSERT INTO nodes (id,public_key,node_type,name) VALUES ('00000000-0000-0000-0000-000000000002','\x02',2,'Repeater');
INSERT INTO packets (packet_hash,payload_type,payload_version,route_type,raw_payload,raw_header,origin_pubkey,first_heard_at,last_heard_at)
@@ -152,22 +136,17 @@ func TestAnalyticsRetentionPostgres(t *testing.T) {
FROM packets CROSS JOIN (VALUES ('YVR',1),('YVR',2),('YYZ',3)) v(iata,n);
INSERT INTO channel_messages (channel_id,packet_hash,sender_name,content,sent_at)
SELECT 1,packet_hash,'Sender','body that must expire',first_heard_at FROM packets;
CREATE MATERIALIZED VIEW mv_hourly_iata_stats AS SELECT iata,date_trunc('hour',heard_at) AS hour,count(*) AS observation_count,count(DISTINCT packet_hash) AS unique_packets,count(DISTINCT observer_id) AS active_observers FROM packet_observations GROUP BY 1,2;
`)
if err != nil {
t.Fatal(err)
}
for _, migration := range []string{"016_mv_payload_breakdown.sql", "017_mv_top_observers.sql", "018_mv_top_talkers.sql", "019_mv_top_advertisers.sql", "021_mv_top_advertisers_route_type.sql", "032_mv_observer_activity.sql", "035_mv_signal_stats.sql", "036_mv_path_stats.sql"} {
applyStatsMigration(t, ctx, tx, migration)
}
applyStatsMigration(t, ctx, tx, "039_analytics_retention.sql")
for _, name := range []string{"idx_mv_signal_stats_hourly_hour", "idx_mv_path_stats_hourly_hour"} {
var definition string
if err := tx.QueryRow(ctx, "SELECT pg_get_indexdef(to_regclass($1))", name).Scan(&definition); err != nil || !strings.Contains(definition, "(hour)") {
t.Fatalf("missing unfiltered time index %s: %s %v", name, definition, err)
}
}
// The signal/path stores must read the 039 view definitions, not just 035/036.
refreshRetainedViews(t, ctx, tx)
store := &Store{q: sqlc.New(tx)}
hour := time.Now().UTC().Truncate(time.Hour)
signal, err := store.GetSignalStats(ctx, hour.Add(-6*24*time.Hour), hour, nil)
@@ -176,14 +155,14 @@ func TestAnalyticsRetentionPostgres(t *testing.T) {
}
// 12 observations; only the n=1 rows carry a signal (snr 0, rssi -100).
if signal.Receptions != 12 || signal.SNR.Samples != 4 || signal.RSSI.Samples != 4 {
t.Fatalf("signal stats on 039 views: %+v", signal)
t.Fatalf("signal stats: %+v", signal)
}
paths, err := store.GetPathStats(ctx, hour.Add(-6*24*time.Hour), hour, nil)
if err != nil {
t.Fatal(err)
}
if paths.Receptions != 12 {
t.Fatalf("path stats on 039 views: %+v", paths)
t.Fatalf("path stats: %+v", paths)
}
before := analyticsSnapshot(t, ctx, tx)
q := sqlc.New(tx)
@@ -216,18 +195,17 @@ func TestAnalyticsRetentionPostgres(t *testing.T) {
t.Errorf("%s lost or double-counted history after raw expiry: before=%s after=%s", view, before[view], after[view])
}
}
// Retry and migration journal retry preserve the exact summaries.
// A retry preserves the exact summaries.
if n, err := q.DeleteOldPackets(ctx, sqlc.DeleteOldPacketsParams{Cutoff: pgtype.Timestamptz{Time: cutoff, Valid: true}, BatchSize: 1000}); err != nil || n != 0 {
t.Fatalf("retry: %d %v", n, err)
}
applyStatsMigration(t, ctx, tx, "039_analytics_retention.sql")
for view, rows := range analyticsSnapshot(t, ctx, tx) {
if rows != before[view] {
t.Errorf("retry changed %s", view)
}
}
// Later receptions on a still-live packet must join its old bucket before expiry.
if _, err := tx.Exec(ctx, `INSERT INTO observers (id) VALUES ('00000000-0000-0000-0000-000000000004'); INSERT INTO packet_observations (packet_hash,observer_id,iata,heard_at,path_length_byte,hash_size,hop_count,snr,rssi,payload_type)
if _, err := tx.Exec(ctx, `INSERT INTO observers (id,public_key) VALUES ('00000000-0000-0000-0000-000000000004','\x04'); INSERT INTO packet_observations (packet_hash,observer_id,iata,heard_at,path_length_byte,hash_size,hop_count,snr,rssi,payload_type)
SELECT packet_hash,'00000000-0000-0000-0000-000000000004','YVR',first_heard_at,0,1,0,'NaN',-80,NULL FROM packets;
UPDATE packets SET last_heard_at=first_heard_at`); err != nil {
t.Fatal(err)
@@ -282,13 +260,13 @@ func TestAnalyticsRetentionPostgres(t *testing.T) {
}
// More than one populated cohort, with shared hourly keys across the boundary.
if _, err := tx.Exec(ctx, `
INSERT INTO observers (id) VALUES ('00000000-0000-0000-0000-000000000001');
INSERT INTO observers (id,public_key) VALUES ('00000000-0000-0000-0000-000000000001','\x01');
INSERT INTO nodes (id,public_key,node_type) VALUES ('00000000-0000-0000-0000-000000000002','\x02',2);
INSERT INTO packets (packet_hash,payload_type,route_type,origin_pubkey,first_heard_at,last_heard_at,raw_payload)
SELECT int4send(i),4,1,'\x02',date_trunc('hour',NOW())-interval '5 days',date_trunc('hour',NOW())-interval '5 days',decode(repeat('ab',1024),'hex') FROM generate_series(10000,11000) i;
INSERT INTO packets (packet_hash,payload_type,payload_version,route_type,origin_pubkey,first_heard_at,last_heard_at,raw_payload,raw_header)
SELECT int4send(i),4,0,1,'\x02',date_trunc('hour',NOW())-interval '5 days',date_trunc('hour',NOW())-interval '5 days',decode(repeat('ab',1024),'hex'),'\x00' FROM generate_series(10000,11000) i;
INSERT INTO packet_observations (packet_hash,observer_id,iata,heard_at,path_length_byte,hash_size,hop_count,payload_type,snr,rssi)
SELECT packet_hash,'00000000-0000-0000-0000-000000000001','YVR',first_heard_at,0,1,0,4,0,-100 FROM packets;
INSERT INTO channel_messages (packet_hash,sender_name,sent_at) SELECT packet_hash,'Sender',first_heard_at FROM packets;
INSERT INTO channel_messages (channel_id,packet_hash,sender_name,sent_at) SELECT 1,packet_hash,'Sender',first_heard_at FROM packets;
`); err != nil {
t.Fatal(err)
}
db/baseline_integration_test.go
+119
View File
@@ -0,0 +1,119 @@
// Copyright 2026 Beacon Contributors
// SPDX-License-Identifier: AGPL-3.0-or-later
package db
import (
"context"
"errors"
"os"
"strings"
"testing"
"time"
"github.com/google/uuid"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgxpool"
)
// Views cannot reference temporary tables. Keep fixtures and real migration DDL
// in a unique schema inside the caller's rolled-back transaction instead.
func isolateStatsSchema(t *testing.T, ctx context.Context, tx pgx.Tx) {
t.Helper()
schema := pgx.Identifier{"stats_test_" + uuid.NewString()}.Sanitize()
if _, err := tx.Exec(ctx, "CREATE SCHEMA "+schema+"; SET LOCAL search_path TO "+schema); err != nil {
t.Fatal(err)
}
}
// applyBaseline builds the real schema in an isolated schema, plus the IATAs fixtures use.
func applyBaseline(t *testing.T, ctx context.Context, tx pgx.Tx) {
t.Helper()
isolateStatsSchema(t, ctx, tx)
ddl, err := migrationFiles.ReadFile("migrations/" + baselineMigration)
if err != nil {
t.Fatal(err)
}
if _, err := tx.Exec(ctx, string(ddl)); err != nil {
t.Fatal(err)
}
if _, err := tx.Exec(ctx, "INSERT INTO iata_codes (iata) VALUES ('YVR'),('YYJ'),('YYZ'),('YOW')"); err != nil {
t.Fatal(err)
}
}
// schemaPool returns a pool whose connections all use a fresh, dropped-on-cleanup schema.
func schemaPool(t *testing.T) (context.Context, *pgxpool.Pool) {
t.Helper()
dsn := os.Getenv("BEACON_TEST_POSTGRES_DSN")
if dsn == "" {
t.Skip("set BEACON_TEST_POSTGRES_DSN for PostgreSQL regression")
}
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
t.Cleanup(cancel)
setup, err := pgx.Connect(ctx, dsn)
if err != nil {
t.Fatal(err)
}
schema := "migrate_test_" + strings.ReplaceAll(uuid.NewString(), "-", "")
quoted := pgx.Identifier{schema}.Sanitize()
if _, err := setup.Exec(ctx, "CREATE SCHEMA "+quoted); err != nil {
t.Fatal(err)
}
t.Cleanup(func() {
if _, err := setup.Exec(context.Background(), "DROP SCHEMA "+quoted+" CASCADE"); err != nil {
t.Error(err)
}
setup.Close(context.Background())
})
cfg, err := pgxpool.ParseConfig(dsn)
if err != nil {
t.Fatal(err)
}
cfg.ConnConfig.RuntimeParams["search_path"] = schema
cfg.MaxConns = 4
pool, err := pgxpool.NewWithConfig(ctx, cfg)
if err != nil {
t.Fatal(err)
}
t.Cleanup(pool.Close)
return ctx, pool
}
func TestRunMigrationsBaselinePostgres(t *testing.T) {
ctx, pool := schemaPool(t)
for range 2 {
if err := RunMigrations(ctx, pool); err != nil {
t.Fatal(err)
}
}
rows, err := pool.Query(ctx, "SELECT filename FROM schema_migrations ORDER BY filename")
if err != nil {
t.Fatal(err)
}
ledger, err := pgx.CollectRows(rows, pgx.RowTo[string])
if err != nil || len(ledger) != 1 || ledger[0] != baselineMigration {
t.Fatalf("ledger %v, %v", ledger, err)
}
// Stats refresh uses CONCURRENTLY, which needs populated views.
if _, err := pool.Exec(ctx, "REFRESH MATERIALIZED VIEW CONCURRENTLY mv_hourly_iata_stats"); err != nil {
t.Fatal(err)
}
}
func TestRunMigrationsRefusesPreBaselinePostgres(t *testing.T) {
for name, setup := range map[string]string{
"1.x ledger": "CREATE TABLE schema_migrations (filename text PRIMARY KEY); INSERT INTO schema_migrations VALUES ('001_initial_schema.sql')",
"pre-ledger 1.x": "CREATE TABLE packets (packet_hash bytea PRIMARY KEY)",
} {
t.Run(name, func(t *testing.T) {
ctx, pool := schemaPool(t)
if _, err := pool.Exec(ctx, setup); err != nil {
t.Fatal(err)
}
if err := RunMigrations(ctx, pool); !errors.Is(err, errPreBaseline) {
t.Fatalf("got %v, want errPreBaseline", err)
}
})
}
}
db/channel_cursor_integration_test.go
+2 -4
View File
@@ -56,10 +56,8 @@ func TestChannelCursorPostgres(t *testing.T) {
t.Fatal(err)
}
}
exec(`CREATE TEMP TABLE channels (LIKE public.channels INCLUDING ALL) ON COMMIT DROP;
CREATE TEMP TABLE channel_iatas (LIKE public.channel_iatas INCLUDING ALL) ON COMMIT DROP;
CREATE INDEX ON channels(last_seen DESC,id DESC);
INSERT INTO channels(id,channel_hash,key_fingerprint,last_seen) VALUES
applyBaseline(t, ctx, tx)
exec(`INSERT INTO channels(id,channel_hash,key_fingerprint,last_seen) VALUES
(1,'\xaa','\x01','2026-09-08 12:00:00+00'),(2,'\xaa','\x02','2026-09-08 12:00:00+00'),
(3,'\xaa','\x03','2026-09-08 12:00:00+00'),(4,'\xbb','\x04','2026-09-08 11:59:59+00'),
(5,'\xcc','\x05','2026-09-08 12:00:00.000321+00'),
db/channel_index_migration_test.go
-74
View File
@@ -1,74 +0,0 @@
// Copyright 2026 Beacon Contributors
// SPDX-License-Identifier: AGPL-3.0-or-later
package db
import (
"context"
"fmt"
"os"
"testing"
"time"
"github.com/jackc/pgx/v5"
)
func TestChannelIndexMigrationsPostgres(t *testing.T) {
dsn := os.Getenv("BEACON_TEST_POSTGRES_DSN")
if dsn == "" {
t.Skip("set BEACON_TEST_POSTGRES_DSN for PostgreSQL regression")
}
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
defer cancel()
conn, err := pgx.Connect(ctx, dsn)
if err != nil {
t.Fatal(err)
}
defer conn.Close(context.Background())
exec := func(query string) {
t.Helper()
if _, err := conn.Exec(ctx, query); err != nil {
t.Fatal(err)
}
}
schema := pgx.Identifier{fmt.Sprintf("channel_index_test_%d", time.Now().UnixNano())}.Sanitize()
exec("CREATE SCHEMA " + schema)
defer func() {
cleanup, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
if _, err := conn.Exec(cleanup, "DROP SCHEMA "+schema+" CASCADE"); err != nil {
t.Error(err)
}
}()
exec("SET search_path TO " + schema)
exec(`CREATE TABLE channels (id bigint PRIMARY KEY, last_seen timestamptz NOT NULL);
CREATE INDEX idx_channels_last_seen ON channels(last_seen DESC);
INSERT INTO channels VALUES (1, '2026-09-09 12:00:00+00');`)
migrate := func(name string) {
t.Helper()
query, err := migrationFiles.ReadFile("migrations/" + name)
if err != nil {
t.Fatal(err)
}
exec(string(query)) // Concurrent DDL must remain outside an explicit transaction.
}
checkReplacement := func() {
t.Helper()
var valid bool
if err := conn.QueryRow(ctx, `SELECT indisvalid AND indisready AND indnkeyatts=2
FROM pg_index WHERE indexrelid='idx_channels_last_seen_id'::regclass`).Scan(&valid); err != nil || !valid {
t.Fatalf("replacement index valid=%v, err=%v", valid, err)
}
}
migrate("028_channels_last_seen_id.sql")
checkReplacement()
for range 2 { // Retry after the drop succeeds but before its migration record is saved.
migrate("029_drop_channels_last_seen.sql")
var preserved bool
if err := conn.QueryRow(ctx, `SELECT to_regclass('idx_channels_last_seen') IS NULL
AND (SELECT count(*) FROM channels)=1`).Scan(&preserved); err != nil || !preserved {
t.Fatalf("old index removed and row preserved=%v, err=%v", preserved, err)
}
checkReplacement()
}
}
db/migrate.go
+24 -29
View File
@@ -82,6 +82,28 @@ func applyMigration(ctx context.Context, db execQuerier, sql string) error {
return err
}
const baselineMigration = "001_baseline.sql"
var errPreBaseline = errors.New("database schema predates Beacon 2.0.0; 2.0.0 needs a fresh database")
// checkBaseline refuses databases built by the 1.x migration chain, which the
// 2.0.0 baseline replaced; applying it on top would collide with existing tables.
func checkBaseline(ctx context.Context, db execQuerier) error {
var ledger, hasBaseline, hasPackets bool
err := db.QueryRow(ctx, `
SELECT EXISTS(SELECT 1 FROM schema_migrations),
EXISTS(SELECT 1 FROM schema_migrations WHERE filename = $1),
to_regclass('packets') IS NOT NULL`, baselineMigration,
).Scan(&ledger, &hasBaseline, &hasPackets)
if err != nil {
return fmt.Errorf("failed to check migration baseline: %w", err)
}
if (ledger && !hasBaseline) || (!ledger && hasPackets) {
return errPreBaseline
}
return nil
}
func RunMigrations(ctx context.Context, pool *pgxpool.Pool) error {
_, err := pool.Exec(ctx, `
CREATE TABLE IF NOT EXISTS schema_migrations (
@@ -93,35 +115,8 @@ func RunMigrations(ctx context.Context, pool *pgxpool.Pool) error {
return fmt.Errorf("failed to create schema_migrations: %w", err)
}
// After creating schema_migrations table, check if we need to bootstrap
var count int
err = pool.QueryRow(ctx, "SELECT COUNT(*) FROM schema_migrations").Scan(&count)
if err != nil {
return fmt.Errorf("failed to check migrations count: %w", err)
}
if count == 0 {
// Check if db is already initialized by looking for a known table
var exists bool
err = pool.QueryRow(ctx, `
SELECT EXISTS(
SELECT 1 FROM information_schema.tables
WHERE table_name = 'packets'
)
`).Scan(&exists)
if err != nil {
return fmt.Errorf("failed to check existing schema: %w", err)
}
if exists {
// Mark 001 as already applied
if _, err := pool.Exec(
ctx,
"INSERT INTO schema_migrations (filename) VALUES ($1)",
"001_initial_schema.sql",
); err != nil {
return fmt.Errorf("failed to bootstrap migrations: %w", err)
}
slog.Info("bootstrapped existing schema as 001_initial_schema.sql", "component", "db")
}
if err := checkBaseline(ctx, pool); err != nil {
return err
}
entries, err := fs.ReadDir(migrationFiles, "migrations")
db/migrations/001_baseline.sql
+1314
View File
File diff suppressed because it is too large Load Diff
db/migrations/001_initial_schema.sql
-439
View File
@@ -1,439 +0,0 @@
-- Copyright 2026 Beacon Contributors
-- SPDX-License-Identifier: agpl
-- ============================================================
-- Beacon schema migration
-- ============================================================
-- ============================================================
-- IATA CODES
-- ============================================================
CREATE TABLE iata_codes (
iata CHAR(3) PRIMARY KEY,
display_name TEXT,
approx_lat DOUBLE PRECISION,
approx_lng DOUBLE PRECISION,
added_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
-- ============================================================
-- REGIONS
-- ============================================================
CREATE TABLE regions (
id SERIAL PRIMARY KEY,
slug TEXT UNIQUE NOT NULL,
name TEXT NOT NULL,
description TEXT,
display_order INT DEFAULT 0,
center_lat DOUBLE PRECISION,
center_lng DOUBLE PRECISION,
zoom_level INT DEFAULT 8,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE TABLE region_iatas (
region_id INT NOT NULL REFERENCES regions(id) ON DELETE CASCADE,
iata CHAR(3) NOT NULL REFERENCES iata_codes(iata) ON DELETE CASCADE,
added_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
PRIMARY KEY (region_id, iata)
);
CREATE INDEX idx_region_iatas_iata ON region_iatas(iata);
-- ============================================================
-- TRANSPORT SCOPES
-- ============================================================
CREATE TABLE transport_scopes (
id SERIAL PRIMARY KEY,
name TEXT NOT NULL UNIQUE, -- normalized key string e.g. "#bc"
display_name TEXT, -- optional friendly name
transport_key BYTEA NOT NULL, -- 16-byte derived key
key_fingerprint BYTEA NOT NULL, -- SHA256(transport_key)[:8] for fast lookup
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_transport_scopes_fingerprint ON transport_scopes(key_fingerprint);
-- ============================================================
-- NODES (must come before observers due to observer_owners FK)
-- ============================================================
CREATE TABLE nodes (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
public_key BYTEA UNIQUE NOT NULL,
node_type SMALLINT NOT NULL,
name TEXT,
latitude DOUBLE PRECISION,
longitude DOUBLE PRECISION,
location_source TEXT,
last_advert_at TIMESTAMPTZ,
supports_multibyte_paths BOOLEAN NOT NULL DEFAULT FALSE,
supports_multibyte_traces BOOLEAN NOT NULL DEFAULT FALSE,
default_scope_id INT REFERENCES transport_scopes(id),
min_firmware_version TEXT GENERATED ALWAYS AS (
CASE
WHEN supports_multibyte_paths THEN '1.14.0+'
WHEN supports_multibyte_traces THEN '1.11.0+'
ELSE NULL
END
) STORED,
first_seen TIMESTAMPTZ NOT NULL DEFAULT NOW(),
last_seen TIMESTAMPTZ NOT NULL DEFAULT NOW(),
radio_freq_mhz REAL,
radio_sf SMALLINT,
radio_bw_khz REAL,
metadata JSONB
);
CREATE INDEX idx_nodes_type_last_seen ON nodes(node_type, last_seen DESC);
CREATE INDEX idx_nodes_location ON nodes(latitude, longitude)
WHERE latitude IS NOT NULL AND longitude IS NOT NULL;
CREATE INDEX idx_nodes_pubkey ON nodes(public_key);
CREATE INDEX idx_nodes_multibyte_paths ON nodes(supports_multibyte_paths) WHERE supports_multibyte_paths;
CREATE INDEX idx_nodes_multibyte_traces ON nodes(supports_multibyte_traces) WHERE supports_multibyte_traces;
CREATE INDEX idx_nodes_min_firmware ON nodes(min_firmware_version) WHERE min_firmware_version IS NOT NULL;
-- ============================================================
-- OBSERVERS
-- ============================================================
CREATE TABLE observers (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
public_key BYTEA UNIQUE NOT NULL,
display_name TEXT,
observer_type TEXT,
software_version TEXT,
hardware_model TEXT,
firmware_version TEXT,
firmware_build TEXT,
radio_freq_mhz REAL,
radio_sf SMALLINT,
radio_bw_khz REAL,
radio_cr SMALLINT,
battery_level REAL,
uptime_seconds BIGINT,
status_metadata JSONB,
last_status_at TIMESTAMPTZ,
first_seen TIMESTAMPTZ NOT NULL DEFAULT NOW(),
last_seen TIMESTAMPTZ NOT NULL DEFAULT NOW(),
observation_count BIGINT DEFAULT 0,
metadata JSONB
);
CREATE INDEX idx_observers_last_seen ON observers(last_seen DESC);
CREATE INDEX idx_observers_pubkey ON observers(public_key);
CREATE INDEX idx_observers_type ON observers(observer_type) WHERE observer_type IS NOT NULL;
CREATE TABLE observer_brokers (
observer_id UUID NOT NULL REFERENCES observers(id) ON DELETE CASCADE,
broker_name TEXT NOT NULL,
first_seen TIMESTAMPTZ NOT NULL DEFAULT NOW(),
last_seen TIMESTAMPTZ NOT NULL DEFAULT NOW(),
last_packet_at TIMESTAMPTZ,
auth_ok BOOLEAN DEFAULT TRUE,
PRIMARY KEY (observer_id, broker_name)
);
CREATE TABLE observer_locations (
observer_id UUID NOT NULL REFERENCES observers(id) ON DELETE CASCADE,
iata CHAR(3) REFERENCES iata_codes(iata),
latitude DOUBLE PRECISION,
longitude DOUBLE PRECISION,
reported_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
PRIMARY KEY (observer_id, reported_at)
);
CREATE TABLE observer_scopes (
observer_id UUID NOT NULL REFERENCES observers(id) ON DELETE CASCADE,
scope_id INT NOT NULL REFERENCES transport_scopes(id) ON DELETE CASCADE,
first_seen TIMESTAMPTZ NOT NULL DEFAULT NOW(),
last_seen TIMESTAMPTZ NOT NULL DEFAULT NOW(),
PRIMARY KEY (observer_id, scope_id)
);
CREATE INDEX idx_observer_locations_recent ON observer_locations(observer_id, reported_at DESC);
CREATE TABLE observer_telemetry (
id BIGSERIAL PRIMARY KEY,
observer_id UUID NOT NULL REFERENCES observers(id) ON DELETE CASCADE,
reported_at TIMESTAMPTZ NOT NULL,
battery_voltage_mv INT,
airtime_tx_pct REAL,
airtime_rx_pct REAL,
noise_floor_db REAL,
uptime_seconds BIGINT,
queue_length INT,
debug_flags INT,
receive_errors INT,
UNIQUE (observer_id, reported_at)
);
CREATE INDEX idx_telemetry_reported_brin ON observer_telemetry USING BRIN (reported_at);
CREATE INDEX idx_telemetry_observer_recent ON observer_telemetry(observer_id, reported_at DESC);
CREATE TABLE observer_owners (
observer_id UUID PRIMARY KEY REFERENCES observers(id) ON DELETE CASCADE,
owner_node_id UUID REFERENCES nodes(id),
owner_pubkey BYTEA,
contact_name TEXT,
contact_email TEXT,
notes TEXT,
source TEXT,
added_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_observer_owners_node ON observer_owners(owner_node_id) WHERE owner_node_id IS NOT NULL;
-- ============================================================
-- NODE IATAS AND SHORT IDS
-- ============================================================
CREATE TABLE node_iatas (
node_id UUID NOT NULL REFERENCES nodes(id) ON DELETE CASCADE,
iata CHAR(3) NOT NULL REFERENCES iata_codes(iata) ON DELETE CASCADE,
first_heard TIMESTAMPTZ NOT NULL DEFAULT NOW(),
last_heard TIMESTAMPTZ NOT NULL DEFAULT NOW(),
observation_count BIGINT DEFAULT 0,
PRIMARY KEY (node_id, iata)
);
CREATE INDEX idx_node_iatas_iata ON node_iatas(iata, last_heard DESC);
CREATE TABLE node_short_ids (
node_id UUID NOT NULL REFERENCES nodes(id) ON DELETE CASCADE,
iata CHAR(3) NOT NULL REFERENCES iata_codes(iata) ON DELETE CASCADE,
prefix_4 BYTEA NOT NULL,
prefix_1 BYTEA GENERATED ALWAYS AS (substring(prefix_4 from 1 for 1)) STORED,
prefix_2 BYTEA GENERATED ALWAYS AS (substring(prefix_4 from 1 for 2)) STORED,
prefix_3 BYTEA GENERATED ALWAYS AS (substring(prefix_4 from 1 for 3)) STORED,
PRIMARY KEY (node_id, iata)
);
CREATE INDEX idx_short_ids_p1 ON node_short_ids(iata, prefix_1);
CREATE INDEX idx_short_ids_p2 ON node_short_ids(iata, prefix_2);
CREATE INDEX idx_short_ids_p3 ON node_short_ids(iata, prefix_3);
CREATE INDEX idx_short_ids_p4 ON node_short_ids(iata, prefix_4);
-- ============================================================
-- PACKETS
-- ============================================================
CREATE TABLE packets (
packet_hash BYTEA PRIMARY KEY,
payload_type SMALLINT NOT NULL,
payload_version SMALLINT NOT NULL,
route_type SMALLINT NOT NULL,
transport_codes_present BOOLEAN DEFAULT FALSE,
region_code INT,
sub_region_code INT,
scope_id INT REFERENCES transport_scopes(id),
origin_pubkey BYTEA,
raw_payload BYTEA NOT NULL,
raw_header BYTEA NOT NULL,
parsed_payload JSONB,
decrypted BOOLEAN DEFAULT FALSE,
channel_hash BYTEA,
trace_tag BYTEA,
first_heard_at TIMESTAMPTZ NOT NULL,
last_heard_at TIMESTAMPTZ NOT NULL
);
CREATE INDEX idx_packets_first_heard_brin ON packets USING BRIN (first_heard_at);
CREATE INDEX idx_packets_payload_type ON packets(payload_type, first_heard_at DESC);
CREATE INDEX idx_packets_route_type ON packets(route_type, first_heard_at DESC);
CREATE INDEX idx_packets_origin ON packets(origin_pubkey, first_heard_at DESC)
WHERE origin_pubkey IS NOT NULL;
CREATE INDEX idx_packets_channel ON packets(channel_hash, first_heard_at DESC)
WHERE channel_hash IS NOT NULL;
CREATE INDEX idx_packets_trace_tag ON packets(trace_tag) WHERE trace_tag IS NOT NULL;
-- ============================================================
-- PACKET OBSERVATIONS
-- ============================================================
CREATE TABLE packet_observations (
id BIGSERIAL PRIMARY KEY,
packet_hash BYTEA NOT NULL REFERENCES packets(packet_hash) ON DELETE CASCADE,
observer_id UUID NOT NULL REFERENCES observers(id),
iata CHAR(3) NOT NULL REFERENCES iata_codes(iata),
heard_at TIMESTAMPTZ NOT NULL,
path_length_byte SMALLINT NOT NULL,
hash_size SMALLINT NOT NULL,
hop_count SMALLINT NOT NULL,
path_bytes BYTEA,
rssi SMALLINT,
snr REAL,
propagation_time_ms INT,
radio_freq_mhz REAL,
spread_factor SMALLINT,
bandwidth_khz REAL,
coding_rate SMALLINT,
source_broker TEXT,
UNIQUE (packet_hash, observer_id, heard_at)
);
CREATE INDEX idx_observations_heard_brin ON packet_observations USING BRIN (heard_at);
CREATE INDEX idx_observations_iata_heard ON packet_observations(iata, heard_at DESC);
CREATE INDEX idx_observations_observer ON packet_observations(observer_id, heard_at DESC);
CREATE INDEX idx_observations_packet ON packet_observations(packet_hash);
-- ============================================================
-- CHANNELS AND CHAT MESSAGES
-- ============================================================
CREATE TABLE channels (
id SERIAL PRIMARY KEY,
channel_hash BYTEA NOT NULL,
-- key_fingerprint is the first 8 bytes of SHA256(key_bytes).
-- Combined with channel_hash it uniquely identifies a channel,
-- since a 1-byte hash has high collision probability.
-- NULL when the key is unknown (hash-only record).
key_fingerprint BYTEA,
name TEXT,
hashtag TEXT UNIQUE, -- e.g. "meshcore"; only set for hashtag-derived channels
is_hashtag BOOLEAN DEFAULT FALSE,
is_public BOOLEAN DEFAULT FALSE,
key_known BOOLEAN DEFAULT FALSE,
first_seen TIMESTAMPTZ NOT NULL DEFAULT NOW(),
last_seen TIMESTAMPTZ NOT NULL DEFAULT NOW(),
message_count BIGINT DEFAULT 0,
-- Unique on (hash, fingerprint): allows multiple channels with the same
-- hash as long as they have different keys. NULL fingerprint is allowed
-- once per hash (the unknown-key record).
UNIQUE (channel_hash, key_fingerprint)
);
CREATE UNIQUE INDEX idx_channels_hash_no_key
ON channels (channel_hash)
WHERE key_fingerprint IS NULL;
CREATE INDEX idx_channels_last_seen ON channels(last_seen DESC);
CREATE INDEX idx_channels_hash ON channels(channel_hash);
CREATE INDEX idx_channels_hashtag ON channels(hashtag) WHERE hashtag IS NOT NULL;
CREATE TABLE channel_keys (
channel_id INT PRIMARY KEY REFERENCES channels(id) ON DELETE CASCADE,
key_bytes BYTEA NOT NULL,
key_fingerprint BYTEA NOT NULL, -- first 8 bytes of SHA256(key_bytes)
added_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
added_by TEXT
);
CREATE TABLE channel_messages (
id BIGSERIAL PRIMARY KEY,
channel_id INT NOT NULL REFERENCES channels(id),
packet_hash BYTEA NOT NULL REFERENCES packets(packet_hash),
sender_name TEXT,
sender_pubkey BYTEA,
content TEXT,
sent_at TIMESTAMPTZ NOT NULL,
UNIQUE (packet_hash)
);
CREATE INDEX idx_channel_messages_channel ON channel_messages(channel_id, sent_at DESC);
CREATE INDEX idx_channel_messages_sent_brin ON channel_messages USING BRIN (sent_at);
-- ============================================================
-- ROUTES
-- ============================================================
CREATE TABLE known_routes (
id BIGSERIAL PRIMARY KEY,
node_ids UUID[] NOT NULL, -- resolved node UUIDs in hop order
hash_prefix BYTEA[] NOT NULL, -- raw hash bytes in hop order for prefix matching
iata CHAR(3) NOT NULL REFERENCES iata_codes(iata) ON DELETE CASCADE,
hop_count INT NOT NULL,
first_seen TIMESTAMPTZ NOT NULL DEFAULT NOW(),
last_seen TIMESTAMPTZ NOT NULL DEFAULT NOW(),
observation_count BIGINT NOt NULL DEFAULT 1,
UNIQUE (node_ids, iata)
);
CREATE INDEX idx_known_routes_iata ON known_routes(iata);
CREATE INDEX idx_known_routes_hop_count ON known_routes(iata, hop_count);
-- ============================================================
-- NEIGHBORS
-- ============================================================
CREATE TABLE node_neighbors (
node_id UUID NOT NULL REFERENCES nodes(id) ON DELETE CASCADE,
neighbor_id UUID NOT NULL REFERENCES nodes(id) ON DELETE CASCADE,
iata CHAR(3) NOT NULL REFERENCES iata_codes(iata) ON DELETE CASCADE,
first_seen TIMESTAMPTZ NOT NULL DEFAULT NOW(),
last_seen TIMESTAMPTZ NOT NULL DEFAULT NOW(),
observation_count BIGINT NOT NULL DEFAULT 1,
PRIMARY KEY (node_id, neighbor_id, iata)
);
CREATE INDEX idx_node_neighbors_node ON node_neighbors(node_id, iata);
CREATE INDEX idx_node_neighbors_neighbor ON node_neighbors(neighbor_id, iata);
-- ============================================================
-- MATERIALIZED VIEWS
-- ============================================================
CREATE MATERIALIZED VIEW mv_hourly_iata_stats AS
SELECT
iata,
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
FROM packet_observations
WHERE heard_at > NOW() - INTERVAL '7 days'
GROUP BY iata, date_trunc('hour', heard_at);
CREATE MATERIALIZED VIEW mv_top_nodes_by_iata AS
SELECT
ni.iata,
ni.node_id,
n.name,
n.node_type,
ni.observation_count,
ni.last_heard
FROM node_iatas ni
JOIN nodes n ON n.id = ni.node_id
WHERE ni.last_heard > NOW() - INTERVAL '7 days';
CREATE UNIQUE INDEX idx_mv_top_nodes
ON mv_top_nodes_by_iata(iata, node_id);
CREATE UNIQUE INDEX idx_mv_hourly_iata_stats
ON mv_hourly_iata_stats(iata, hour);
CREATE MATERIALIZED VIEW mv_radio_presets AS
SELECT
concat(o.radio_freq_mhz, ',', o.radio_bw_khz, ',', o.radio_sf)::text AS preset,
(SELECT po.iata FROM packet_observations po WHERE po.observer_id = o.id ORDER BY po.heard_at DESC LIMIT 1) AS iata,
'observer' AS source_type,
COUNT(*) AS count
FROM observers o
WHERE o.radio_freq_mhz IS NOT NULL
AND o.radio_sf IS NOT NULL
AND o.radio_bw_khz IS NOT NULL
GROUP BY concat(o.radio_freq_mhz, ',', o.radio_bw_khz, ',', o.radio_sf)::text,
(SELECT po.iata FROM packet_observations po WHERE po.observer_id = o.id ORDER BY po.heard_at DESC LIMIT 1)
HAVING (SELECT po.iata FROM packet_observations po WHERE po.observer_id = o.id ORDER BY po.heard_at DESC LIMIT 1) IS NOT NULL
UNION ALL
SELECT
concat(n.radio_freq_mhz, ',', n.radio_bw_khz, ',', n.radio_sf) AS preset,
ni.iata,
'node' AS source_type,
COUNT(*) AS count
FROM nodes n
JOIN node_iatas ni ON ni.node_id = n.id
WHERE n.radio_freq_mhz IS NOT NULL
AND n.radio_sf IS NOT NULL
AND n.radio_bw_khz IS NOT NULL
GROUP BY concat(n.radio_freq_mhz, ',', n.radio_bw_khz, ',', n.radio_sf), ni.iata;
CREATE UNIQUE INDEX idx_mv_radio_presets
ON mv_radio_presets(preset, iata, source_type);
db/migrations/002_drop_single_hop_routes.sql
-7
View File
@@ -1,7 +0,0 @@
-- Copyright 2026 Beacon Contributors
-- SPDX-License-Identifier: agpl
-- 002_drop_single_hop_routes.sql
-- routes were being stored as long as the path was
-- greater than 0.
DELETE FROM known_routes WHERE hop_count < 2;
db/migrations/003_tag_ping_traces.sql
-9
View File
@@ -1,9 +0,0 @@
-- Copyright 2026 Beacon Contributors
-- SPDX-License-Identifier: agpl
-- 003_tag_ping_traces.sql
UPDATE packets
SET parsed_payload = jsonb_set(parsed_payload, '{type}', '"PING"')
WHERE trace_tag IS NOT NULL
AND jsonb_array_length(parsed_payload->'pathHashes') = 1
AND parsed_payload->>'type' = 'TRACE';
db/migrations/004_clean_non_infra_neighbors.sql
-11
View File
@@ -1,11 +0,0 @@
-- Remove neighbor records where either side is not infrastructure (repeater or room).
-- These are dirty rows from historical prefix collisions resolving to non-forwarding nodes.
DELETE FROM node_neighbors nn
USING nodes n
WHERE nn.neighbor_id = n.id
AND n.node_type NOT IN (2, 3);
DELETE FROM node_neighbors nn
USING nodes n
WHERE nn.node_id = n.id
AND n.node_type NOT IN (2, 3);
db/migrations/005_add_neighbor_snr.sql
-5
View File
@@ -1,5 +0,0 @@
-- Adds an optional SNR reading to node_neighbors. Most rows (advert
-- path-hop-derived topology) will leave this null; it's populated only
-- when the neighbor relationship comes with an actual signal reading
-- (e.g. observer-detected via DISCOVER_RESP).
ALTER TABLE node_neighbors ADD COLUMN snr REAL;
db/migrations/006_fix_observation_dedup_key.sql
-24
View File
@@ -1,24 +0,0 @@
-- 006_fix_observation_dedup_key.sql
--
-- The previous unique constraint on (packet_hash, observer_id, heard_at) was
-- too permissive: heard_at is stamped at ingest time (time.Now()), so MQTT
-- redeliveries of the same packet arrive with slightly different timestamps
-- and bypass dedup, producing duplicate observations for the same observer.
--
-- An observer either heard a packet or they didn't. Drop heard_at from the
-- constraint so that (packet_hash, observer_id) is the dedup key.
-- Remove duplicate observations, keeping the earliest (lowest id) per (packet_hash, observer_id).
DELETE FROM packet_observations
WHERE id NOT IN (
SELECT MIN(id)
FROM packet_observations
GROUP BY packet_hash, observer_id
);
ALTER TABLE packet_observations
DROP CONSTRAINT packet_observations_packet_hash_observer_id_heard_at_key;
ALTER TABLE packet_observations
ADD CONSTRAINT packet_observations_packet_hash_observer_id_key
UNIQUE (packet_hash, observer_id);
db/migrations/007_dedupe_hashonly_channels.sql
-21
View File
@@ -1,21 +0,0 @@
-- Copyright 2026 Beacon Contributors
-- SPDX-License-Identifier: AGPL-3.0-or-later
-- Cleans up the ghost "unknown key" channel rows created by the
-- side_effects.go bug where UpsertChannelHashOnly was called
-- unconditionally on every GRP_TXT packet, even when the key was
-- known and the message decrypted successfully.
--
-- Safe because hash-only rows (key_fingerprint IS NULL) never have
-- channel_messages attached to them -- InsertChannelMessage is only
-- ever called against a keyed channel row. We only delete a hash-only
-- row when a keyed row already exists for the same channel_hash;
-- hashes with no known key at all (genuinely unknown) are left as-is.
DELETE FROM channels c
WHERE c.key_fingerprint IS NULL
AND EXISTS (
SELECT 1 FROM channels k
WHERE k.channel_hash = c.channel_hash
AND k.key_fingerprint IS NOT NULL
);
db/migrations/008_packets_last_heard_index.sql
-8
View File
@@ -1,8 +0,0 @@
-- 008_packets_last_heard_index.sql
--
-- The packet list orders by last_heard_at but there was no index on it, so
-- every page seq-scanned and sorted all packets (~2.8s per page on a 1.5M
-- row table; 31ms with the index). Affordable to maintain now that presence
-- coalescing keeps this column from being rewritten on every observation.
CREATE INDEX idx_packets_last_heard ON packets(last_heard_at DESC);
db/migrations/009_clean_zeroed_observer_telemetry.sql
-19
View File
@@ -1,19 +0,0 @@
-- Copyright 2026 Beacon Contributors
-- SPDX-License-Identifier: AGPL-3.0-or-later
-- Cleans up observer_telemetry rows written from /status messages whose
-- "stats" object was missing or renamed. status.go's json.Unmarshal
-- silently tolerated the missing object, leaving the stats struct at Go
-- zero-values, and the row was inserted unconditionally from there --
-- uptime_secs=0, battery_mv=0, noise_floor=0, tx/rx_air_secs=0. With the
-- hourly ON CONFLICT (observer_id, reported_at) DO NOTHING dedup,
-- whichever message landed first per hour won, so these zero rows
-- regularly became the hour's telemetry and corrupted the 24h/7d/30d
-- aggregates (AVG(noise_floor_db), AVG(battery_voltage_mv), MAX-MIN
-- airtime).
--
-- Safe because a running observer never reports uptime_seconds == 0;
-- this is the same signal now used by status.go to skip the insert going
-- forward (see internal/ingest/status.go handleStatus).
DELETE FROM observer_telemetry WHERE uptime_seconds = 0;
db/migrations/010_add_node_clock_drift.sql
-16
View File
@@ -1,16 +0,0 @@
-- Copyright 2026 Beacon Contributors
-- SPDX-License-Identifier: AGPL-3.0-or-later
-- Stores the device-reported clock drift measured from each ADVERT: the signed
-- advert body already carries the device's own wall-clock timestamp
-- (advert.Timestamp, epoch seconds), which the server previously decoded and
-- then discarded before it reached the nodes table. This column stores
-- (device timestamp - server receive time) in seconds, computed at the same
-- moment nodes.last_advert_at is set to NOW(), so last_advert_at doubles as
-- the "when was this drift measured" timestamp with no separate column
-- needed. NULL until a node's first successfully-verified advert is upserted.
--
-- +ve = device clock ahead of the server; -ve = behind. Surfaced on the node
-- API for repeaters/room servers only (nodeType 2/3) -- see internal/api/nodes.go.
ALTER TABLE nodes ADD COLUMN device_clock_drift_seconds INTEGER;
db/migrations/011_channel_iatas.sql
-23
View File
@@ -1,23 +0,0 @@
-- 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
-23
View File
@@ -1,23 +0,0 @@
-- 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
-3
View File
@@ -1,3 +0,0 @@
-- 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
-3
View File
@@ -1,3 +0,0 @@
-- 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
-17
View File
@@ -1,17 +0,0 @@
-- 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
-17
View File
@@ -1,17 +0,0 @@
-- 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
-19
View File
@@ -1,19 +0,0 @@
-- 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
-19
View File
@@ -1,19 +0,0 @@
-- 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
-22
View File
@@ -1,22 +0,0 @@
-- 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/migrations/020_add_iata_border.sql
-21
View File
@@ -1,21 +0,0 @@
-- Copyright 2026 Beacon Contributors
-- SPDX-License-Identifier: AGPL-3.0-or-later
-- Optional GeoJSON border for the region border map feature. Stores a full
-- GeoJSON Feature (type/bbox/properties/geometry), not a bare geometry, so
-- properties can be added later without a schema change. Geometry is always
-- Polygon or MultiPolygon; bbox is computed and stored at write time (see
-- internal/config/border.go) so reads never need to recompute it.
--
-- NULL for the vast majority of IATAs, matching display_name/approx_lat/
-- approx_lng's existing "optional, mostly unset" pattern. Written via the
-- same config-file-driven seeding path as those columns (internal/config/
-- seed.go), not a runtime HTTP write endpoint -- this codebase has none.
--
-- Deliberately excluded from GET /api/v1/iatas (border polygons are large --
-- thousands of vertices, MultiPolygon for archipelago coastlines -- and
-- would bloat every list response for the ~all IATAs that don't have one).
-- Served only via GET /api/v1/iatas/{iata}/border, fetched lazily one at a
-- time on client-side region selection.
ALTER TABLE iata_codes ADD COLUMN border JSONB;
db/migrations/021_mv_top_advertisers_route_type.sql
-32
View File
@@ -1,32 +0,0 @@
-- Copyright 2026 Beacon Contributors
-- SPDX-License-Identifier: AGPL-3.0-or-later
-- Splits mv_top_advertisers_by_iata's advert_count into flood- and direct-routed counts,
-- alongside the existing combined total, so /stats/top-advertisers can report how a node is
-- actually being heard rather than only a single blended number.
--
-- Flood = route_type 0 (transport_flood) or 1 (flood) -- broadcast, no prior path needed.
-- Direct = route_type 2 (direct) or 3 (transport_direct) -- routed along a known path.
DROP MATERIALIZED VIEW mv_top_advertisers_by_iata;
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,
COUNT(DISTINCT p.packet_hash) FILTER (WHERE p.route_type IN (0, 1)) AS flood_advert_count,
COUNT(DISTINCT p.packet_hash) FILTER (WHERE p.route_type IN (2, 3)) AS direct_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/migrations/022_neighbor_scopes.sql
-20
View File
@@ -1,20 +0,0 @@
-- Copyright 2026 Beacon Contributors
-- SPDX-License-Identifier: AGPL-3.0-or-later
-- Supports the observer /neighbors MQTT topic: a periodic report listing the
-- OTA-configured region scope of the observer itself ("self") and of each
-- zero-hop neighbor the observer was able to query.
--
-- NOTE: these are plaintext OTA "region scope" strings (comma-separated,
-- e.g. "*" or "US,CA") read from device config over the air. They are
-- unrelated to the transport_scopes / observer_scopes / default_scope_id
-- machinery elsewhere in this schema, which matches encrypted transport
-- packets to named routing scopes by key fingerprint. Deliberately named
-- differently (region_scope, not scope) to avoid confusion between the two.
ALTER TABLE observers ADD COLUMN region_scope TEXT;
-- Per-neighbor-edge, not per-node: the OTA scope query is answered by
-- whichever node reports it, in the context of a specific observer's
-- zero-hop neighbor list, so it lives alongside snr on node_neighbors.
ALTER TABLE node_neighbors ADD COLUMN region_scope TEXT;
db/migrations/023_channel_messages_cascade.sql
-12
View File
@@ -1,12 +0,0 @@
-- Copyright 2026 Beacon Contributors
-- SPDX-License-Identifier: AGPL-3.0-or-later
-- The packet_hash FK had no ON DELETE action, so packets carrying a channel
-- message could never age out and DeleteOldPackets aborted. Match packet_observations.
ALTER TABLE channel_messages
DROP CONSTRAINT IF EXISTS channel_messages_packet_hash_fkey;
ALTER TABLE channel_messages
ADD CONSTRAINT channel_messages_packet_hash_fkey
FOREIGN KEY (packet_hash) REFERENCES packets(packet_hash) ON DELETE CASCADE;
db/migrations/024_known_routes_pathkey.sql
-43
View File
@@ -1,43 +0,0 @@
-- Copyright 2026 Beacon Contributors
-- SPDX-License-Identifier: AGPL-3.0-or-later
-- known_routes grew unbounded (28 GB / 32.8M rows in 24 days on prod) and its
-- UNIQUE(node_ids, iata) key indexed whole UUID arrays. Rebuild keyed on a
-- 16-byte md5 of node_ids and add reconfirm bookkeeping. No rows are pruned
-- here: retention is deployment config (routes.retention/grace), so the first
-- cleanup tick after startup enforces it; a schema migration must not bake in
-- one deployment's policy. id stays (the API serializes it) but loses its
-- index; uniqueness lives on (iata, path_key).
CREATE TABLE known_routes_new (
id BIGINT GENERATED BY DEFAULT AS IDENTITY NOT NULL,
path_key BYTEA NOT NULL,
node_ids UUID[] NOT NULL,
hash_prefix BYTEA[] NOT NULL,
iata CHAR(3) NOT NULL REFERENCES iata_codes(iata) ON DELETE CASCADE,
hop_count INT NOT NULL,
first_seen TIMESTAMPTZ NOT NULL DEFAULT NOW(),
last_seen TIMESTAMPTZ NOT NULL DEFAULT NOW(),
observation_count BIGINT NOT NULL DEFAULT 1,
last_reconfirmed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
PRIMARY KEY (iata, path_key)
);
INSERT INTO known_routes_new
(id, path_key, node_ids, hash_prefix, iata, hop_count, first_seen, last_seen, observation_count)
SELECT id,
decode(md5(array_to_string(node_ids, ',')), 'hex'),
node_ids, hash_prefix, iata, hop_count, first_seen, last_seen, observation_count
FROM known_routes;
SELECT setval(pg_get_serial_sequence('known_routes_new', 'id'),
(SELECT COALESCE(MAX(id), 1) FROM known_routes_new));
DROP TABLE known_routes;
ALTER TABLE known_routes_new RENAME TO known_routes;
ALTER TABLE known_routes RENAME CONSTRAINT known_routes_new_pkey TO known_routes_pkey;
ALTER TABLE known_routes RENAME CONSTRAINT known_routes_new_iata_fkey TO known_routes_iata_fkey;
CREATE INDEX idx_known_routes_hop_count ON known_routes(iata, hop_count);
CREATE INDEX idx_known_routes_last_seen ON known_routes(last_seen DESC);
CREATE INDEX idx_known_routes_reconfirm ON known_routes(last_reconfirmed_at);
db/migrations/026_observation_endpoint_snapshots.sql
-9
View File
@@ -1,9 +0,0 @@
-- Copyright 2026 Beacon Contributors
-- SPDX-License-Identifier: AGPL-3.0-or-later
-- Preserve the endpoint resolution already computed for the live observation.
-- SQL NULL means no endpoint nodes were captured; empty legacy captures also
-- permit detail-time lookup once the node is known (for example, first adverts).
-- No default/backfill: current node metadata cannot reconstruct past names.
ALTER TABLE packet_observations
ADD COLUMN IF NOT EXISTS resolved_endpoints JSONB;
db/migrations/027_known_routes_iata_last_seen.sql
-9
View File
@@ -1,9 +0,0 @@
-- Copyright 2026 Beacon Contributors
-- SPDX-License-Identifier: AGPL-3.0-or-later
-- RunMigrations executes each file outside an explicit transaction. Keep this
-- as one statement so other subscribers can keep writing during the build.
-- If interrupted, check pg_index.indisvalid and remove this named index before
-- retrying; do not silently accept an invalid index with IF NOT EXISTS.
CREATE INDEX CONCURRENTLY idx_known_routes_iata_last_seen
ON known_routes (iata, last_seen DESC);
db/migrations/028_channels_last_seen_id.sql
-8
View File
@@ -1,8 +0,0 @@
-- Copyright 2026 Beacon Contributors
-- SPDX-License-Identifier: AGPL-3.0-or-later
-- RunMigrations executes this single statement outside an explicit transaction.
-- If interrupted, check pg_index.indisvalid and remove this named index before
-- retrying; do not silently accept an invalid index with IF NOT EXISTS.
CREATE INDEX CONCURRENTLY idx_channels_last_seen_id
ON channels (last_seen DESC, id DESC);
db/migrations/029_drop_channels_last_seen.sql
-6
View File
@@ -1,6 +0,0 @@
-- Copyright 2026 Beacon Contributors
-- SPDX-License-Identifier: AGPL-3.0-or-later
-- Migration 028 replaces this index with (last_seen DESC, id DESC).
-- Keep this separate for concurrent execution; IF EXISTS allows a completed drop to retry.
DROP INDEX CONCURRENTLY IF EXISTS idx_channels_last_seen;
db/migrations/030_observation_airtime.sql
-2
View File
@@ -1,2 +0,0 @@
-- Time-on-air per observation so per-observer activity never joins back to packets.
ALTER TABLE packet_observations ADD COLUMN airtime_ms REAL;
db/migrations/031_observation_airtime_backfill.sql
-30
View File
@@ -1,30 +0,0 @@
-- Separate from 030 so the ALTER's ACCESS EXCLUSIVE lock is released before this backfill runs.
-- Backfill only: mirrors internal/lora.TimeOnAirMs (RadioLib getTimeOnAir, CRC on, explicit header).
CREATE FUNCTION lora_airtime_ms_backfill(frame_len int, sf int, bw real, cr int) RETURNS real
LANGUAGE sql IMMUTABLE AS $$
SELECT (
(CASE WHEN sf <= 8 THEN 32 ELSE 16 END) + 4.25
+ 8 + GREATEST(ceil((8*frame_len - 4*sf + 44)::double precision
/ (4*(sf - 2*(CASE WHEN power(2, sf)/bw >= 16 THEN 1 ELSE 0 END)))) * cr, 0)
) * (power(2, sf)/bw)
$$;
-- Last 7 days like 015; parallelism off so the join spills to disk, not /dev/shm.
SET max_parallel_workers_per_gather = 0;
UPDATE packet_observations po
SET airtime_ms = lora_airtime_ms_backfill(
1 + CASE WHEN p.transport_codes_present THEN 4 ELSE 0 END + 1
+ COALESCE(octet_length(po.path_bytes), 0) + octet_length(p.raw_payload),
po.spread_factor, po.bandwidth_khz, po.coding_rate)
FROM packets p
WHERE p.packet_hash = po.packet_hash
AND po.heard_at > NOW() - INTERVAL '7 days'
AND po.airtime_ms IS NULL
AND po.spread_factor BETWEEN 7 AND 12
AND po.bandwidth_khz > 0
AND po.coding_rate BETWEEN 5 AND 8;
RESET max_parallel_workers_per_gather;
DROP FUNCTION lora_airtime_ms_backfill(int, int, real, int);
db/migrations/032_mv_observer_activity.sql
-24
View File
@@ -1,24 +0,0 @@
-- Hourly per-observer rollup so 7d/30d activity sums a few thousand rows instead of scanning
-- a month of observations. payload_type is a key so the breakdown comes from the same rows.
-- sqlc types the cast aggregates as non-null even though airtime_ms/snr_*/rssi_sum are NULL for
-- empty groups, so readers must gate on the matching *_n counts.
CREATE MATERIALIZED VIEW mv_observer_activity_hourly AS
SELECT
observer_id,
payload_type,
date_trunc('hour', heard_at)::timestamptz AS bucket,
COUNT(*)::bigint AS observations,
SUM(airtime_ms)::real AS airtime_ms,
COUNT(airtime_ms)::bigint AS airtime_n,
SUM(snr) FILTER (WHERE NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0))::real AS snr_sum,
COUNT(snr) FILTER (WHERE NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0))::bigint AS snr_n,
MIN(snr) FILTER (WHERE NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0))::real AS snr_min,
SUM(rssi) FILTER (WHERE NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0))::bigint AS rssi_sum,
COUNT(rssi) FILTER (WHERE NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0))::bigint AS rssi_n
FROM packet_observations
WHERE heard_at > NOW() - INTERVAL '30 days'
AND payload_type IS NOT NULL
GROUP BY observer_id, payload_type, date_trunc('hour', heard_at);
CREATE UNIQUE INDEX idx_mv_observer_activity_hourly
ON mv_observer_activity_hourly(observer_id, payload_type, bucket);
db/migrations/033_telemetry_airtime_secs.sql
-4
View File
@@ -1,4 +0,0 @@
-- These always held seconds of radio time from the observer's tx_air_secs/rx_air_secs
-- status fields, never a percentage; name them for what they are.
ALTER TABLE observer_telemetry RENAME COLUMN airtime_tx_pct TO airtime_tx_secs;
ALTER TABLE observer_telemetry RENAME COLUMN airtime_rx_pct TO airtime_rx_secs;
db/migrations/034_accounts.sql
-12
View File
@@ -1,12 +0,0 @@
-- Copyright 2026 Beacon Contributors
-- SPDX-License-Identifier: AGPL-3.0-or-later
CREATE TABLE IF NOT EXISTS accounts (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
name TEXT NOT NULL CHECK (char_length(name) BETWEEN 1 AND 128 AND name = btrim(name)),
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
deactivated_at TIMESTAMPTZ
);
CREATE UNIQUE INDEX IF NOT EXISTS accounts_name_uidx
ON accounts (name) WHERE deactivated_at IS NULL;
db/migrations/035_mv_signal_stats.sql
-33
View File
@@ -1,33 +0,0 @@
-- Reception-IATA rollups include legacy NULL payload types and exclude non-finite
-- SNR. mv_observer_activity_hourly has a different grain/filter, so reusing its
-- means would disagree with these histograms and the reception coverage counts.
CREATE MATERIALIZED VIEW IF NOT EXISTS mv_signal_stats_hourly AS
WITH samples AS (
SELECT iata, date_trunc('hour', heard_at, 'UTC') AS hour,
CASE WHEN NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0)
AND snr > '-Infinity'::real AND snr < 'Infinity'::real
THEN snr::double precision END AS snr,
CASE WHEN NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0)
THEN rssi::double precision END AS rssi
FROM packet_observations
WHERE heard_at >= date_trunc('hour', NOW(), 'UTC') - INTERVAL '720 hours'
AND heard_at < date_trunc('hour', NOW(), 'UTC')
), binned AS (
SELECT *, width_bucket(snr, -30, 30, 12) AS snr_bin,
width_bucket(rssi, -140, 0, 14) AS rssi_bin
FROM samples
)
SELECT iata, hour, grouping(snr_bin, rssi_bin)::integer AS kind,
COALESCE(snr_bin, -1)::integer AS snr_bin,
COALESCE(rssi_bin, -1)::integer AS rssi_bin,
count(*)::bigint AS receptions,
count(snr)::bigint AS snr_samples,
COALESCE(sum(snr), 0)::double precision AS snr_sum,
count(rssi)::bigint AS rssi_samples,
COALESCE(sum(rssi), 0)::double precision AS rssi_sum
FROM binned
GROUP BY GROUPING SETS ((iata, hour), (iata, hour, snr_bin), (iata, hour, rssi_bin));
CREATE UNIQUE INDEX IF NOT EXISTS idx_mv_signal_stats_hourly
ON mv_signal_stats_hourly (iata, hour, kind, snr_bin, rssi_bin);
CREATE INDEX IF NOT EXISTS idx_mv_signal_stats_hourly_hour ON mv_signal_stats_hourly (hour);
db/migrations/036_mv_path_stats.sql
-31
View File
@@ -1,31 +0,0 @@
-- Classify once per background refresh, preserving the decoder's header rules.
-- No observation rows or deduplication constraints are changed.
CREATE MATERIALIZED VIEW IF NOT EXISTS mv_path_stats_hourly AS
WITH classified AS (
SELECT iata, heard_at, hash_size, hop_count,
CASE WHEN payload_type = 9 THEN 2
-- Match meshcore-go IsValidPathLen (1/2/3-byte hashes, max 64 path bytes).
WHEN payload_type IS NULL OR payload_type NOT BETWEEN 0 AND 15
OR NOT (path_length_byte BETWEEN 0 AND 191
AND hash_size BETWEEN 1 AND 3 AND hop_count BETWEEN 0 AND 63
AND hash_size = (path_length_byte >> 6) + 1
AND hop_count = (path_length_byte & 63)
AND hash_size::integer * hop_count::integer <= 64
AND COALESCE(octet_length(path_bytes), 0) = hash_size::integer * hop_count::integer)
THEN 3
WHEN hop_count = 0 THEN 1
ELSE 0 END::integer AS category
FROM packet_observations
WHERE heard_at >= date_trunc('hour', NOW(), 'UTC') - INTERVAL '720 hours'
AND heard_at < date_trunc('hour', NOW(), 'UTC')
), buckets AS (
SELECT iata, date_trunc('hour', heard_at, 'UTC') AS hour, category,
CASE WHEN category = 0 THEN hash_size ELSE 0 END::integer AS hash_bytes,
CASE WHEN category = 0 THEN hop_count ELSE 0 END::integer AS entries
FROM classified
)
SELECT iata, hour, category, hash_bytes, entries, count(*)::bigint AS receptions
FROM buckets GROUP BY iata, hour, category, hash_bytes, entries;
CREATE UNIQUE INDEX IF NOT EXISTS idx_mv_path_stats_hourly
ON mv_path_stats_hourly (iata, hour, category, hash_bytes, entries);
CREATE INDEX IF NOT EXISTS idx_mv_path_stats_hourly_hour ON mv_path_stats_hourly (hour);
db/migrations/037_autovacuum_tuning.sql
-14
View File
@@ -1,14 +0,0 @@
-- Copyright 2026 Beacon Contributors
-- SPDX-License-Identifier: AGPL-3.0-or-later
-- Default 20% trigger never fired on known_routes, bloating its indexes.
ALTER TABLE known_routes SET (
autovacuum_vacuum_scale_factor = 0.02,
autovacuum_analyze_scale_factor = 0.02,
autovacuum_vacuum_cost_limit = 1000
);
ALTER TABLE packets SET (autovacuum_vacuum_scale_factor = 0.05, autovacuum_vacuum_cost_limit = 1000);
ALTER TABLE packet_observations SET (autovacuum_vacuum_scale_factor = 0.05, autovacuum_vacuum_cost_limit = 1000);
ALTER TABLE nodes SET (autovacuum_vacuum_scale_factor = 0.05);
ALTER TABLE node_iatas SET (autovacuum_vacuum_scale_factor = 0.05);
ALTER TABLE node_short_ids SET (autovacuum_vacuum_scale_factor = 0.05);
db/migrations/038_drop_resolved_endpoints.sql
-6
View File
@@ -1,6 +0,0 @@
-- Copyright 2026 Beacon Contributors
-- SPDX-License-Identifier: AGPL-3.0-or-later
-- Endpoints now resolve at read time like path hops. The per-observation
-- snapshots were ~60% of packet_observations on prod.
ALTER TABLE packet_observations DROP COLUMN IF EXISTS resolved_endpoints;
db/migrations/039_analytics_retention.sql
-613
View File
@@ -1,613 +0,0 @@
-- Copyright 2026 Beacon Contributors
-- SPDX-License-Identifier: AGPL-3.0-or-later
-- Archive only aggregates of deleted packet cohorts, never packet/message bodies
-- or raw paths. Live and archived cohorts are disjoint in every refresh snapshot.
-- Thirty days matches the longest analytics window; raw retention is independent.
CREATE TABLE IF NOT EXISTS analytics_hourly_iata_stats (
iata char(3),
hour timestamptz,
observation_count bigint NOT NULL DEFAULT 0,
unique_packets bigint NOT NULL DEFAULT 0,
PRIMARY KEY (iata,hour)
);
CREATE INDEX IF NOT EXISTS idx_analytics_hourly_iata_stats_time ON analytics_hourly_iata_stats (hour);
CREATE TABLE IF NOT EXISTS analytics_payload_breakdown_by_iata (
iata char(3),
payload_type smallint,
bucket timestamptz,
count bigint NOT NULL DEFAULT 0,
PRIMARY KEY (iata,payload_type,bucket)
);
CREATE INDEX IF NOT EXISTS idx_analytics_payload_breakdown_by_iata_time ON analytics_payload_breakdown_by_iata (bucket);
CREATE TABLE IF NOT EXISTS analytics_top_observers_by_iata (
iata char(3),
observer_id uuid,
bucket timestamptz,
observation_count bigint NOT NULL DEFAULT 0,
display_name text,
observer_type text,
PRIMARY KEY (iata,observer_id,bucket)
);
CREATE INDEX IF NOT EXISTS idx_analytics_top_observers_by_iata_time ON analytics_top_observers_by_iata (bucket);
CREATE TABLE IF NOT EXISTS analytics_top_talkers_by_iata (
iata char(3),
sender_name text,
bucket timestamptz,
message_count bigint NOT NULL DEFAULT 0,
last_sent timestamptz,
PRIMARY KEY (iata,sender_name,bucket)
);
CREATE INDEX IF NOT EXISTS idx_analytics_top_talkers_by_iata_time ON analytics_top_talkers_by_iata (bucket);
CREATE TABLE IF NOT EXISTS analytics_top_advertisers_by_iata (
iata char(3),
node_id uuid,
bucket timestamptz,
advert_count bigint NOT NULL DEFAULT 0,
flood_advert_count bigint NOT NULL DEFAULT 0,
direct_advert_count bigint NOT NULL DEFAULT 0,
last_heard timestamptz,
name text,
node_type smallint,
PRIMARY KEY (iata,node_id,bucket)
);
CREATE INDEX IF NOT EXISTS idx_analytics_top_advertisers_by_iata_time ON analytics_top_advertisers_by_iata (bucket);
CREATE TABLE IF NOT EXISTS analytics_observer_activity_hourly (
observer_id uuid,
payload_type smallint,
bucket timestamptz,
observations bigint NOT NULL DEFAULT 0,
airtime_ms real,
airtime_n bigint NOT NULL DEFAULT 0,
snr_sum real,
snr_n bigint NOT NULL DEFAULT 0,
snr_min real,
rssi_sum bigint,
rssi_n bigint NOT NULL DEFAULT 0,
PRIMARY KEY (observer_id,payload_type,bucket)
);
CREATE INDEX IF NOT EXISTS idx_analytics_observer_activity_hourly_time ON analytics_observer_activity_hourly (bucket);
CREATE TABLE IF NOT EXISTS analytics_signal_stats_hourly (
iata char(3),
hour timestamptz,
kind integer,
snr_bin integer,
rssi_bin integer,
receptions bigint NOT NULL DEFAULT 0,
snr_samples bigint NOT NULL DEFAULT 0,
snr_sum double precision,
rssi_samples bigint NOT NULL DEFAULT 0,
rssi_sum double precision,
PRIMARY KEY (iata,hour,kind,snr_bin,rssi_bin)
);
CREATE INDEX IF NOT EXISTS idx_analytics_signal_stats_hourly_time ON analytics_signal_stats_hourly (hour);
CREATE TABLE IF NOT EXISTS analytics_path_stats_hourly (
iata char(3),
hour timestamptz,
category integer,
hash_bytes integer,
entries integer,
receptions bigint NOT NULL DEFAULT 0,
PRIMARY KEY (iata,hour,category,hash_bytes,entries)
);
CREATE INDEX IF NOT EXISTS idx_analytics_path_stats_hourly_time ON analytics_path_stats_hourly (hour);
DROP MATERIALIZED VIEW IF EXISTS mv_hourly_iata_stats;
CREATE OR REPLACE VIEW analytics_live_hourly_iata_stats AS
SELECT
iata,
date_trunc('hour', heard_at, 'UTC')::timestamptz AS hour,
COUNT(*) AS observation_count,
COUNT(DISTINCT packet_hash) AS unique_packets
FROM packet_observations
WHERE heard_at > NOW() - INTERVAL '30 days'
GROUP BY iata, date_trunc('hour', heard_at, 'UTC');
CREATE MATERIALIZED VIEW mv_hourly_iata_stats AS
WITH combined AS (
SELECT iata, hour, observation_count, unique_packets FROM analytics_live_hourly_iata_stats
UNION ALL
SELECT iata, hour, observation_count, unique_packets FROM analytics_hourly_iata_stats
WHERE hour >= date_trunc('hour', NOW(), 'UTC') - INTERVAL '30 days'
), observers_by_hour AS (
SELECT iata, date_trunc('hour', heard_at, 'UTC') AS hour, observer_id
FROM packet_observations WHERE heard_at > NOW() - INTERVAL '30 days'
UNION
SELECT iata, bucket AS hour, observer_id FROM analytics_top_observers_by_iata
WHERE bucket >= date_trunc('hour', NOW(), 'UTC') - INTERVAL '30 days'
), observer_counts AS (SELECT iata, hour, COUNT(*) AS n FROM observers_by_hour GROUP BY iata,hour)
SELECT combined.iata, combined.hour, SUM(observation_count)::bigint AS observation_count, SUM(unique_packets)::bigint AS unique_packets, MAX(o.n)::bigint AS active_observers
FROM combined JOIN observer_counts o USING (iata,hour) GROUP BY combined.iata,combined.hour;
CREATE UNIQUE INDEX idx_analytics_hourly_iata_stats_view ON mv_hourly_iata_stats (iata,hour);
DROP MATERIALIZED VIEW IF EXISTS mv_payload_breakdown_by_iata;
CREATE OR REPLACE VIEW analytics_live_payload_breakdown_by_iata AS
SELECT
iata,
payload_type,
date_trunc('hour', heard_at, 'UTC')::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, 'UTC');
CREATE MATERIALIZED VIEW mv_payload_breakdown_by_iata AS
WITH combined AS (
SELECT iata, payload_type, bucket, count FROM analytics_live_payload_breakdown_by_iata
UNION ALL
SELECT iata, payload_type, bucket, count FROM analytics_payload_breakdown_by_iata
WHERE bucket >= date_trunc('hour', NOW(), 'UTC') - INTERVAL '30 days'
)
SELECT iata, payload_type, bucket, SUM(count)::bigint AS count
FROM combined GROUP BY iata,payload_type,bucket;
CREATE UNIQUE INDEX idx_analytics_payload_breakdown_by_iata_view ON mv_payload_breakdown_by_iata (iata,payload_type,bucket);
DROP MATERIALIZED VIEW IF EXISTS mv_top_observers_by_iata;
CREATE OR REPLACE VIEW analytics_live_top_observers_by_iata AS
SELECT
po.iata,
po.observer_id,
o.display_name,
o.observer_type,
date_trunc('hour', po.heard_at, 'UTC')::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, 'UTC');
CREATE MATERIALIZED VIEW mv_top_observers_by_iata AS
WITH combined AS (
SELECT iata, observer_id, bucket, observation_count, display_name, observer_type FROM analytics_live_top_observers_by_iata
UNION ALL
SELECT iata, observer_id, bucket, observation_count, display_name, observer_type FROM analytics_top_observers_by_iata
WHERE bucket >= date_trunc('hour', NOW(), 'UTC') - INTERVAL '30 days'
), totals AS (SELECT iata, observer_id, bucket, SUM(observation_count)::bigint AS observation_count, MAX(display_name) AS display_name, MAX(observer_type) AS observer_type
FROM combined GROUP BY iata,observer_id,bucket)
SELECT t.iata, t.observer_id, COALESCE(o.display_name,t.display_name) AS display_name,
COALESCE(o.observer_type,t.observer_type) AS observer_type, t.bucket, t.observation_count
FROM totals t LEFT JOIN observers o ON o.id=t.observer_id;
CREATE UNIQUE INDEX idx_analytics_top_observers_by_iata_view ON mv_top_observers_by_iata (iata,observer_id,bucket);
DROP MATERIALIZED VIEW IF EXISTS mv_top_talkers_by_iata;
CREATE OR REPLACE VIEW analytics_live_top_talkers_by_iata AS
SELECT
po.iata,
cm.sender_name,
date_trunc('hour', cm.sent_at, 'UTC')::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, 'UTC');
CREATE MATERIALIZED VIEW mv_top_talkers_by_iata AS
WITH combined AS (
SELECT iata, sender_name, bucket, message_count, last_sent FROM analytics_live_top_talkers_by_iata
UNION ALL
SELECT iata, sender_name, bucket, message_count, last_sent FROM analytics_top_talkers_by_iata
WHERE bucket >= date_trunc('hour', NOW(), 'UTC') - INTERVAL '30 days'
)
SELECT iata, sender_name, bucket, SUM(message_count)::bigint AS message_count, MAX(last_sent) AS last_sent
FROM combined GROUP BY iata,sender_name,bucket;
CREATE UNIQUE INDEX idx_analytics_top_talkers_by_iata_view ON mv_top_talkers_by_iata (iata,sender_name,bucket);
DROP MATERIALIZED VIEW IF EXISTS mv_top_advertisers_by_iata;
CREATE OR REPLACE VIEW analytics_live_top_advertisers_by_iata AS
SELECT
po.iata,
n.id AS node_id,
n.name,
n.node_type,
date_trunc('hour', po.heard_at, 'UTC')::timestamptz AS bucket,
COUNT(DISTINCT p.packet_hash) AS advert_count,
COUNT(DISTINCT p.packet_hash) FILTER (WHERE p.route_type IN (0, 1)) AS flood_advert_count,
COUNT(DISTINCT p.packet_hash) FILTER (WHERE p.route_type IN (2, 3)) AS direct_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, 'UTC');
CREATE MATERIALIZED VIEW mv_top_advertisers_by_iata AS
WITH combined AS (
SELECT iata, node_id, bucket, advert_count, flood_advert_count, direct_advert_count, last_heard, name, node_type FROM analytics_live_top_advertisers_by_iata
UNION ALL
SELECT iata, node_id, bucket, advert_count, flood_advert_count, direct_advert_count, last_heard, name, node_type FROM analytics_top_advertisers_by_iata
WHERE bucket >= date_trunc('hour', NOW(), 'UTC') - INTERVAL '30 days'
), totals AS (SELECT iata, node_id, bucket, SUM(advert_count)::bigint AS advert_count, SUM(flood_advert_count)::bigint AS flood_advert_count, SUM(direct_advert_count)::bigint AS direct_advert_count, MAX(last_heard) AS last_heard, MAX(name) AS name, MAX(node_type) AS node_type
FROM combined GROUP BY iata,node_id,bucket)
SELECT t.iata, t.node_id, COALESCE(n.name,t.name) AS name, COALESCE(n.node_type,t.node_type) AS node_type,
t.bucket, t.advert_count, t.flood_advert_count, t.direct_advert_count, t.last_heard
FROM totals t LEFT JOIN nodes n ON n.id=t.node_id;
CREATE UNIQUE INDEX idx_analytics_top_advertisers_by_iata_view ON mv_top_advertisers_by_iata (iata,node_id,bucket);
DROP MATERIALIZED VIEW IF EXISTS mv_observer_activity_hourly;
CREATE OR REPLACE VIEW analytics_live_observer_activity_hourly AS
SELECT
observer_id,
COALESCE(payload_type, -1)::smallint AS payload_type,
date_trunc('hour', heard_at, 'UTC')::timestamptz AS bucket,
COUNT(*)::bigint AS observations,
SUM(airtime_ms)::real AS airtime_ms,
COUNT(airtime_ms)::bigint AS airtime_n,
SUM(snr) FILTER (WHERE NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0))::real AS snr_sum,
COUNT(snr) FILTER (WHERE NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0))::bigint AS snr_n,
MIN(snr) FILTER (WHERE NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0))::real AS snr_min,
SUM(rssi) FILTER (WHERE NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0))::bigint AS rssi_sum,
COUNT(rssi) FILTER (WHERE NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0))::bigint AS rssi_n
FROM packet_observations
WHERE heard_at > NOW() - INTERVAL '30 days'
GROUP BY observer_id, COALESCE(payload_type, -1), date_trunc('hour', heard_at, 'UTC');
CREATE MATERIALIZED VIEW mv_observer_activity_hourly AS
WITH combined AS (
SELECT observer_id, payload_type, bucket, observations, airtime_ms, airtime_n, snr_sum, snr_n, snr_min, rssi_sum, rssi_n FROM analytics_live_observer_activity_hourly
UNION ALL
SELECT observer_id, payload_type, bucket, observations, airtime_ms, airtime_n, snr_sum, snr_n, snr_min, rssi_sum, rssi_n FROM analytics_observer_activity_hourly
WHERE bucket >= date_trunc('hour', NOW(), 'UTC') - INTERVAL '30 days'
)
SELECT observer_id, payload_type, bucket, SUM(observations)::bigint AS observations, SUM(airtime_ms)::real AS airtime_ms, SUM(airtime_n)::bigint AS airtime_n, SUM(snr_sum)::real AS snr_sum, SUM(snr_n)::bigint AS snr_n, MIN(snr_min)::real AS snr_min, SUM(rssi_sum)::bigint AS rssi_sum, SUM(rssi_n)::bigint AS rssi_n
FROM combined GROUP BY observer_id,payload_type,bucket;
CREATE UNIQUE INDEX idx_analytics_observer_activity_hourly_view ON mv_observer_activity_hourly (observer_id,payload_type,bucket);
DROP MATERIALIZED VIEW IF EXISTS mv_signal_stats_hourly;
CREATE OR REPLACE VIEW analytics_live_signal_stats_hourly AS
WITH samples AS (
SELECT iata, date_trunc('hour', heard_at, 'UTC') AS hour,
CASE WHEN NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0)
AND snr > '-Infinity'::real AND snr < 'Infinity'::real
THEN snr::double precision END AS snr,
CASE WHEN NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0)
THEN rssi::double precision END AS rssi
FROM packet_observations
WHERE heard_at >= date_trunc('hour', NOW(), 'UTC') - INTERVAL '720 hours'
AND heard_at < date_trunc('hour', NOW(), 'UTC')
), binned AS (
SELECT *, width_bucket(snr, -30, 30, 12) AS snr_bin,
width_bucket(rssi, -140, 0, 14) AS rssi_bin
FROM samples
)
SELECT iata, hour, grouping(snr_bin, rssi_bin)::integer AS kind,
COALESCE(snr_bin, -1)::integer AS snr_bin,
COALESCE(rssi_bin, -1)::integer AS rssi_bin,
count(*)::bigint AS receptions,
count(snr)::bigint AS snr_samples,
COALESCE(sum(snr), 0)::double precision AS snr_sum,
count(rssi)::bigint AS rssi_samples,
COALESCE(sum(rssi), 0)::double precision AS rssi_sum
FROM binned
GROUP BY GROUPING SETS ((iata, hour), (iata, hour, snr_bin), (iata, hour, rssi_bin));
CREATE MATERIALIZED VIEW mv_signal_stats_hourly AS
WITH combined AS (
SELECT iata, hour, kind, snr_bin, rssi_bin, receptions, snr_samples, snr_sum, rssi_samples, rssi_sum FROM analytics_live_signal_stats_hourly
UNION ALL
SELECT iata, hour, kind, snr_bin, rssi_bin, receptions, snr_samples, snr_sum, rssi_samples, rssi_sum FROM analytics_signal_stats_hourly
WHERE hour >= date_trunc('hour', NOW(), 'UTC') - INTERVAL '30 days'
AND hour < date_trunc('hour', NOW(), 'UTC')
)
SELECT iata, hour, kind, snr_bin, rssi_bin, SUM(receptions)::bigint AS receptions, SUM(snr_samples)::bigint AS snr_samples, SUM(snr_sum)::double precision AS snr_sum, SUM(rssi_samples)::bigint AS rssi_samples, SUM(rssi_sum)::double precision AS rssi_sum
FROM combined GROUP BY iata,hour,kind,snr_bin,rssi_bin;
CREATE UNIQUE INDEX idx_analytics_signal_stats_hourly_view ON mv_signal_stats_hourly (iata,hour,kind,snr_bin,rssi_bin);
CREATE INDEX idx_mv_signal_stats_hourly_hour ON mv_signal_stats_hourly (hour);
DROP MATERIALIZED VIEW IF EXISTS mv_path_stats_hourly;
CREATE OR REPLACE VIEW analytics_live_path_stats_hourly AS
WITH classified AS (
SELECT iata, heard_at, hash_size, hop_count,
CASE WHEN payload_type = 9 THEN 2
-- Match meshcore-go IsValidPathLen (1/2/3-byte hashes, max 64 path bytes).
WHEN payload_type IS NULL OR payload_type NOT BETWEEN 0 AND 15
OR NOT (path_length_byte BETWEEN 0 AND 191
AND hash_size BETWEEN 1 AND 3 AND hop_count BETWEEN 0 AND 63
AND hash_size = (path_length_byte >> 6) + 1
AND hop_count = (path_length_byte & 63)
AND hash_size::integer * hop_count::integer <= 64
AND COALESCE(octet_length(path_bytes), 0) = hash_size::integer * hop_count::integer)
THEN 3
WHEN hop_count = 0 THEN 1
ELSE 0 END::integer AS category
FROM packet_observations
WHERE heard_at >= date_trunc('hour', NOW(), 'UTC') - INTERVAL '720 hours'
AND heard_at < date_trunc('hour', NOW(), 'UTC')
), buckets AS (
SELECT iata, date_trunc('hour', heard_at, 'UTC') AS hour, category,
CASE WHEN category = 0 THEN hash_size ELSE 0 END::integer AS hash_bytes,
CASE WHEN category = 0 THEN hop_count ELSE 0 END::integer AS entries
FROM classified
)
SELECT iata, hour, category, hash_bytes, entries, count(*)::bigint AS receptions
FROM buckets GROUP BY iata, hour, category, hash_bytes, entries;
CREATE MATERIALIZED VIEW mv_path_stats_hourly AS
WITH combined AS (
SELECT iata, hour, category, hash_bytes, entries, receptions FROM analytics_live_path_stats_hourly
UNION ALL
SELECT iata, hour, category, hash_bytes, entries, receptions FROM analytics_path_stats_hourly
WHERE hour >= date_trunc('hour', NOW(), 'UTC') - INTERVAL '30 days'
AND hour < date_trunc('hour', NOW(), 'UTC')
)
SELECT iata, hour, category, hash_bytes, entries, SUM(receptions)::bigint AS receptions
FROM combined GROUP BY iata,hour,category,hash_bytes,entries;
CREATE UNIQUE INDEX idx_analytics_path_stats_hourly_view ON mv_path_stats_hourly (iata,hour,category,hash_bytes,entries);
CREATE INDEX idx_mv_path_stats_hourly_hour ON mv_path_stats_hourly (hour);
-- VOLATILE gives the archive statement a new READ COMMITTED snapshot after
-- acquiring parent locks. FK inserts cannot race between that snapshot and the
-- cascade. SKIP LOCKED leaves packets being ingested for a later cleanup tick.
CREATE OR REPLACE FUNCTION archive_delete_packets(cutoff timestamptz, batch_size integer)
RETURNS bigint LANGUAGE plpgsql VOLATILE AS $$
DECLARE hashes bytea[]; deleted bigint;
BEGIN
-- Serialize cleanup runners before overlapping aggregate upserts.
PERFORM pg_advisory_xact_lock(hashtext('beacon.archive_delete_packets'));
IF batch_size < 1 THEN RAISE EXCEPTION 'batch_size must be positive'; END IF;
SELECT array_agg(p.packet_hash) INTO hashes FROM (
SELECT ep.packet_hash FROM packets ep
WHERE ep.last_heard_at < cutoff
ORDER BY ep.last_heard_at, ep.packet_hash
LIMIT batch_size FOR UPDATE OF ep SKIP LOCKED
) p;
WITH expired_observations AS MATERIALIZED (
SELECT po.* FROM packet_observations po WHERE po.packet_hash = ANY(hashes)
),
archived_hourly_iata_stats AS (
INSERT INTO analytics_hourly_iata_stats (iata, hour, observation_count, unique_packets)
SELECT iata, hour, observation_count, unique_packets FROM (
SELECT
iata,
date_trunc('hour', heard_at, 'UTC')::timestamptz AS hour,
COUNT(*) AS observation_count,
COUNT(DISTINCT packet_hash) AS unique_packets
FROM expired_observations
WHERE heard_at >= date_trunc('hour', NOW(), 'UTC') - INTERVAL '30 days'
GROUP BY iata, date_trunc('hour', heard_at, 'UTC')
) batch
ON CONFLICT (iata,hour) DO UPDATE SET
observation_count = analytics_hourly_iata_stats.observation_count + EXCLUDED.observation_count,
unique_packets = analytics_hourly_iata_stats.unique_packets + EXCLUDED.unique_packets
),
archived_payload_breakdown_by_iata AS (
INSERT INTO analytics_payload_breakdown_by_iata (iata, payload_type, bucket, count)
SELECT iata, payload_type, bucket, count FROM (
SELECT
iata,
payload_type,
date_trunc('hour', heard_at, 'UTC')::timestamptz AS bucket,
COUNT(*) AS count
FROM expired_observations
WHERE heard_at >= date_trunc('hour', NOW(), 'UTC') - INTERVAL '30 days'
AND payload_type IS NOT NULL
GROUP BY iata, payload_type, date_trunc('hour', heard_at, 'UTC')
) batch
ON CONFLICT (iata,payload_type,bucket) DO UPDATE SET
count = analytics_payload_breakdown_by_iata.count + EXCLUDED.count
),
archived_top_observers_by_iata AS (
INSERT INTO analytics_top_observers_by_iata (iata, observer_id, bucket, observation_count, display_name, observer_type)
SELECT iata, observer_id, bucket, observation_count, display_name, observer_type FROM (
SELECT
po.iata,
po.observer_id,
o.display_name,
o.observer_type,
date_trunc('hour', po.heard_at, 'UTC')::timestamptz AS bucket,
COUNT(*) AS observation_count
FROM expired_observations po
JOIN observers o ON o.id = po.observer_id
WHERE po.heard_at >= date_trunc('hour', NOW(), 'UTC') - INTERVAL '30 days'
GROUP BY po.iata, po.observer_id, o.display_name, o.observer_type, date_trunc('hour', po.heard_at, 'UTC')
) batch
ON CONFLICT (iata,observer_id,bucket) DO UPDATE SET
observation_count = analytics_top_observers_by_iata.observation_count + EXCLUDED.observation_count,
display_name = EXCLUDED.display_name,
observer_type = EXCLUDED.observer_type
),
archived_top_talkers_by_iata AS (
INSERT INTO analytics_top_talkers_by_iata (iata, sender_name, bucket, message_count, last_sent)
SELECT iata, sender_name, bucket, message_count, last_sent FROM (
SELECT
po.iata,
cm.sender_name,
date_trunc('hour', cm.sent_at, 'UTC')::timestamptz AS bucket,
COUNT(DISTINCT cm.id) AS message_count,
MAX(cm.sent_at) AS last_sent
FROM channel_messages cm
JOIN expired_observations po ON po.packet_hash = cm.packet_hash
WHERE cm.sender_name IS NOT NULL
AND cm.sent_at >= date_trunc('hour', NOW(), 'UTC') - INTERVAL '30 days'
GROUP BY po.iata, cm.sender_name, date_trunc('hour', cm.sent_at, 'UTC')
) batch
ON CONFLICT (iata,sender_name,bucket) DO UPDATE SET
message_count = analytics_top_talkers_by_iata.message_count + EXCLUDED.message_count,
last_sent = GREATEST(analytics_top_talkers_by_iata.last_sent, EXCLUDED.last_sent)
),
archived_top_advertisers_by_iata AS (
INSERT INTO analytics_top_advertisers_by_iata (iata, node_id, bucket, advert_count, flood_advert_count, direct_advert_count, last_heard, name, node_type)
SELECT iata, node_id, bucket, advert_count, flood_advert_count, direct_advert_count, last_heard, name, node_type FROM (
SELECT
po.iata,
n.id AS node_id,
n.name,
n.node_type,
date_trunc('hour', po.heard_at, 'UTC')::timestamptz AS bucket,
COUNT(DISTINCT p.packet_hash) AS advert_count,
COUNT(DISTINCT p.packet_hash) FILTER (WHERE p.route_type IN (0, 1)) AS flood_advert_count,
COUNT(DISTINCT p.packet_hash) FILTER (WHERE p.route_type IN (2, 3)) AS direct_advert_count,
MAX(po.heard_at) AS last_heard
FROM packets p
JOIN expired_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 >= date_trunc('hour', NOW(), 'UTC') - INTERVAL '30 days'
GROUP BY po.iata, n.id, n.name, n.node_type, date_trunc('hour', po.heard_at, 'UTC')
) batch
ON CONFLICT (iata,node_id,bucket) DO UPDATE SET
advert_count = analytics_top_advertisers_by_iata.advert_count + EXCLUDED.advert_count,
flood_advert_count = analytics_top_advertisers_by_iata.flood_advert_count + EXCLUDED.flood_advert_count,
direct_advert_count = analytics_top_advertisers_by_iata.direct_advert_count + EXCLUDED.direct_advert_count,
last_heard = GREATEST(analytics_top_advertisers_by_iata.last_heard, EXCLUDED.last_heard),
name = EXCLUDED.name,
node_type = EXCLUDED.node_type
),
archived_observer_activity_hourly AS (
INSERT INTO analytics_observer_activity_hourly (observer_id, payload_type, bucket, observations, airtime_ms, airtime_n, snr_sum, snr_n, snr_min, rssi_sum, rssi_n)
SELECT observer_id, payload_type, bucket, observations, airtime_ms, airtime_n, snr_sum, snr_n, snr_min, rssi_sum, rssi_n FROM (
SELECT
observer_id,
COALESCE(payload_type, -1)::smallint AS payload_type,
date_trunc('hour', heard_at, 'UTC')::timestamptz AS bucket,
COUNT(*)::bigint AS observations,
SUM(airtime_ms)::real AS airtime_ms,
COUNT(airtime_ms)::bigint AS airtime_n,
SUM(snr) FILTER (WHERE NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0))::real AS snr_sum,
COUNT(snr) FILTER (WHERE NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0))::bigint AS snr_n,
MIN(snr) FILTER (WHERE NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0))::real AS snr_min,
SUM(rssi) FILTER (WHERE NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0))::bigint AS rssi_sum,
COUNT(rssi) FILTER (WHERE NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0))::bigint AS rssi_n
FROM expired_observations
WHERE heard_at >= date_trunc('hour', NOW(), 'UTC') - INTERVAL '30 days'
GROUP BY observer_id, COALESCE(payload_type, -1), date_trunc('hour', heard_at, 'UTC')
) batch
ON CONFLICT (observer_id,payload_type,bucket) DO UPDATE SET
observations = analytics_observer_activity_hourly.observations + EXCLUDED.observations,
airtime_ms = CASE WHEN analytics_observer_activity_hourly.airtime_ms IS NULL AND EXCLUDED.airtime_ms IS NULL THEN NULL ELSE COALESCE(analytics_observer_activity_hourly.airtime_ms, 0) + COALESCE(EXCLUDED.airtime_ms, 0) END,
airtime_n = analytics_observer_activity_hourly.airtime_n + EXCLUDED.airtime_n,
snr_sum = CASE WHEN analytics_observer_activity_hourly.snr_sum IS NULL AND EXCLUDED.snr_sum IS NULL THEN NULL ELSE COALESCE(analytics_observer_activity_hourly.snr_sum, 0) + COALESCE(EXCLUDED.snr_sum, 0) END,
snr_n = analytics_observer_activity_hourly.snr_n + EXCLUDED.snr_n,
snr_min = LEAST(analytics_observer_activity_hourly.snr_min, EXCLUDED.snr_min),
rssi_sum = CASE WHEN analytics_observer_activity_hourly.rssi_sum IS NULL AND EXCLUDED.rssi_sum IS NULL THEN NULL ELSE COALESCE(analytics_observer_activity_hourly.rssi_sum, 0) + COALESCE(EXCLUDED.rssi_sum, 0) END,
rssi_n = analytics_observer_activity_hourly.rssi_n + EXCLUDED.rssi_n
),
archived_signal_stats_hourly AS (
INSERT INTO analytics_signal_stats_hourly (iata, hour, kind, snr_bin, rssi_bin, receptions, snr_samples, snr_sum, rssi_samples, rssi_sum)
SELECT iata, hour, kind, snr_bin, rssi_bin, receptions, snr_samples, snr_sum, rssi_samples, rssi_sum FROM (
WITH samples AS (
SELECT iata, date_trunc('hour', heard_at, 'UTC') AS hour,
CASE WHEN NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0)
AND snr > '-Infinity'::real AND snr < 'Infinity'::real
THEN snr::double precision END AS snr,
CASE WHEN NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0)
THEN rssi::double precision END AS rssi
FROM expired_observations
WHERE heard_at >= date_trunc('hour', NOW(), 'UTC') - INTERVAL '720 hours'
), binned AS (
SELECT *, width_bucket(snr, -30, 30, 12) AS snr_bin,
width_bucket(rssi, -140, 0, 14) AS rssi_bin
FROM samples
)
SELECT iata, hour, grouping(snr_bin, rssi_bin)::integer AS kind,
COALESCE(snr_bin, -1)::integer AS snr_bin,
COALESCE(rssi_bin, -1)::integer AS rssi_bin,
count(*)::bigint AS receptions,
count(snr)::bigint AS snr_samples,
COALESCE(sum(snr), 0)::double precision AS snr_sum,
count(rssi)::bigint AS rssi_samples,
COALESCE(sum(rssi), 0)::double precision AS rssi_sum
FROM binned
GROUP BY GROUPING SETS ((iata, hour), (iata, hour, snr_bin), (iata, hour, rssi_bin))
) batch
ON CONFLICT (iata,hour,kind,snr_bin,rssi_bin) DO UPDATE SET
receptions = analytics_signal_stats_hourly.receptions + EXCLUDED.receptions,
snr_samples = analytics_signal_stats_hourly.snr_samples + EXCLUDED.snr_samples,
snr_sum = analytics_signal_stats_hourly.snr_sum + EXCLUDED.snr_sum,
rssi_samples = analytics_signal_stats_hourly.rssi_samples + EXCLUDED.rssi_samples,
rssi_sum = analytics_signal_stats_hourly.rssi_sum + EXCLUDED.rssi_sum
),
archived_path_stats_hourly AS (
INSERT INTO analytics_path_stats_hourly (iata, hour, category, hash_bytes, entries, receptions)
SELECT iata, hour, category, hash_bytes, entries, receptions FROM (
WITH classified AS (
SELECT iata, heard_at, hash_size, hop_count,
CASE WHEN payload_type = 9 THEN 2
-- Match meshcore-go IsValidPathLen (1/2/3-byte hashes, max 64 path bytes).
WHEN payload_type IS NULL OR payload_type NOT BETWEEN 0 AND 15
OR NOT (path_length_byte BETWEEN 0 AND 191
AND hash_size BETWEEN 1 AND 3 AND hop_count BETWEEN 0 AND 63
AND hash_size = (path_length_byte >> 6) + 1
AND hop_count = (path_length_byte & 63)
AND hash_size::integer * hop_count::integer <= 64
AND COALESCE(octet_length(path_bytes), 0) = hash_size::integer * hop_count::integer)
THEN 3
WHEN hop_count = 0 THEN 1
ELSE 0 END::integer AS category
FROM expired_observations
WHERE heard_at >= date_trunc('hour', NOW(), 'UTC') - INTERVAL '720 hours'
), buckets AS (
SELECT iata, date_trunc('hour', heard_at, 'UTC') AS hour, category,
CASE WHEN category = 0 THEN hash_size ELSE 0 END::integer AS hash_bytes,
CASE WHEN category = 0 THEN hop_count ELSE 0 END::integer AS entries
FROM classified
)
SELECT iata, hour, category, hash_bytes, entries, count(*)::bigint AS receptions
FROM buckets GROUP BY iata, hour, category, hash_bytes, entries
) batch
ON CONFLICT (iata,hour,category,hash_bytes,entries) DO UPDATE SET
receptions = analytics_path_stats_hourly.receptions + EXCLUDED.receptions
)
DELETE FROM packets WHERE packet_hash = ANY(hashes);
GET DIAGNOSTICS deleted = ROW_COUNT;
-- Once per completed cleanup, including when there were no raw packets to delete.
IF deleted < batch_size THEN
DELETE FROM analytics_hourly_iata_stats WHERE hour < date_trunc('hour', NOW(), 'UTC') - INTERVAL '30 days';
DELETE FROM analytics_payload_breakdown_by_iata WHERE bucket < date_trunc('hour', NOW(), 'UTC') - INTERVAL '30 days';
DELETE FROM analytics_top_observers_by_iata WHERE bucket < date_trunc('hour', NOW(), 'UTC') - INTERVAL '30 days';
DELETE FROM analytics_top_talkers_by_iata WHERE bucket < date_trunc('hour', NOW(), 'UTC') - INTERVAL '30 days';
DELETE FROM analytics_top_advertisers_by_iata WHERE bucket < date_trunc('hour', NOW(), 'UTC') - INTERVAL '30 days';
DELETE FROM analytics_observer_activity_hourly WHERE bucket < date_trunc('hour', NOW(), 'UTC') - INTERVAL '30 days';
DELETE FROM analytics_signal_stats_hourly WHERE hour < date_trunc('hour', NOW(), 'UTC') - INTERVAL '30 days';
DELETE FROM analytics_path_stats_hourly WHERE hour < date_trunc('hour', NOW(), 'UTC') - INTERVAL '30 days';
END IF;
RETURN deleted;
END $$;
db/migrations/041_route_evidence_index.sql
-9
View File
@@ -1,9 +0,0 @@
-- Copyright 2026 Beacon Contributors
-- SPDX-License-Identifier: AGPL-3.0-or-later
-- One statement outside a transaction. RunMigrations recovers interrupted builds;
-- avoid IF NOT EXISTS, which would silently accept an invalid index.
-- A compact digest narrows candidates; queries still compare the complete bytes.
CREATE INDEX CONCURRENTLY idx_observations_route_evidence
ON packet_observations (iata, hash_size, (decode(md5(path_bytes), 'hex')), heard_at DESC, id DESC)
WHERE path_bytes IS NOT NULL AND payload_type IS NOT NULL AND payload_type <> 9 AND hop_count >= 2;
db/migrations/042_meshmapper_scopes.sql
-17
View File
@@ -1,17 +0,0 @@
-- Copyright 2026 Beacon Contributors
-- SPDX-License-Identifier: AGPL-3.0-or-later
-- Preserve historical identities while distinguishing import-only matching keys.
ALTER TABLE transport_scopes ADD COLUMN IF NOT EXISTS imported_only BOOLEAN NOT NULL DEFAULT FALSE;
CREATE TABLE IF NOT EXISTS meshmapper_scope_catalogues (
iata TEXT NOT NULL,
url TEXT NOT NULL,
payload JSONB,
etag TEXT,
checked_at TIMESTAMPTZ,
attempted_at TIMESTAMPTZ NOT NULL,
next_attempt TIMESTAMPTZ NOT NULL,
last_error TEXT NOT NULL DEFAULT '',
PRIMARY KEY (iata, url)
);
db/migrations/043_clear_zero_locations.sql
-6
View File
@@ -1,6 +0,0 @@
-- Copyright 2026 Beacon Contributors
-- SPDX-License-Identifier: AGPL-3.0-or-later
-- An explicit 0/0 advert now clears the stored position; drop old 0/0 rows and stale sources.
UPDATE nodes SET latitude = NULL, longitude = NULL, location_source = NULL
WHERE location_source = 'advert' AND (latitude IS NULL OR (latitude = 0 AND longitude = 0));
db/migrations/044_meshmapper_zones.sql
-15
View File
@@ -1,15 +0,0 @@
-- Copyright 2026 Beacon Contributors
-- SPDX-License-Identifier: AGPL-3.0-or-later
-- Imported MeshMapper region boundaries. Kept apart from iata_codes.border so
-- startup seeding of manual borderFile borders never undoes the override.
CREATE TABLE IF NOT EXISTS meshmapper_zone_boundaries (
iata TEXT PRIMARY KEY,
url TEXT NOT NULL,
feature JSONB,
etag TEXT,
checked_at TIMESTAMPTZ,
attempted_at TIMESTAMPTZ NOT NULL,
next_attempt TIMESTAMPTZ NOT NULL,
last_error TEXT NOT NULL DEFAULT ''
);
db/migrations/045_clean_partial_observer_telemetry.sql
-7
View File
@@ -1,7 +0,0 @@
-- Copyright 2026 Beacon Contributors
-- SPDX-License-Identifier: AGPL-3.0-or-later
-- Some /status messages carry uptime but no radio stats. Those rows zero the airtime and
-- noise floor, dragging bucket averages down; ingest now skips them (statusStats.usable).
DELETE FROM observer_telemetry
WHERE noise_floor_db = 0 AND airtime_tx_secs = 0 AND airtime_rx_secs = 0;
db/observer_metrics_integration_test.go
+4 -18
View File
@@ -7,35 +7,21 @@ import (
sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc"
"github.com/google/uuid"
"github.com/jackc/pgx/v5/pgtype"
"strings"
"testing"
"time"
)
func TestObserverMetricsPostgres(t *testing.T) {
ctx, tx := retentionTx(t)
analyticsTables(t, ctx, tx)
// Use the real observer DDL without depending on a preinstalled public schema.
initial, err := migrationFiles.ReadFile("migrations/001_initial_schema.sql")
if err != nil {
t.Fatal(err)
}
ddl := string(initial)
start := strings.Index(ddl, "CREATE TABLE observers (")
end := strings.Index(ddl[start:], "\n);") + start + 3
if _, err := tx.Exec(ctx, "DROP TABLE observers; "+ddl[start:end]+" ALTER TABLE observers ADD COLUMN region_scope TEXT;"); err != nil {
t.Fatal(err)
}
applyBaseline(t, ctx, tx)
if _, err := tx.Exec(ctx, `
CREATE TABLE observer_brokers (observer_id uuid,broker_name text,last_seen timestamptz,last_packet_at timestamptz,PRIMARY KEY(observer_id,broker_name));
INSERT INTO observers (id,public_key) VALUES ('00000000-0000-0000-0000-000000000001','\x01');
INSERT INTO packets(packet_hash,payload_type,last_heard_at,first_heard_at) SELECT int4send(i),4,NOW()-interval '5 days',NOW()-interval '5 days' FROM generate_series(1,3) i;
INSERT INTO packets(packet_hash,payload_type,payload_version,route_type,raw_payload,raw_header,last_heard_at,first_heard_at) SELECT int4send(i),4,0,1,'\x00','\x00',NOW()-interval '5 days',NOW()-interval '5 days' FROM generate_series(1,3) i;
INSERT INTO packet_observations(packet_hash,observer_id,iata,heard_at,path_length_byte,hash_size,hop_count,payload_type,snr,rssi)
SELECT packet_hash,'00000000-0000-0000-0000-000000000001','YOW',date_trunc('hour',NOW())-interval '5 days',0,1,0,CASE WHEN packet_hash=int4send(1) THEN 4 END,CASE WHEN packet_hash IN (int4send(1),int4send(2)) THEN 'NaN'::real ELSE 0 END,-100 FROM packets;
`); err != nil {
t.Fatal(err)
}
applyStatsMigration(t, ctx, tx, "039_analytics_retention.sql")
store := &Store{q: sqlc.New(tx)}
id := uuid.MustParse("00000000-0000-0000-0000-000000000001")
if err := store.DeleteOldPackets(ctx, time.Now().Add(-72*time.Hour)); err != nil {
@@ -59,9 +45,9 @@ func TestObserverMetricsPostgres(t *testing.T) {
if activity.WindowEnd != until.UnixMilli() || activity.WindowStart != until.Add(-7*24*time.Hour).UnixMilli() || activity.Source != "hourly" {
t.Fatalf("window: %+v", activity)
}
if _, err := tx.Exec(ctx, `INSERT INTO packets(packet_hash,last_heard_at) VALUES ('\xaa',NOW());
if _, err := tx.Exec(ctx, `INSERT INTO packets(packet_hash,payload_type,payload_version,route_type,raw_payload,raw_header,first_heard_at,last_heard_at) VALUES ('\xaa',4,0,1,'\x00','\x00',NOW(),NOW());
INSERT INTO packet_observations(packet_hash,observer_id,iata,heard_at,path_length_byte,hash_size,hop_count,snr,rssi) VALUES ('\xaa','00000000-0000-0000-0000-000000000001','YOW',date_trunc('hour',NOW())-interval '30 minutes',0,1,0,'NaN'::real,-100);
INSERT INTO packet_observations(packet_hash,observer_id,iata,heard_at) VALUES ('\xaa','00000000-0000-0000-0000-000000000001','YOW',NOW()) ON CONFLICT DO NOTHING;`); err != nil {
INSERT INTO packet_observations(packet_hash,observer_id,iata,heard_at,path_length_byte,hash_size,hop_count) VALUES ('\xaa','00000000-0000-0000-0000-000000000001','YOW',NOW(),0,1,0) ON CONFLICT DO NOTHING;`); err != nil {
t.Fatal(err)
}
raw, err := store.GetObserverActivity(ctx, id, time.Hour, 15*time.Minute, until)
db/paths_integration_test.go
+19 -7
View File
@@ -36,11 +36,11 @@ func TestPathsPostgres(t *testing.T) {
t.Fatal(err)
}
defer tx.Rollback(context.Background())
isolateStatsSchema(t, ctx, tx)
applyBaseline(t, ctx, tx)
_, err = tx.Exec(ctx, `
SET LOCAL TIME ZONE 'America/Vancouver';
CREATE TABLE packet_observations(heard_at timestamptz NOT NULL,iata char(3) NOT NULL,payload_type smallint,path_length_byte smallint NOT NULL,hash_size smallint NOT NULL,hop_count smallint NOT NULL,path_bytes bytea);
INSERT INTO packet_observations
CREATE TEMP TABLE fixture(heard_at timestamptz NOT NULL,iata char(3) NOT NULL,payload_type smallint,path_length_byte smallint NOT NULL,hash_size smallint NOT NULL,hop_count smallint NOT NULL,path_bytes bytea);
INSERT INTO fixture
SELECT CASE id WHEN 1 THEN '2026-01-01 00:30+00'::timestamptz WHEN 17 THEN '2026-01-01 04:00+00'::timestamptz WHEN 18 THEN '2026-01-01 04:30+00'::timestamptz WHEN 20 THEN '2026-01-01 03:00+00'::timestamptz
ELSE '2026-01-01 00:30+00'::timestamptz+(id/10)*interval '2 hour'+(id%10)*interval '1 minute' END,
iata,payload,raw,width,hops,CASE WHEN size IS NULL THEN NULL ELSE decode(repeat('ab',size),'hex') END
@@ -52,7 +52,13 @@ FROM (VALUES
(13,'YYZ',5,0,1,0,0),(14,'YVR',16,1,1,1,1),(15,'YVR',4,63,1,63,63),
(16,'YVR',4,149,3,21,63),(17,'YVR',4,1,1,1,1),(18,'YVR',4,1,1,1,1),
(19,'YYJ',9,3,1,3,3),(20,'YVR',4,64,2,0,0)
) v(id,iata,payload,raw,width,hops,size);`)
) v(id,iata,payload,raw,width,hops,size);
ALTER TABLE fixture ADD COLUMN n serial;
INSERT INTO observers (id,public_key) VALUES ('00000000-0000-0000-0000-000000000001','\x01');
INSERT INTO packets (packet_hash,payload_type,payload_version,route_type,raw_payload,raw_header,first_heard_at,last_heard_at)
SELECT int4send(n),4,0,1,'\x00','\x00',heard_at,heard_at FROM fixture;
INSERT INTO packet_observations (packet_hash,observer_id,heard_at,iata,payload_type,path_length_byte,hash_size,hop_count,path_bytes)
SELECT int4send(n),'00000000-0000-0000-0000-000000000001',heard_at,iata,payload_type,path_length_byte,hash_size,hop_count,path_bytes FROM fixture;`)
if err != nil {
t.Fatal(err)
}
@@ -60,8 +66,9 @@ FROM (VALUES
if _, err := tx.Exec(ctx, "UPDATE packet_observations SET heard_at=to_timestamp(extract(epoch FROM $1::timestamptz)+extract(epoch FROM heard_at)-extract(epoch FROM '2026-01-01 00:00+00'::timestamptz))", since); err != nil {
t.Fatal(err)
}
applyStatsMigration(t, ctx, tx, "036_mv_path_stats.sql")
applyStatsMigration(t, ctx, tx, "036_mv_path_stats.sql") // interrupted journal retry
if _, err := tx.Exec(ctx, "REFRESH MATERIALIZED VIEW mv_path_stats_hourly"); err != nil {
t.Fatal(err)
}
store := &Store{q: sqlc.New(tx)}
until := since.Add(4 * time.Hour)
for _, mode := range []string{"force_custom_plan", "force_generic_plan"} {
@@ -116,7 +123,12 @@ FROM (VALUES
if _, err = tx.Exec(ctx, "TRUNCATE packet_observations"); err != nil {
t.Fatal(err)
}
if _, err = tx.Exec(ctx, `INSERT INTO packet_observations SELECT $1::timestamptz+interval '1 hour','YVR',4,i,(i>>6)+1,i&63,decode(repeat('ab',((i>>6)+1)*(i&63)),'hex') FROM generate_series(0,255)i;`, since); err != nil {
if _, err = tx.Exec(ctx, `INSERT INTO packets (packet_hash,payload_type,payload_version,route_type,raw_payload,raw_header,first_heard_at,last_heard_at)
SELECT int4send(1000+i),4,0,1,'\x00','\x00',$1,$1 FROM generate_series(0,255)i`, since); err != nil {
t.Fatal(err)
}
if _, err = tx.Exec(ctx, `INSERT INTO packet_observations (packet_hash,observer_id,heard_at,iata,payload_type,path_length_byte,hash_size,hop_count,path_bytes)
SELECT int4send(1000+i),'00000000-0000-0000-0000-000000000001',$1::timestamptz+interval '1 hour','YVR',4,i,(i>>6)+1,i&63,decode(repeat('ab',((i>>6)+1)*(i&63)),'hex') FROM generate_series(0,255)i;`, since); err != nil {
t.Fatal(err)
}
stale, err := store.GetPathStats(ctx, since, until, nil)
db/retention_integration_test.go
+2 -2
View File
@@ -46,9 +46,9 @@ func countRows(t *testing.T, ctx context.Context, tx pgx.Tx, sql string) int {
// More than one batch of old packets; observations go with them.
func TestDeleteOldPacketsBatchesPostgres(t *testing.T) {
ctx, tx := retentionTx(t)
analyticsTables(t, ctx, tx)
applyStatsMigration(t, ctx, tx, "039_analytics_retention.sql")
applyBaseline(t, ctx, tx)
_, err := tx.Exec(ctx, `
INSERT INTO observers (id, public_key) VALUES ('00000000-0000-0000-0000-000000000001', '\x01');
INSERT INTO packets (packet_hash, payload_type, payload_version, route_type, raw_payload, raw_header, first_heard_at, last_heard_at)
SELECT int4send(i), 4, 0, 1, '\x00', '\x00',
CASE WHEN i <= 2500 THEN '2026-01-01'::timestamptz ELSE '2026-02-01'::timestamptz END,
db/route_evidence_integration_test.go
+26 -71
View File
@@ -7,8 +7,6 @@ import (
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"os"
"strings"
"testing"
"time"
@@ -35,30 +33,17 @@ func (c *evidenceCapture) Query(ctx context.Context, sql string, args ...any) (p
func TestRouteEvidencePostgres(t *testing.T) {
ctx, tx := retentionTx(t)
analyticsTables(t, ctx, tx)
ddl, err := migrationFiles.ReadFile("migrations/024_known_routes_pathkey.sql")
applyBaseline(t, ctx, tx)
_, err := tx.Exec(ctx, `
INSERT INTO iata_codes(iata) VALUES ('YOW'),('YVR') ON CONFLICT DO NOTHING;
INSERT INTO observers(id,public_key,display_name) VALUES ('00000000-0000-0000-0000-000000000001','\x01','One'),('00000000-0000-0000-0000-000000000002','\x02','Two');
INSERT INTO nodes(id,public_key,node_type,name) VALUES ('00000000-0000-0000-0000-000000000001','\xaa0102',2,'A'),('00000000-0000-0000-0000-000000000002','\xbb0304',2,'B');
INSERT INTO packets(packet_hash,payload_type,payload_version,route_type,raw_payload,raw_header,first_heard_at,last_heard_at)
SELECT int4send(i),4,0,1,'\x00','\x00',NOW(),NOW() FROM generate_series(1,10) i;`)
if err != nil {
t.Fatal(err)
}
text := string(ddl)
start := strings.Index(text, "CREATE TABLE known_routes_new (")
end := start + strings.Index(text[start:], "\n);") + 3
table := strings.ReplaceAll(text[start:end], "known_routes_new", "known_routes")
_, err = tx.Exec(ctx, `CREATE TABLE iata_codes (iata char(3) PRIMARY KEY); INSERT INTO iata_codes VALUES ('YOW'),('YVR');
ALTER TABLE nodes ADD COLUMN latitude double precision, ADD COLUMN longitude double precision;`+table+`
INSERT INTO observers(id,display_name) VALUES ('00000000-0000-0000-0000-000000000001','One'),('00000000-0000-0000-0000-000000000002','Two');
INSERT INTO nodes(id,public_key,name) VALUES ('00000000-0000-0000-0000-000000000001','\xaa0102','A'),('00000000-0000-0000-0000-000000000002','\xbb0304','B');
INSERT INTO packets(packet_hash) SELECT int4send(i) FROM generate_series(1,10) i;`)
if err != nil {
t.Fatal(err)
}
index, err := migrationFiles.ReadFile("migrations/041_route_evidence_index.sql")
if err != nil {
t.Fatal(err)
}
if _, err := tx.Exec(ctx, strings.Replace(string(index), "CREATE INDEX CONCURRENTLY", "CREATE INDEX", 1)); err != nil {
t.Fatal(err)
}
capture := &evidenceCapture{Tx: tx}
store := &Store{q: sqlc.New(capture)}
ids := []uuid.UUID{uuid.MustParse("00000000-0000-0000-0000-000000000001"), uuid.MustParse("00000000-0000-0000-0000-000000000002")}
@@ -69,17 +54,20 @@ func TestRouteEvidencePostgres(t *testing.T) {
since := time.Now().UTC().Truncate(time.Millisecond).Add(-time.Hour)
until := since.Add(time.Hour)
at := since.Add(time.Minute + 123456*time.Microsecond)
_, err = tx.Exec(ctx, `INSERT INTO packet_observations(id,packet_hash,observer_id,iata,heard_at,hash_size,hop_count,path_bytes,payload_type,snr)
_, err = tx.Exec(ctx, `INSERT INTO packet_observations(id,packet_hash,observer_id,iata,heard_at,hash_size,hop_count,path_bytes,payload_type,snr,path_length_byte)
SELECT i,int4send(i), '00000000-0000-0000-0000-000000000001',CASE WHEN i=5 THEN 'YVR' ELSE 'YOW' END,
CASE WHEN i=9 THEN $2::timestamptz WHEN i=1 THEN $1::timestamptz-interval '2 microseconds' ELSE $1::timestamptz END,
CASE WHEN i=4 THEN 2 ELSE 1 END,CASE WHEN i=4 THEN 1 WHEN i=10 THEN 3 ELSE 2 END,
CASE WHEN i=6 THEN '\xabbb'::bytea WHEN i=10 THEN '\xaabbcc'::bytea ELSE '\xaabb'::bytea END,
CASE WHEN i=7 THEN 9 WHEN i=8 THEN NULL ELSE 4 END,CASE WHEN i=1 THEN 'NaN'::real ELSE 0 END FROM generate_series(1,10) i;
CASE WHEN i=7 THEN 9 WHEN i=8 THEN NULL ELSE 4 END,CASE WHEN i=1 THEN 'NaN'::real ELSE 0 END,0 FROM generate_series(1,10) i;
`, at, until)
if err != nil {
t.Fatal(err)
}
if _, err = tx.Exec(ctx, `INSERT INTO packet_observations(id,packet_hash,observer_id,iata,heard_at,hash_size,hop_count,path_bytes,payload_type) VALUES (11,int4send(1),'00000000-0000-0000-0000-000000000002','YOW',$1,1,2,'\xaabb',4)`, at.Add(-time.Microsecond)); err != nil {
if _, err = tx.Exec(ctx, `INSERT INTO packet_observations(id,packet_hash,observer_id,iata,heard_at,hash_size,hop_count,path_bytes,payload_type,path_length_byte) VALUES (11,int4send(1),'00000000-0000-0000-0000-000000000002','YOW',$1,1,2,'\xaabb',4,0)`, at.Add(-time.Microsecond)); err != nil {
t.Fatal(err)
}
if _, err = tx.Exec(ctx, `UPDATE packet_observations SET path_length_byte=((hash_size-1)<<6)|hop_count`); err != nil {
t.Fatal(err)
}
query := api.RouteEvidenceQuery{Since: since, Until: until, Limit: 2}
@@ -190,12 +178,13 @@ func TestRouteEvidencePostgres(t *testing.T) {
// Many unrelated paths must not turn this into a raw-table or route-ID scan.
started := time.Now()
_, err = tx.Exec(ctx, `INSERT INTO packets(packet_hash) SELECT int4send(i) FROM generate_series(100,200099) i;`)
_, err = tx.Exec(ctx, `INSERT INTO packets(packet_hash,payload_type,payload_version,route_type,raw_payload,raw_header,first_heard_at,last_heard_at)
SELECT int4send(i),4,0,1,'\x00','\x00',NOW(),NOW() FROM generate_series(100,200099) i;`)
if err != nil {
t.Fatal(err)
}
_, err = tx.Exec(ctx, `INSERT INTO packet_observations(packet_hash,observer_id,iata,heard_at,hash_size,hop_count,path_bytes,payload_type)
SELECT packet_hash,'00000000-0000-0000-0000-000000000001','YOW',$1,1,2,
_, err = tx.Exec(ctx, `INSERT INTO packet_observations(packet_hash,observer_id,iata,heard_at,path_length_byte,hash_size,hop_count,path_bytes,payload_type)
SELECT packet_hash,'00000000-0000-0000-0000-000000000001','YOW',$1,2,1,2,
CASE WHEN get_byte(packet_hash,3)=0 THEN '\xaabb'::bytea ELSE packet_hash END,4 FROM packets;`, at)
if err != nil {
t.Fatal(err)
@@ -231,52 +220,18 @@ func TestRouteEvidencePostgres(t *testing.T) {
}
func TestRouteEvidenceIndexPostgres(t *testing.T) {
dsn := os.Getenv("BEACON_TEST_POSTGRES_DSN")
if dsn == "" {
t.Skip("set BEACON_TEST_POSTGRES_DSN")
}
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
defer cancel()
conn, err := pgx.Connect(ctx, dsn)
if err != nil {
t.Fatal(err)
}
defer conn.Close(context.Background())
schema := fmt.Sprintf("route_evidence_%d", time.Now().UnixNano())
ident := pgx.Identifier{schema}.Sanitize()
if _, err = conn.Exec(ctx, "CREATE SCHEMA "+ident+"; SET search_path TO "+ident); err != nil {
t.Fatal(err)
}
defer func() { _, _ = conn.Exec(context.Background(), "DROP SCHEMA "+ident+" CASCADE") }()
if _, err = conn.Exec(ctx, `CREATE TABLE packet_observations (id bigint,iata char(3),hash_size smallint,path_bytes bytea,heard_at timestamptz,payload_type smallint,hop_count smallint)`); err != nil {
t.Fatal(err)
}
migration, err := migrationFiles.ReadFile("migrations/041_route_evidence_index.sql")
if err != nil {
t.Fatal(err)
}
for i := 0; i < 2; i++ {
if err = applyMigration(ctx, conn, string(migration)); err != nil {
t.Fatalf("migration/retry: %v", err)
}
}
if _, err = conn.Exec(ctx, `UPDATE pg_index SET indisvalid=false WHERE indexrelid='idx_observations_route_evidence'::regclass`); err != nil {
t.Fatal(err)
}
if err = applyMigration(ctx, conn, string(migration)); err != nil {
t.Fatal(err)
}
var valid bool
if err = conn.QueryRow(ctx, `SELECT indisvalid FROM pg_index WHERE indexrelid='idx_observations_route_evidence'::regclass`).Scan(&valid); err != nil || !valid {
t.Fatalf("invalid index after recovery: %v", err)
}
if _, err = conn.Exec(ctx, `INSERT INTO packet_observations(id,iata,hash_size,path_bytes,heard_at,payload_type,hop_count)
SELECT n,'YOW',1,'\xaabb',NOW(),4,n FROM generate_series(0,2) n;
ctx, tx := retentionTx(t)
applyBaseline(t, ctx, tx)
if _, err := tx.Exec(ctx, `INSERT INTO observers(id,public_key) VALUES ('00000000-0000-0000-0000-000000000001','\x01');
INSERT INTO packets(packet_hash,payload_type,payload_version,route_type,raw_payload,raw_header,first_heard_at,last_heard_at)
SELECT int4send(n),4,0,1,'\x00','\x00',NOW(),NOW() FROM generate_series(0,2) n;
INSERT INTO packet_observations(packet_hash,observer_id,iata,path_length_byte,hash_size,path_bytes,heard_at,payload_type,hop_count)
SELECT int4send(n),'00000000-0000-0000-0000-000000000001','YOW',n,1,'\xaabb',NOW(),4,n FROM generate_series(0,2) n;
ANALYZE packet_observations`); err != nil {
t.Fatal(err)
}
var indexed float32
if err = conn.QueryRow(ctx, `SELECT reltuples FROM pg_class WHERE oid='idx_observations_route_evidence'::regclass`).Scan(&indexed); err != nil || indexed != 1 {
if err := tx.QueryRow(ctx, `SELECT reltuples FROM pg_class WHERE oid='idx_observations_route_evidence'::regclass`).Scan(&indexed); err != nil || indexed != 1 {
t.Fatalf("0/1-hop rows entered the route index: %v %v", indexed, err)
}
}
db/routes_integration_test.go
+3 -11
View File
@@ -20,7 +20,7 @@ import (
//go:embed queries/queries.sql
var routeTestQueries string
// All synthetic rows and indexes are temporary and rolled back. The sparse
// All synthetic rows are rolled back. The sparse
// region sits behind 100,000 newer routes to expose global timestamp scans.
func TestListKnownRoutesPostgres(t *testing.T) {
dsn := os.Getenv("BEACON_TEST_POSTGRES_DSN")
@@ -39,8 +39,9 @@ func TestListKnownRoutesPostgres(t *testing.T) {
t.Fatal(err)
}
defer tx.Rollback(context.Background())
applyBaseline(t, ctx, tx)
_, err = tx.Exec(ctx, `
CREATE TEMP TABLE known_routes (LIKE public.known_routes INCLUDING ALL) ON COMMIT DROP;
INSERT INTO iata_codes (iata) VALUES ('GPT');
INSERT INTO known_routes (id,path_key,node_ids,hash_prefix,iata,hop_count,first_seen,last_seen)
SELECT i, decode(md5(i::text),'hex'), ARRAY[md5(i::text)::uuid], ARRAY['\x01'::bytea],
CASE WHEN i > 100000 THEN 'GPT' ELSE 'YYZ' END, i % 4 + 2,
@@ -51,15 +52,6 @@ ANALYZE known_routes;`)
if err != nil {
t.Fatal(err)
}
migration, err := migrationFiles.ReadFile("migrations/027_known_routes_iata_last_seen.sql")
if err != nil {
t.Fatal(err)
}
// A temporary table inside this rollback-only fixture cannot be indexed
// concurrently. Use the migration's exact definition with that option off.
if _, err := tx.Exec(ctx, strings.Replace(string(migration), "CREATE INDEX CONCURRENTLY", "CREATE INDEX", 1)); err != nil {
t.Fatal(err)
}
queries := strings.ReplaceAll(routeTestQueries, "\r\n", "\n")
querySQL := strings.Split(strings.Split(queries, "-- name: ListKnownRoutes :many\n")[1], "\n-- name:")[0]
if _, err := tx.Exec(ctx, "PREPARE route_page AS "+querySQL); err != nil {
db/signal_integration_test.go
+13 -6
View File
@@ -37,11 +37,11 @@ func TestSignalPostgres(t *testing.T) {
t.Fatal(err)
}
defer tx.Rollback(context.Background())
isolateStatsSchema(t, ctx, tx)
applyBaseline(t, ctx, tx)
_, err = tx.Exec(ctx, `
SET LOCAL TIME ZONE 'America/Vancouver';
CREATE TABLE packet_observations (heard_at timestamptz NOT NULL, iata char(3) NOT NULL, snr real, rssi smallint);
INSERT INTO packet_observations
CREATE TEMP TABLE fixture (heard_at timestamptz NOT NULL, iata char(3) NOT NULL, snr real, rssi smallint);
INSERT INTO fixture
SELECT ('2026-01-01 '||at||'+00')::timestamptz,iata,snr::real,rssi::smallint FROM (VALUES
('00:30:00','YVR','0',-100), ('00:45:00','YVR','0',0), ('00:59:00','YVR',NULL,NULL),
('01:00:00','YVR','-30',-140), ('01:10:00','YVR','-30.25',-141), ('01:20:00','YVR','30',0),
@@ -50,7 +50,13 @@ SELECT ('2026-01-01 '||at||'+00')::timestamptz,iata,snr::real,rssi::smallint FRO
('02:20:00','YYJ','10',-50), ('03:00:00','YYZ','-5',-110),
('04:00:00','YVR','20',-40), ('04:30:00','YVR','20',-40),
('03:10:00','YVR','0',NULL), ('03:20:00','YVR',NULL,0)
) v(at,iata,snr,rssi);`)
) v(at,iata,snr,rssi);
ALTER TABLE fixture ADD COLUMN n serial;
INSERT INTO observers (id,public_key) VALUES ('00000000-0000-0000-0000-000000000001','\x01');
INSERT INTO packets (packet_hash,payload_type,payload_version,route_type,raw_payload,raw_header,first_heard_at,last_heard_at)
SELECT int4send(n),4,0,1,'\x00','\x00',heard_at,heard_at FROM fixture;
INSERT INTO packet_observations (packet_hash,observer_id,heard_at,iata,snr,rssi,path_length_byte,hash_size,hop_count)
SELECT int4send(n),'00000000-0000-0000-0000-000000000001',heard_at,iata,snr,rssi,0,1,0 FROM fixture;`)
if err != nil {
t.Fatal(err)
}
@@ -58,8 +64,9 @@ SELECT ('2026-01-01 '||at||'+00')::timestamptz,iata,snr::real,rssi::smallint FRO
if _, err := tx.Exec(ctx, "UPDATE packet_observations SET heard_at=to_timestamp(extract(epoch FROM $1::timestamptz)+extract(epoch FROM heard_at)-extract(epoch FROM '2026-01-01 00:00+00'::timestamptz))", since); err != nil {
t.Fatal(err)
}
applyStatsMigration(t, ctx, tx, "035_mv_signal_stats.sql")
applyStatsMigration(t, ctx, tx, "035_mv_signal_stats.sql") // interrupted journal retry
if _, err := tx.Exec(ctx, "REFRESH MATERIALIZED VIEW mv_signal_stats_hourly"); err != nil {
t.Fatal(err)
}
store := &Store{q: sqlc.New(tx)}
until := since.Add(4 * time.Hour)
for _, tc := range []struct {
db/stats_migration_integration_test.go
-33
View File
@@ -1,33 +0,0 @@
// Copyright 2026 Beacon Contributors
// SPDX-License-Identifier: AGPL-3.0-or-later
package db
import (
"context"
"testing"
"github.com/google/uuid"
"github.com/jackc/pgx/v5"
)
// Views cannot reference temporary tables. Keep fixtures and real migration DDL
// in a unique schema inside the caller's rolled-back transaction instead.
func isolateStatsSchema(t *testing.T, ctx context.Context, tx pgx.Tx) {
t.Helper()
schema := pgx.Identifier{"stats_test_" + uuid.NewString()}.Sanitize()
if _, err := tx.Exec(ctx, "CREATE SCHEMA "+schema+"; SET LOCAL search_path TO "+schema); err != nil {
t.Fatal(err)
}
}
func applyStatsMigration(t *testing.T, ctx context.Context, tx pgx.Tx, name string) {
t.Helper()
migration, err := migrationFiles.ReadFile("migrations/" + name)
if err != nil {
t.Fatal(err)
}
if _, err := tx.Exec(ctx, string(migration)); err != nil {
t.Fatal(err)
}
}
internal/config/border.go
+1 -1
View File
@@ -15,7 +15,7 @@ import (
// map: well-formed Feature, geometry Polygon or MultiPolygon, every coordinate within
// lon [-180,180] / lat [-90,90], every ring closed (first point == last point, 4+ points --
// see orb.Ring.Closed). On success, returns the Feature re-marshaled to JSON with its bbox
// computed and set, so reads never need to recompute it (db/migrations/011_add_iata_border.sql).
// computed and set, so reads never need to recompute it (iata_codes.border).
//
// Coordinate order: GeoJSON packs coordinates [lon, lat], not [lat, lon]. A border authored
// with the axes swapped is still well-formed JSON and its numbers can still individually fall