feat: background task scheduler

move refresh views and delete tasks to scheduler
add new config for refresh and cleanup intervals

new deployments can set these in minutes to see data faster

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"
@@ -100,6 +101,18 @@ func main() {
 		maxConnsPerIP = 5
 	}
 
+	// resolve background intervals with defaults
+	viewRefreshInterval := cfg.Background.ViewRefresh.Duration
+	if viewRefreshInterval == 0 {
+		viewRefreshInterval = 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 cleanup=%s",
+		telemetryResolution, telemetryRetention, packetRetention, maxConnsPerIP, viewRefreshInterval, cleanupInterval)
+
 	// ── Hub ──────────────────────────────────────────────────────────────────
 	h := hub.New()
 	go h.Run()
@@ -147,9 +160,6 @@ func main() {
 		}
 	}
 
-	// refresh materialized 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)
@@ -250,25 +260,11 @@ 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),
+	})
+	go scheduler.Start(ctx)
 
 	// ── HTTP server ──────────────────────────────────────────────────────────
 	r := router.New(h, reader, []*ingest.Worker{broker1, broker2}, maxConnsPerIP, cfg.CORS)
@@ -320,15 +316,3 @@ func getEnv(key string) string {
 	}
 	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)
-	}
-}

config.yaml.example

@@ -117,3 +117,9 @@ 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
+#  cleanup: 1h        # default: 1h

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
+				}
+			}
+		}()
+	}
+}

internal/background/tasks.go

@@ -0,0 +1,49 @@
+// Copyright 2026 Beacon Contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package background
+
+import (
+	"context"
+	"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
+		},
+	}
+}

internal/config/config.go

@@ -24,6 +24,18 @@ 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"`
+
+	// 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.