diff --git a/.github/workflows/codeql.yml b/.github/workflows/codeql.yml index dd81eaf..0e4d68f 100644 --- a/.github/workflows/codeql.yml +++ b/.github/workflows/codeql.yml @@ -1,101 +1,44 @@ -# For most projects, this workflow file will not need changing; you simply need -# to commit it to your repository. -# -# You may wish to alter this file to override the set of languages analyzed, -# or to provide custom queries or build logic. -# -# ******** NOTE ******** -# We have attempted to detect the languages in your repository. Please check -# the `language` matrix defined below to confirm you have the correct set of -# supported CodeQL languages. -# -name: "CodeQL Advanced" +# SPDX-License-Identifier: AGPL-3.0-or-later +name: CodeQL on: push: - branches: [ "main", dev" ] + branches: [main, dev] pull_request: - branches: [ "dev" ] + branches: [main, dev] schedule: - - cron: '43 11 * * 1' + - cron: '0 3 * * 1' jobs: analyze: - name: Analyze (${{ matrix.language }}) - # Runner size impacts CodeQL analysis time. To learn more, please see: - # - https://gh.io/recommended-hardware-resources-for-running-codeql - # - https://gh.io/supported-runners-and-hardware-resources - # - https://gh.io/using-larger-runners (GitHub.com only) - # Consider using larger runners or machines with greater resources for possible analysis time improvements. - runs-on: ${{ (matrix.language == 'swift' && 'macos-latest') || 'ubuntu-latest' }} + name: Analyze (go) + runs-on: ubuntu-latest permissions: - # required for all workflows security-events: write - - # required to fetch internal or private CodeQL packs packages: read - - # only required for workflows in private repositories actions: read contents: read - strategy: - fail-fast: false - matrix: - include: - - language: actions - build-mode: none - - language: go - build-mode: autobuild - # CodeQL supports the following values keywords for 'language': 'actions', 'c-cpp', 'csharp', 'go', 'java-kotlin', 'javascript-typescript', 'python', 'ruby', 'rust', 'swift' - # Use `c-cpp` to analyze code written in C, C++ or both - # Use 'java-kotlin' to analyze code written in Java, Kotlin or both - # Use 'javascript-typescript' to analyze code written in JavaScript, TypeScript or both - # To learn more about changing the languages that are analyzed or customizing the build mode for your analysis, - # see https://docs.github.com/en/code-security/code-scanning/creating-an-advanced-setup-for-code-scanning/customizing-your-advanced-setup-for-code-scanning. - # If you are analyzing a compiled language, you can modify the 'build-mode' for that language to customize how - # your codebase is analyzed, see https://docs.github.com/en/code-security/code-scanning/creating-an-advanced-setup-for-code-scanning/codeql-code-scanning-for-compiled-languages steps: - - name: Checkout repository - uses: actions/checkout@v4 + - name: Checkout + uses: actions/checkout@v4 - # Add any setup steps before running the `github/codeql-action/init` action. - # This includes steps like installing compilers or runtimes (`actions/setup-node` - # or others). This is typically only required for manual builds. - # - name: Setup runtime (example) - # uses: actions/setup-example@v1 + - name: Set up Go + uses: actions/setup-go@v5 + with: + go-version-file: go.mod + cache: true - # Initializes the CodeQL tools for scanning. - - name: Initialize CodeQL - uses: github/codeql-action/init@v4 - with: - languages: ${{ matrix.language }} - build-mode: ${{ matrix.build-mode }} - # If you wish to specify custom queries, you can do so here or in a config file. - # By default, queries listed here will override any specified in a config file. - # Prefix the list here with "+" to use these queries and those in the config file. + - name: Initialize CodeQL + uses: github/codeql-action/init@v4 + with: + languages: go + queries: security-and-quality - # For more details on CodeQL's query packs, refer to: https://docs.github.com/en/code-security/code-scanning/automatically-scanning-your-code-for-vulnerabilities-and-errors/configuring-code-scanning#using-queries-in-ql-packs - # queries: security-extended,security-and-quality + - name: Build + run: go build ./... - # If the analyze step fails for one of the languages you are analyzing with - # "We were unable to automatically build your code", modify the matrix above - # to set the build mode to "manual" for that language. Then modify this step - # to build your code. - # ℹī¸ Command-line programs to run using the OS shell. - # 📚 See https://docs.github.com/en/actions/using-workflows/workflow-syntax-for-github-actions#jobsjob_idstepsrun - - name: Run manual build steps - if: matrix.build-mode == 'manual' - shell: bash - run: | - echo 'If you are using a "manual" build mode for one or more of the' \ - 'languages you are analyzing, replace this with the commands to build' \ - 'your code, for example:' - echo ' make bootstrap' - echo ' make release' - exit 1 - - - name: Perform CodeQL Analysis - uses: github/codeql-action/analyze@v4 - with: - category: "/language:${{matrix.language}}" + - name: Perform CodeQL Analysis + uses: github/codeql-action/analyze@v4 + with: + category: /language:go diff --git a/README.md b/README.md index 99cd203..b4489eb 100644 --- a/README.md +++ b/README.md @@ -5,7 +5,7 @@ more MeshCore MQTT brokers, ingests LoRa packet traffic in real time, stores it in PostgreSQL, and streams live events to WebSocket clients. [![CI](https://github.com/MeshCore-Beacon/beacon-server/actions/workflows/ci.yml/badge.svg)](https://github.com/MeshCore-Beacon/beacon-server/actions/workflows/ci.yml) -[![CodeQL](https://github.com/MeshCore-Beacon/beacon-server/actions/workflows/codeql.yml/badge.svg)](https://github.com/MeshCore-Beacon/beacon-server/actions/workflows/codeql.yml) +[![CodeQL](https://github.com/MeshCore-Beacon/beacon-server/actions/workflows/codeql.yml/badge.svg?branch=main)](https://github.com/MeshCore-Beacon/beacon-server/actions/workflows/codeql.yml) ![Coverage](https://img.shields.io/endpoint?url=https://gist.githubusercontent.com/446564/3e707bdf3f06ecb4575166ce598051c3/raw/beacon-coverage.json) [![Docker](https://github.com/MeshCore-Beacon/beacon-server/actions/workflows/docker-publish.yml/badge.svg)](https://github.com/MeshCore-Beacon/beacon-server/actions/workflows/docker-publish.yml) @@ -324,7 +324,7 @@ Not yet implemented — see the Authentication section above. ### Endpoints | Method | Path | Description | -| ------ | ----------------------------------- | -------------------------------------------------------------------------------------------------- | ----- | +| ------ | ----------------------------------- | -------------------------------------------------------------------------------------------------- | | `GET` | `/brokers` | List MQTT brokers and connection status | | `GET` | `/channels` | List channels (optional: `?hash=&iata=&limit=50`) | | `GET` | `/channels/{id}` | Get channel detail by integer ID | @@ -357,7 +357,7 @@ Not yet implemented — see the Authentication section above. | `GET` | `/stats/scopes` | Configured region scopes and breakdown of packets, nodes, observers | | `GET` | `/stats/top-nodes` | Top N nodes by observation count (from materialized view) | | `GET` | `/stats/top-observers` | Top N observers by observation count (last 24h by default) | -| `GET` | `/traces` | List trace tags with filters (optional: ?type=TRACE | PING) | +| `GET` | `/traces` | List trace tags with filters (optional: ?type=TRACE\|PING) | | `GET` | `/traces/{tag}` | Get full trace detail with resolved routes | --- diff --git a/cmd/beacon/main.go b/cmd/beacon/main.go index 3a77fca..abffeaa 100644 --- a/cmd/beacon/main.go +++ b/cmd/beacon/main.go @@ -20,6 +20,7 @@ import ( _ "github.com/MeshCore-Beacon/beacon-server/docs" "github.com/MeshCore-Beacon/beacon-server/internal/api" "github.com/MeshCore-Beacon/beacon-server/internal/api/router" + "github.com/MeshCore-Beacon/beacon-server/internal/background" "github.com/MeshCore-Beacon/beacon-server/internal/cache" "github.com/MeshCore-Beacon/beacon-server/internal/config" "github.com/MeshCore-Beacon/beacon-server/internal/hub" @@ -33,7 +34,7 @@ import ( ) // @title MeshCore Beacon API -// @version 1.3.0 +// @version 1.4.0 // @description MeshCore network observation backend. Ingests LoRa packets from MQTT brokers, stores in PostgreSQL, and streams live events via WebSocket. // @termsOfService https://github.com/MeshCore-Beacon/beacon-server @@ -100,6 +101,23 @@ func main() { maxConnsPerIP = 5 } + // resolve background intervals with defaults + viewRefreshInterval := cfg.Background.ViewRefresh.Duration + if viewRefreshInterval == 0 { + viewRefreshInterval = time.Hour + } + reconfirmInterval := cfg.Background.Reconfirm.Duration + if reconfirmInterval == 0 { + reconfirmInterval = time.Hour + } + cleanupInterval := cfg.Background.Cleanup.Duration + if cleanupInterval == 0 { + cleanupInterval = time.Hour + } + + log.Printf("config: loaded — telemetryResolution=%s telemetryRetention=%s packetRetention=%s maxConnsPerIP=%d viewRefresh=%s reconfirm=%s cleanup=%s", + telemetryResolution, telemetryRetention, packetRetention, maxConnsPerIP, viewRefreshInterval, reconfirmInterval, cleanupInterval) + // ── Hub ────────────────────────────────────────────────────────────────── h := hub.New() go h.Run() @@ -108,7 +126,7 @@ func main() { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - pool, err := pgxpool.New(ctx, mustEnv("POSTGRES_DSN")) + pool, err := pgxpool.New(ctx, getEnv("POSTGRES_DSN")) if err != nil { log.Fatalf("failed to connect to postgres at %s: %v", os.Getenv("POSTGRES_DSN_HOST"), err) } @@ -147,9 +165,6 @@ func main() { } } - // refresh meterialized views on boot or restart to stay fresh - refreshMaterializedViews(ctx, store) - // ── Seed config data ───────────────────────────────────────────────────── if err := config.Seed(ctx, cfg, store); err != nil { log.Fatalf("failed to seed config: %v", err) @@ -215,9 +230,9 @@ func main() { broker1 := ingest.New( ingest.Config{ BrokerName: "mqtt1", - URL: mustEnv("MQTT_BROKER_1_URL"), - Username: mustEnv("MQTT_BROKER_1_USERNAME"), - Password: mustEnv("MQTT_BROKER_1_PASSWORD"), + URL: getEnv("MQTT_BROKER_1_URL"), + Username: getEnv("MQTT_BROKER_1_USERNAME"), + Password: getEnv("MQTT_BROKER_1_PASSWORD"), TelemetryResolution: telemetryResolution, AllowedIATAs: allowedIATAs, }, @@ -230,9 +245,9 @@ func main() { broker2 := ingest.New( ingest.Config{ BrokerName: "mqtt2", - URL: mustEnv("MQTT_BROKER_2_URL"), - Username: mustEnv("MQTT_BROKER_2_USERNAME"), - Password: mustEnv("MQTT_BROKER_2_PASSWORD"), + URL: getEnv("MQTT_BROKER_2_URL"), + Username: getEnv("MQTT_BROKER_2_USERNAME"), + Password: getEnv("MQTT_BROKER_2_PASSWORD"), TelemetryResolution: telemetryResolution, AllowedIATAs: allowedIATAs, }, @@ -250,25 +265,12 @@ func main() { go broker1.Start(ctx) go broker2.Start(ctx) - // ── cleanup and materialized view refresh goroutine ───────────────────────────────────────── - go func() { - ticker := time.NewTicker(time.Hour) - defer ticker.Stop() - for { - select { - case <-ticker.C: - if err := store.DeleteOldTelemetry(ctx, time.Now().Add(-telemetryRetention)); err != nil { - log.Printf("cleanup: delete old telemetry failed: %v", err) - } - if err := store.DeleteOldPackets(ctx, time.Now().Add(-packetRetention)); err != nil { - log.Printf("cleanup: delete old packets failed: %v", err) - } - refreshMaterializedViews(ctx, store) - case <-ctx.Done(): - return - } - } - }() + scheduler := background.New([]background.Task{ + background.ViewRefreshTask(store, viewRefreshInterval), + background.CleanupTask(store, telemetryRetention, packetRetention, cleanupInterval), + background.ReconfirmTask(store, reconfirmInterval), + }) + go scheduler.Start(ctx) // ── HTTP server ────────────────────────────────────────────────────────── r := router.New(h, reader, []*ingest.Worker{broker1, broker2}, maxConnsPerIP, cfg.CORS) @@ -292,7 +294,9 @@ func main() { log.Println("shutting down...") cancel() // stops ingest workers - if err := srv.Shutdown(context.Background()); err != nil { + shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 30*time.Second) + defer shutdownCancel() + if err := srv.Shutdown(shutdownCtx); err != nil { log.Printf("server shutdown error: %v", err) } } @@ -308,25 +312,13 @@ func entryExists(entries []keystore.Entry, e keystore.Entry) bool { return false } -// mustEnv returns the value of an env var and logs a warning if it is unset. +// getEnv returns the value of an env var and logs a warning if it is unset. // Callers that require the value to be non-empty should fatal themselves; // ingest workers tolerate missing broker config and will fail on connect instead. -func mustEnv(key string) string { +func getEnv(key string) string { v := os.Getenv(key) if v == "" { log.Printf("warning: %s is not set", key) } return v } - -func refreshMaterializedViews(ctx context.Context, store *db.Store) { - if err := store.RefreshHourlyStats(ctx); err != nil { - log.Printf("refresh: materialized view for hourly stats failed: %v", err) - } - if err := store.RefreshTopNodes(ctx); err != nil { - log.Printf("refresh: materialized view for top nodes failed: %v", err) - } - if err := store.RefreshRadioPresets(ctx); err != nil { - log.Printf("refresh: materialized view for radio presets failed: %v", err) - } -} diff --git a/config.yaml.example b/config.yaml.example index 657a8d9..d99f4de 100644 --- a/config.yaml.example +++ b/config.yaml.example @@ -117,3 +117,10 @@ cache: # allow_countries: [CA, US] # allow_continents: [NA] +# Background task intervals. +# Shorter intervals are useful during initial deployment to confirm data is +# flowing. Back off to 1h or more once stable. +#background: +# view_refresh: 1h # default: 1h +# reconfirm: 1h # default: 1h +# cleanup: 1h # default: 1h diff --git a/db/migrations/004_clean_non_infra_neighbors.sql b/db/migrations/004_clean_non_infra_neighbors.sql new file mode 100644 index 0000000..6cebb97 --- /dev/null +++ b/db/migrations/004_clean_non_infra_neighbors.sql @@ -0,0 +1,11 @@ +-- 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); diff --git a/db/nodes.go b/db/nodes.go index fb4353c..38db9a6 100644 --- a/db/nodes.go +++ b/db/nodes.go @@ -245,3 +245,7 @@ func (s *Store) GetNodeNeighbors(ctx context.Context, nodeID uuid.UUID) ([]api.N } return items, nil } + +func (s *Store) ReconfirmNeighbors(ctx context.Context) error { + return s.q.ReconfirmNeighbors(ctx) +} diff --git a/db/queries/queries.sql b/db/queries/queries.sql index 1a25367..935d22d 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -17,11 +17,12 @@ SELECT * FROM iata_codes WHERE iata = $1; SELECT * FROM iata_codes ORDER BY iata; -- name: UpsertIATADetails :exec -UPDATE iata_codes SET - display_name = $2, - approx_lat = $3, - approx_lng = $4 -WHERE iata = $1; +INSERT INTO iata_codes (iata, display_name, approx_lat, approx_lng) +VALUES ($1, $2, $3, $4) +ON CONFLICT (iata) DO UPDATE SET + display_name = EXCLUDED.display_name, + approx_lat = EXCLUDED.approx_lat, + approx_lng = EXCLUDED.approx_lng; -- ============================================================ -- TRANSPORT CODES @@ -919,6 +920,7 @@ SELECT ns.prefix_4 AS hash, n.id AS node_id, n.name, n.latitude, n.longitude, n. FROM node_short_ids ns JOIN nodes n ON n.id = ns.node_id WHERE ns.iata = $1 + AND n.node_type IN (2, 3) AND CASE WHEN cardinality($2::bytea[]) > 0 AND length($2[1]) = 1 THEN ns.prefix_1 = ANY($2) WHEN cardinality($2::bytea[]) > 0 AND length($2[1]) = 2 THEN ns.prefix_2 = ANY($2) @@ -935,3 +937,45 @@ REFRESH MATERIALIZED VIEW CONCURRENTLY mv_top_nodes_by_iata; -- name: RefreshRadioPresets :exec REFRESH MATERIALIZED VIEW CONCURRENTLY mv_radio_presets; + +-- name: ReconfirmRoutes :exec +-- Delete known_routes where any hop node has departed from node_short_ids for +-- that IATA, or where any hop's prefix_4 is now ambiguous (matches >1 node). +DELETE FROM known_routes kr +WHERE EXISTS ( + SELECT 1 + FROM unnest(kr.node_ids) AS hop_node_id + WHERE NOT EXISTS ( + SELECT 1 FROM node_short_ids ns + WHERE ns.node_id = hop_node_id + AND ns.iata = kr.iata + ) +) +OR EXISTS ( + SELECT 1 + FROM unnest(kr.hash_prefix) AS hop_prefix + WHERE ( + SELECT COUNT(*) FROM node_short_ids ns + WHERE ns.iata = kr.iata + AND ns.prefix_4 = hop_prefix + ) > 1 +); + +-- name: ReconfirmNeighbors :exec +-- Delete node_neighbors where the neighbor has departed from node_short_ids +-- for that IATA, or where its prefix_4 is now ambiguous. +DELETE FROM node_neighbors nn +WHERE NOT EXISTS ( + SELECT 1 FROM node_short_ids ns + WHERE ns.node_id = nn.neighbor_id + AND ns.iata = nn.iata +) +OR ( + SELECT COUNT(*) FROM node_short_ids ns + WHERE ns.iata = nn.iata + AND ns.prefix_4 = ( + SELECT prefix_4 FROM node_short_ids + WHERE node_id = nn.neighbor_id + AND iata = nn.iata + ) +) > 1; diff --git a/db/routes.go b/db/routes.go index d1aef44..3fd1783 100644 --- a/db/routes.go +++ b/db/routes.go @@ -259,6 +259,10 @@ func (s *Store) SearchCrossIATARoutes(ctx context.Context, fromHash, fromIATA, t return results, nil } +func (s *Store) ReconfirmRoutes(ctx context.Context) error { + return s.q.ReconfirmRoutes(ctx) +} + // extractFromNode returns the portion of a route starting at the given node. func extractFromNode(hops []api.RouteHop, nodeID uuid.UUID) []api.RouteHop { for i, hop := range hops { diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index 7b61752..c385605 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -2704,6 +2704,60 @@ func (q *Queries) ListTraceTags(ctx context.Context, arg ListTraceTagsParams) ([ return items, nil } +const reconfirmNeighbors = `-- name: ReconfirmNeighbors :exec +DELETE FROM node_neighbors nn +WHERE NOT EXISTS ( + SELECT 1 FROM node_short_ids ns + WHERE ns.node_id = nn.neighbor_id + AND ns.iata = nn.iata +) +OR ( + SELECT COUNT(*) FROM node_short_ids ns + WHERE ns.iata = nn.iata + AND ns.prefix_4 = ( + SELECT prefix_4 FROM node_short_ids + WHERE node_id = nn.neighbor_id + AND iata = nn.iata + ) +) > 1 +` + +// Delete node_neighbors where the neighbor has departed from node_short_ids +// for that IATA, or where its prefix_4 is now ambiguous. +func (q *Queries) ReconfirmNeighbors(ctx context.Context) error { + _, err := q.db.Exec(ctx, reconfirmNeighbors) + return err +} + +const reconfirmRoutes = `-- name: ReconfirmRoutes :exec +DELETE FROM known_routes kr +WHERE EXISTS ( + SELECT 1 + FROM unnest(kr.node_ids) AS hop_node_id + WHERE NOT EXISTS ( + SELECT 1 FROM node_short_ids ns + WHERE ns.node_id = hop_node_id + AND ns.iata = kr.iata + ) +) +OR EXISTS ( + SELECT 1 + FROM unnest(kr.hash_prefix) AS hop_prefix + WHERE ( + SELECT COUNT(*) FROM node_short_ids ns + WHERE ns.iata = kr.iata + AND ns.prefix_4 = hop_prefix + ) > 1 +) +` + +// Delete known_routes where any hop node has departed from node_short_ids for +// that IATA, or where any hop's prefix_4 is now ambiguous (matches >1 node). +func (q *Queries) ReconfirmRoutes(ctx context.Context) error { + _, err := q.db.Exec(ctx, reconfirmRoutes) + return err +} + const refreshHourlyStats = `-- name: RefreshHourlyStats :exec REFRESH MATERIALIZED VIEW CONCURRENTLY mv_hourly_iata_stats ` @@ -2737,6 +2791,7 @@ SELECT ns.prefix_4 AS hash, n.id AS node_id, n.name, n.latitude, n.longitude, n. FROM node_short_ids ns JOIN nodes n ON n.id = ns.node_id WHERE ns.iata = $1 + AND n.node_type IN (2, 3) AND CASE WHEN cardinality($2::bytea[]) > 0 AND length($2[1]) = 1 THEN ns.prefix_1 = ANY($2) WHEN cardinality($2::bytea[]) > 0 AND length($2[1]) = 2 THEN ns.prefix_2 = ANY($2) @@ -3025,11 +3080,12 @@ func (q *Queries) UpsertIATA(ctx context.Context, iata string) error { } const upsertIATADetails = `-- name: UpsertIATADetails :exec -UPDATE iata_codes SET - display_name = $2, - approx_lat = $3, - approx_lng = $4 -WHERE iata = $1 +INSERT INTO iata_codes (iata, display_name, approx_lat, approx_lng) +VALUES ($1, $2, $3, $4) +ON CONFLICT (iata) DO UPDATE SET + display_name = EXCLUDED.display_name, + approx_lat = EXCLUDED.approx_lat, + approx_lng = EXCLUDED.approx_lng ` type UpsertIATADetailsParams struct { diff --git a/docs/docs.go b/docs/docs.go index 0598939..3ed0e40 100644 --- a/docs/docs.go +++ b/docs/docs.go @@ -1274,7 +1274,7 @@ const docTemplate = `{ }, { "type": "integer", - "description": "Route ID of last item for pagination", + "description": "Epoch ms timestamp of last item for pagination", "name": "cursor", "in": "query" }, @@ -3530,7 +3530,7 @@ const docTemplate = `{ // SwaggerInfo holds exported Swagger Info so clients can modify it var SwaggerInfo = &swag.Spec{ - Version: "1.3.0", + Version: "1.4.0", Host: "localhost:8080", BasePath: "/api/v1", Schemes: []string{"http", "https"}, diff --git a/docs/swagger.json b/docs/swagger.json index 3d2a87b..b7c95c6 100644 --- a/docs/swagger.json +++ b/docs/swagger.json @@ -15,7 +15,7 @@ "license": { "name": "AGPL-3-or-later" }, - "version": "1.3.0" + "version": "1.4.0" }, "host": "localhost:8080", "basePath": "/api/v1", @@ -1272,7 +1272,7 @@ }, { "type": "integer", - "description": "Route ID of last item for pagination", + "description": "Epoch ms timestamp of last item for pagination", "name": "cursor", "in": "query" }, diff --git a/docs/swagger.yaml b/docs/swagger.yaml index cb7bcb2..6ff7067 100644 --- a/docs/swagger.yaml +++ b/docs/swagger.yaml @@ -1010,7 +1010,7 @@ info: name: AGPL-3-or-later termsOfService: https://github.com/MeshCore-Beacon/beacon-server title: MeshCore Beacon API - version: 1.3.0 + version: 1.4.0 paths: /brokers: get: @@ -1831,7 +1831,7 @@ paths: in: query name: hopCount type: integer - - description: Route ID of last item for pagination + - description: Epoch ms timestamp of last item for pagination in: query name: cursor type: integer diff --git a/internal/api/handlers/routes.go b/internal/api/handlers/routes.go index 9a1f128..9e819ac 100644 --- a/internal/api/handlers/routes.go +++ b/internal/api/handlers/routes.go @@ -32,7 +32,7 @@ func RoutesRouter(reader api.Reader) http.Handler { // @Produce json // @Param iata query string false "Filter by IATA code" // @Param hopCount query int false "Filter by exact hop count" -// @Param cursor query int false "Route ID of last item for pagination" +// @Param cursor query int false "Epoch ms timestamp of last item for pagination" // @Param limit query int false "Max results (default 50)" // @Success 200 {object} []api.KnownRoute // @Failure 500 {object} handlers.APIError diff --git a/internal/api/nodes.go b/internal/api/nodes.go index 2302bf9..67eb76f 100644 --- a/internal/api/nodes.go +++ b/internal/api/nodes.go @@ -64,6 +64,7 @@ type Node struct { } // NodeTypeName returns a human-readable name for a node type integer. +// NOTE: truncation is fine here until there are at least over 200 types of node func NodeTypeName(t int16) string { switch byte(t) { case meshcore.AdvertTypeChat: diff --git a/internal/background/background.go b/internal/background/background.go new file mode 100644 index 0000000..5e8c189 --- /dev/null +++ b/internal/background/background.go @@ -0,0 +1,50 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +// Package background runs periodic maintenance tasks on independent schedules. +package background + +import ( + "context" + "log" + "time" +) + +// Task is a named unit of work that runs on a fixed interval. +type Task struct { + Name string + Interval time.Duration + Run func(ctx context.Context) error +} + +// Scheduler runs a set of tasks on independent tickers. +type Scheduler struct { + tasks []Task +} + +// New creates a Scheduler with the given tasks. +func New(tasks []Task) *Scheduler { + return &Scheduler{tasks: tasks} +} + +// Start launches each task in its own goroutine. Blocks until ctx is cancelled. +func (s *Scheduler) Start(ctx context.Context) { + for _, t := range s.tasks { + go func() { + ticker := time.NewTicker(t.Interval) + defer ticker.Stop() + for { + select { + case <-ticker.C: + log.Printf("background[%s]: running", t.Name) + if err := t.Run(ctx); err != nil { + log.Printf("background[%s]: %v", t.Name, err) + } + log.Printf("background[%s]: complete", t.Name) + case <-ctx.Done(): + return + } + } + }() + } +} diff --git a/internal/background/tasks.go b/internal/background/tasks.go new file mode 100644 index 0000000..6006cce --- /dev/null +++ b/internal/background/tasks.go @@ -0,0 +1,69 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package background + +import ( + "context" + "fmt" + "log" + "time" + + "github.com/MeshCore-Beacon/beacon-server/db" +) + +// ViewRefreshTask returns a Task that refreshes all materialized views. +func ViewRefreshTask(store *db.Store, interval time.Duration) Task { + return Task{ + Name: "view_refresh", + Interval: interval, + Run: func(ctx context.Context) error { + if err := store.RefreshHourlyStats(ctx); err != nil { + log.Printf("background[view_refresh]: hourly stats: %v", err) + } + if err := store.RefreshTopNodes(ctx); err != nil { + log.Printf("background[view_refresh]: top nodes: %v", err) + } + if err := store.RefreshRadioPresets(ctx); err != nil { + log.Printf("background[view_refresh]: radio presets: %v", err) + } + return nil + }, + } +} + +// CleanupTask returns a Task that prunes old telemetry and packet rows. +func CleanupTask(store *db.Store, telemetryRetention, packetRetention, interval time.Duration) Task { + return Task{ + Name: "cleanup", + Interval: interval, + Run: func(ctx context.Context) error { + if err := store.DeleteOldTelemetry(ctx, time.Now().Add(-telemetryRetention)); err != nil { + return err + } + if err := store.DeleteOldPackets(ctx, time.Now().Add(-packetRetention)); err != nil { + return err + } + return nil + }, + } +} + +// ReconfirmTask returns a Task that prunes stale and ambiguous resolved paths +// and neighbors. Runs after routes to ensure neighbors are cleaned against +// already-reconfirmed path data. +func ReconfirmTask(store *db.Store, interval time.Duration) Task { + return Task{ + Name: "reconfirm", + Interval: interval, + Run: func(ctx context.Context) error { + if err := store.ReconfirmRoutes(ctx); err != nil { + return fmt.Errorf("routes: %w", err) + } + if err := store.ReconfirmNeighbors(ctx); err != nil { + return fmt.Errorf("neighbors: %w", err) + } + return nil + }, + } +} diff --git a/internal/config/config.go b/internal/config/config.go index 52b4104..50002f1 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -24,6 +24,21 @@ type Config struct { Scopes []ScopeConfig `yaml:"scopes"` Cache CacheConfig `yaml:"cache"` CORS CORSConfig `yaml:"cors"` + Background BackgroundConfig `yaml:"background"` +} + +// BackgroundConfig controls the intervals for background maintenance tasks. +type BackgroundConfig struct { + // ViewRefresh is how often materialized views are refreshed. + // Defaults to 1h if not set. + ViewRefresh duration `yaml:"view_refresh"` + + // Reconfirm prunes stale and ambiguous resolved paths and neigbors. + Reconfirm duration `yaml:"reconfirm"` + + // Cleanup is how often old telemetry and packet rows are pruned. + // Defaults to 1h if not set. + Cleanup duration `yaml:"cleanup"` } // CORSConfig controls Cross-Origin Resource Sharing behaviour.