feat(logging): configure levels and structured runtime output (#144)

* feat(logging): configure levels and structured runtime output

* fix(logging): bound request-derived log fields

* fix(logging): retain malformed IATA source attributes safely

* test(logging): capture structured REST rate-limit records

README.md

@@ -453,3 +453,17 @@ Beacon.
 
 Beacon stands on the shoulders of giants. See [SHOULDERS.md](SHOULDERS.md) for
 the full list of open source projects that make this possible.
+
+## Application logging
+
+Beacon writes application logs to stderr. Configure the minimum level and output format in `config.yaml`:
+
+```yaml
+log:
+  level: info   # debug, info, warn, error
+  format: text  # text or json
+```
+
+`LOG_LEVEL` and `LOG_FORMAT` override file settings; empty settings use `info` and `text`. Invalid values prevent startup. Configuration-loading failures can use the bootstrap text logger before file settings are available; failures after initialization retain error severity at every supported level.
+
+Records include a component field. Ingest workers also include their broker name, and HTTP completion records include the validated client address, route, status and duration. Query strings and protocol hello payloads are excluded. Expected ingest skips and routine WebSocket lifecycle details are debug-level. Changing the application's format does not change Caddy/Apache access logs or their fail2ban configuration. Collect/rotate stderr through Docker or systemd.

cmd/beacon/main.go

@@ -7,7 +7,7 @@ import (
 	"context"
 	"encoding/hex"
 	"fmt"
-	"log"
+	"log/slog"
 	"net/http"
 	"os"
 	"os/signal"
@@ -26,6 +26,7 @@ import (
 	"github.com/MeshCore-Beacon/beacon-server/internal/iatadb"
 	"github.com/MeshCore-Beacon/beacon-server/internal/ingest"
 	"github.com/MeshCore-Beacon/beacon-server/internal/keystore"
+	"github.com/MeshCore-Beacon/beacon-server/internal/logging"
 	"github.com/MeshCore-Beacon/beacon-server/internal/presence"
 	"github.com/MeshCore-Beacon/beacon-server/internal/scopestore"
 
@@ -73,7 +74,6 @@ var version = "dev"
 // @tag.name			Stats
 // @tag.description	Network statistics and time series
 func main() {
-	log.Printf("beacon version %s", version)
 	_ = godotenv.Load()
 	addr := os.Getenv("LISTEN_ADDR")
 	if addr == "" {
@@ -87,15 +87,23 @@ func main() {
 
 	cfg, err := config.Load(configPath)
 	if err != nil {
-		log.Fatalf("failed to load config: %v", err)
+		slog.Error("failed to load config", "component", "startup", "error", err)
+		os.Exit(1)
 	}
+	logger, err := logging.New(os.Stderr, cfg.Log)
+	if err != nil {
+		slog.Error("invalid logging configuration", "component", "startup", "error", err)
+		os.Exit(1)
+	}
+	slog.SetDefault(logger)
+	slog.Info("beacon starting", "component", "startup", "version", version)
 	if len(cfg.Server.TrustedProxies) == 0 {
-		log.Print("warning: server.trusted_proxies is empty; client IP headers are ignored and proxied clients share a WebSocket connection limit")
+		slog.Warn("warning: server.trusted_proxies is empty; client IP headers are ignored and proxied clients share a WebSocket connection limit", "component", "startup")
 	}
 
 	resolved := config.Resolve(cfg)
 
-	log.Printf("config: loaded — %s", resolved)
+	slog.Info(fmt.Sprintf("config: loaded — %s", resolved), "component", "startup")
 
 	// ── Hub ──────────────────────────────────────────────────────────────────
 	h := hub.New()
@@ -107,12 +115,15 @@ func main() {
 
 	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)
+		// Parse errors can embed the complete DSN, including its password.
+		slog.Error("invalid PostgreSQL connection configuration; check POSTGRES_DSN", "component", "startup")
+		os.Exit(1)
 	}
 	defer pool.Close()
 
 	if err := db.RunMigrations(ctx, pool); err != nil {
-		log.Fatalf("migrations failed: %v", err)
+		slog.Error("migrations failed", "component", "startup", "error", err)
+		os.Exit(1)
 	}
 
 	store := db.New(pool, resolved.ClockDriftThreshold, resolved.NodeStaleThreshold)
@@ -139,29 +150,30 @@ func main() {
 			}(),
 		)
 		if err := redisClient.Ping(ctx); err != nil {
-			log.Printf("warning: redis unavailable at %s, caching disabled: %v", redisAddr, err)
+			slog.Warn(fmt.Sprintf("warning: redis unavailable at %s, caching disabled", redisAddr), "component", "startup", "error", err)
 		} else {
 			ttls := cache.ResolveTTLs(cfg.Cache)
 			reader = cache.NewCachedReader(store, redisClient, ttls)
 			defer redisClient.Close()
-			log.Printf("cache: Redis connected at %s (stats=%s reference=%s nodes=%s observers=%s)",
-				redisAddr, ttls.Stats, ttls.Reference, ttls.Nodes, ttls.Observers)
+			slog.Info(fmt.Sprintf("cache: Redis connected at %s (stats=%s reference=%s nodes=%s observers=%s)", redisAddr, ttls.Stats, ttls.Reference, ttls.Nodes, ttls.Observers), "component", "startup")
 		}
 	}
 
 	// ── Seed config data ─────────────────────────────────────────────────────
 	if err := config.Seed(ctx, cfg, store); err != nil {
-		log.Fatalf("failed to seed config: %v", err)
+		slog.Error("failed to seed config", "component", "startup", "error", err)
+		os.Exit(1)
 	}
 
 	// ── Build transport scope keystore ───────────────────────────────────────
 	scopes := scopestore.New()
 	scopeEntries, err := store.GetTransportScopes(ctx)
 	if err != nil {
-		log.Fatalf("failed to load transport scopes: %v", err)
+		slog.Error("failed to load transport scopes", "component", "startup", "error", err)
+		os.Exit(1)
 	}
 	scopes.Load(scopeEntries)
-	log.Printf("loaded %d transport scopes", len(scopeEntries))
+	slog.Info(fmt.Sprintf("loaded %d transport scopes", len(scopeEntries)), "component", "startup")
 
 	// ── Build channel keystore ──────────────────────────────────────────────
 	entries := make(map[string][]keystore.Entry)
@@ -178,7 +190,7 @@ func main() {
 		}
 		if !keystore.EntryExists(entries[hashHex], entry) {
 			entries[hashHex] = append(entries[hashHex], entry)
-			log.Printf("config: loaded hashtag channel #%s (hash=%s)", tag, hashHex)
+			slog.Info(fmt.Sprintf("config: loaded hashtag channel #%s (hash=%s)", tag, hashHex), "component", "startup")
 		}
 	}
 
@@ -186,7 +198,7 @@ func main() {
 	for hashHex, keyCfg := range cfg.ChannelKeys.Keys {
 		key, err := hex.DecodeString(keyCfg.Key)
 		if err != nil {
-			log.Printf("warning: invalid channel key for hash %s, skipping: %v", hashHex, err)
+			slog.Warn(fmt.Sprintf("warning: invalid channel key for hash %s, skipping", hashHex), "component", "startup", "error", err)
 			continue
 		}
 		entry := keystore.Entry{
@@ -196,7 +208,7 @@ func main() {
 		}
 		if !keystore.EntryExists(entries[hashHex], entry) {
 			entries[hashHex] = append(entries[hashHex], entry)
-			log.Printf("config: loaded explicit channel key for hash %s name=%q", hashHex, keyCfg.Name)
+			slog.Info(fmt.Sprintf("config: loaded explicit channel key for hash %s name=%q", hashHex, keyCfg.Name), "component", "startup")
 		}
 	}
 
@@ -208,18 +220,17 @@ func main() {
 	// built, so adding a channel key to the config surfaces its history on the next boot
 	// instead of leaving it stranded in the DB indefinitely.
 	if n, err := ingest.BackfillChannelMessages(ctx, store, keys); err != nil {
-		log.Printf("config: channel message backfill failed: %v", err)
+		slog.Error("config: channel message backfill failed", "component", "startup", "error", err)
 	} else if n > 0 {
-		log.Printf("config: backfilled %d previously-undecrypted channel message(s)", n)
+		slog.Info(fmt.Sprintf("config: backfilled %d previously-undecrypted channel message(s)", n), "component", "startup")
 	}
 
 	// ── Build geographic ingest filter ───────────────────────────────────────────────────────────
 	allowedIATAs := iatadb.BuildAllowedSet(cfg.Ingest.AllowCountries, cfg.Ingest.AllowContinents)
 	if allowedIATAs != nil {
-		log.Printf("config: ingest filter active — %d allowed IATAs (countries=%v continents=%v)",
-			len(allowedIATAs), cfg.Ingest.AllowCountries, cfg.Ingest.AllowContinents)
+		slog.Info(fmt.Sprintf("config: ingest filter active — %d allowed IATAs (countries=%v continents=%v)", len(allowedIATAs), cfg.Ingest.AllowCountries, cfg.Ingest.AllowContinents), "component", "startup")
 	} else {
-		log.Printf("config: ingest filter inactive — accepting all IATAs")
+		slog.Info("config: ingest filter inactive — accepting all IATAs", "component", "startup")
 	}
 
 	broker1 := ingest.New(
@@ -279,14 +290,16 @@ func main() {
 	r := router.New(h, reader, []*ingest.Worker{broker1, broker2}, resolved.MaxConnsPerIP, cfg.CORS, cfg.Server, cfg.Auth, resolved.RateLimit)
 
 	srv := &http.Server{
-		Addr:    addr,
-		Handler: r,
+		Addr:     addr,
+		Handler:  r,
+		ErrorLog: slog.NewLogLogger(slog.Default().With("component", "http").Handler(), slog.LevelError),
 	}
 
 	go func() {
-		fmt.Printf("Beacon listening on %s\n", addr)
+		slog.Info(fmt.Sprintf("Beacon listening on %s", addr), "component", "startup")
 		if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
-			log.Fatalf("server error: %v", err)
+			slog.Error("server error", "component", "startup", "error", err)
+			os.Exit(1)
 		}
 	}()
 
@@ -295,12 +308,12 @@ func main() {
 	signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
 	<-quit
 
-	log.Println("shutting down...")
+	slog.Info("shutting down...", "component", "startup")
 	cancel() // stops ingest workers
 	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)
+		slog.Error("server shutdown error", "component", "startup", "error", err)
 	}
 	coalescer.Flush(shutdownCtx)
 }
@@ -311,7 +324,7 @@ func main() {
 func getEnv(key string) string {
 	v := os.Getenv(key)
 	if v == "" {
-		log.Printf("warning: %s is not set", key)
+		slog.Warn(fmt.Sprintf("warning: %s is not set", key), "component", "startup")
 	}
 	return v
 }

cmd/beacon/main_test.go

@@ -0,0 +1,60 @@
+// Copyright 2026 Beacon Contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package main
+
+import (
+	"bytes"
+	"encoding/json"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
+	"testing"
+)
+
+func TestStartupLogging(t *testing.T) {
+	if os.Getenv("BEACON_TEST_LOG_STARTUP") == "1" {
+		main()
+		return
+	}
+	for _, level := range []string{"debug", "info", "warn", "error"} {
+		t.Run(level, func(t *testing.T) {
+			dir := t.TempDir()
+			path := filepath.Join(dir, "config.yaml")
+			if err := os.WriteFile(path, []byte("log: {level: debug, format: text}\n"), 0600); err != nil {
+				t.Fatal(err)
+			}
+			cmd := exec.Command(os.Args[0], "-test.run=^TestStartupLogging$")
+			cmd.Dir = dir
+			// Invalid port fails during parsing, before any database/network connection.
+			cmd.Env = append(os.Environ(), "BEACON_TEST_LOG_STARTUP=1", "CONFIG_PATH="+path, "LOG_LEVEL="+level, "LOG_FORMAT=json", "POSTGRES_DSN=postgres://test:do-not-log-this@localhost:invalid/beacon")
+			out, err := cmd.CombinedOutput()
+			if err == nil {
+				t.Fatal("startup unexpectedly succeeded")
+			}
+			if strings.Contains(string(out), "do-not-log-this") {
+				t.Fatal("DSN credential leaked")
+			}
+			foundError := false
+			for _, line := range bytes.Split(bytes.TrimSpace(out), []byte("\n")) {
+				var record map[string]any
+				if err := json.Unmarshal(line, &record); err != nil {
+					t.Fatalf("non-JSON startup output: %s", line)
+				}
+				if record["level"] == "ERROR" {
+					foundError = true
+				}
+				if level == "error" && record["level"] != "ERROR" {
+					t.Fatal("error threshold ignored")
+				}
+				if level == "warn" && record["level"] != "WARN" && record["level"] != "ERROR" {
+					t.Fatal("warn threshold ignored")
+				}
+			}
+			if !foundError {
+				t.Fatal("startup failure was silent")
+			}
+		})
+	}
+}

config.yaml.example

@@ -6,6 +6,11 @@
 auth:
   api_key: ""
 
+log:
+  # Environment LOG_LEVEL / LOG_FORMAT override these values.
+  level: info  # debug, info, warn, error
+  format: text # text or json
+
 server:
   # Only these direct proxy peers may set the client IP through X-Real-IP.
   # The proxy must overwrite that header, not pass through client input.

db/migrate.go

@@ -9,6 +9,7 @@ import (
 	"errors"
 	"fmt"
 	"io/fs"
+	"log/slog"
 	"regexp"
 	"sort"
 	"strings"
@@ -70,13 +71,13 @@ func applyMigration(ctx context.Context, db execQuerier, sql string) error {
 		return fmt.Errorf("%w (checking index %s: %v)", err, name, scanErr)
 	}
 	if valid {
-		fmt.Printf("index %s already built, recording migration\n", name)
+		slog.Info(fmt.Sprintf("index %s already built, recording migration", name), "component", "db")
 		return nil
 	}
 	if _, dropErr := db.Exec(ctx, "DROP INDEX CONCURRENTLY IF EXISTS "+ident); dropErr != nil {
 		return fmt.Errorf("dropping invalid index %s: %w", name, dropErr)
 	}
-	fmt.Printf("dropped invalid index %s, rebuilding\n", name)
+	slog.Warn(fmt.Sprintf("dropped invalid index %s, rebuilding", name), "component", "db")
 	_, err = db.Exec(ctx, sql)
 	return err
 }
@@ -119,7 +120,7 @@ func RunMigrations(ctx context.Context, pool *pgxpool.Pool) error {
 			); err != nil {
 				return fmt.Errorf("failed to bootstrap migrations: %w", err)
 			}
-			fmt.Println("bootstrapped existing schema as 001_initial_schema.sql")
+			slog.Info("bootstrapped existing schema as 001_initial_schema.sql", "component", "db")
 		}
 	}
 
@@ -166,7 +167,7 @@ func RunMigrations(ctx context.Context, pool *pgxpool.Pool) error {
 			return fmt.Errorf("failed to record migration %s: %w", entry.Name(), err)
 		}
 
-		fmt.Printf("applied migration: %s\n", entry.Name())
+		slog.Info(fmt.Sprintf("applied migration: %s", entry.Name()), "component", "db")
 	}
 
 	return nil

db/nodes.go

@@ -9,7 +9,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"log"
+	"log/slog"
 	"time"
 
 	sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc"
@@ -132,7 +132,7 @@ func (s *Store) ListNodes(ctx context.Context, nodeType int16, iatas []string, s
 		}
 		if len(v.Iatas) > 0 {
 			if err := json.Unmarshal(v.Iatas, &node.IATAs); err != nil {
-				log.Printf("store: failed to unmarshal node iatas: %v", err)
+				slog.Error("store: failed to unmarshal node iatas", "component", "db", "error", err)
 				node.IATAs = []api.NodeIATA{}
 			}
 		}
@@ -184,13 +184,13 @@ func (s *Store) GetNode(ctx context.Context, nodeID uuid.UUID) (*api.Node, error
 	}
 	neighbors, err := s.GetNodeNeighbors(ctx, nodeID)
 	if err != nil {
-		log.Printf("store: GetNodeNeighbors failed for %s: %v", nodeID, err)
+		slog.Error(fmt.Sprintf("store: GetNodeNeighbors failed for %s", nodeID), "component", "db", "error", err)
 		neighbors = []api.NodeNeighbor{}
 	}
 	node.Neighbors = neighbors
 	if len(row.Iatas) > 0 {
 		if err := json.Unmarshal(row.Iatas, &node.IATAs); err != nil {
-			log.Printf("store: failed to unmarshal node iatas: %v", err)
+			slog.Error("store: failed to unmarshal node iatas", "component", "db", "error", err)
 			node.IATAs = []api.NodeIATA{}
 		}
 	}

db/observers.go

@@ -7,7 +7,7 @@ import (
 	"context"
 	"encoding/hex"
 	"fmt"
-	"log"
+	"log/slog"
 	"time"
 
 	sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc"
@@ -122,7 +122,7 @@ func (s *Store) GetObserver(ctx context.Context, observerID uuid.UUID) (*api.Obs
 	}
 	scopes, err := s.GetObserverScopes(ctx, observerID)
 	if err != nil {
-		log.Printf("store: GetObserverScopes failed for %s: %v", observerID, err)
+		slog.Error(fmt.Sprintf("store: GetObserverScopes failed for %s", observerID), "component", "db", "error", err)
 		scopes = []string{}
 	}
 	observer.Scopes = scopes
@@ -357,7 +357,7 @@ func (s *Store) ListObserverAdverts(ctx context.Context, observerID uuid.UUID, c
 		Limit:      limit + 1, // fetch one extra to detect hasMore
 	})
 	if err != nil {
-		log.Printf("api: ListObserverAdverts failed: %v", err)
+		slog.Error("api: ListObserverAdverts failed", "component", "db", "error", err)
 		return api.Page[api.AdvertObservation]{}, err
 	}
 	hasMore := len(rows) > int(limit)

db/packets.go

@@ -10,7 +10,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"log"
+	"log/slog"
 	"time"
 
 	sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc"
@@ -477,7 +477,7 @@ func (s *Store) GetPacket(ctx context.Context, packetHash []byte) (*api.Packet,
 			if len(traceRawHashes) > 0 {
 				resolved, err := s.ResolvePathHashes(ctx, v.Iata, traceRawHashes)
 				if err != nil {
-					log.Printf("store: path resolution failed for observation %d: %v", v.ID, err)
+					slog.Error(fmt.Sprintf("store: path resolution failed for observation %d", v.ID), "component", "db", "error", err)
 				} else {
 					resolvedPath = api.BuildResolvedPath(traceRawHashes, resolved)
 				}
@@ -490,7 +490,7 @@ func (s *Store) GetPacket(ctx context.Context, packetHash []byte) (*api.Packet,
 			}
 			resolved, err := s.ResolvePathHashes(ctx, v.Iata, hashes)
 			if err != nil {
-				log.Printf("store: path resolution failed for observation %d: %v", v.ID, err)
+				slog.Error(fmt.Sprintf("store: path resolution failed for observation %d", v.ID), "component", "db", "error", err)
 			} else {
 				resolvedPath = api.BuildResolvedPath(hashes, resolved)
 			}
@@ -513,7 +513,7 @@ func (s *Store) GetPacket(ctx context.Context, packetHash []byte) (*api.Packet,
 				obs.ResolvedSource = &hop
 			} else if len(sourceHashByte) == 1 {
 				if r, err := s.ResolveEndpointHashes(ctx, v.Iata, [][]byte{sourceHashByte}); err != nil {
-					log.Printf("store: source resolution failed for observation %d: %v", v.ID, err)
+					slog.Error(fmt.Sprintf("store: source resolution failed for observation %d", v.ID), "component", "db", "error", err)
 				} else {
 					hop := api.BuildResolvedPath([][]byte{sourceHashByte}, r)[0]
 					obs.ResolvedSource = &hop
@@ -521,7 +521,7 @@ func (s *Store) GetPacket(ctx context.Context, packetHash []byte) (*api.Packet,
 			}
 			if len(destHashByte) == 1 {
 				if r, err := s.ResolveEndpointHashes(ctx, v.Iata, [][]byte{destHashByte}); err != nil {
-					log.Printf("store: destination resolution failed for observation %d: %v", v.ID, err)
+					slog.Error(fmt.Sprintf("store: destination resolution failed for observation %d", v.ID), "component", "db", "error", err)
 				} else {
 					hop := api.BuildResolvedPath([][]byte{destHashByte}, r)[0]
 					obs.ResolvedDestination = &hop

db/stats.go

@@ -6,7 +6,7 @@ package db
 import (
 	"context"
 	"encoding/json"
-	"log"
+	"log/slog"
 	"time"
 
 	sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc"
@@ -186,7 +186,7 @@ func (s *Store) GetStatsClockDrift(ctx context.Context, iatas []string, limit in
 		}
 		if len(v.Iatas) > 0 {
 			if err := json.Unmarshal(v.Iatas, &entry.IATAs); err != nil {
-				log.Printf("store: failed to unmarshal clock drift node iatas: %v", err)
+				slog.Error("store: failed to unmarshal clock drift node iatas", "component", "db", "error", err)
 				entry.IATAs = []api.NodeIATA{}
 			}
 		}

env.example

@@ -3,6 +3,10 @@ LISTEN_ADDR=:8080
 # Optional admin bearer key. If set, overrides auth.api_key (even when empty).
 # BEACON_API_KEY=
 
+# Optional application logging overrides (defaults: info / text).
+# LOG_LEVEL=info
+# LOG_FORMAT=json
+
 POSTGRES_DSN=postgres://beacon:beacon@localhost:5432/beacon?sslmode=disable
 
 # Redis (optional — leave REDIS_ADDR unset to disable caching)

internal/api/handlers/limits.go

@@ -5,7 +5,7 @@ package handlers
 
 import (
 	"errors"
-	"log"
+	"log/slog"
 	"net/http"
 	"strconv"
 )
@@ -26,7 +26,7 @@ func parseLimit(r *http.Request, defaultLimit int32) (int32, error) {
 		return 0, errors.New("limit must be positive")
 	}
 	if limit > int64(maxListLimit) {
-		log.Printf("api: clamped list limit from %d to %d for %s on %s", limit, maxListLimit, r.RemoteAddr, r.URL.Path)
+		slog.Info("api: clamped list limit", "component", "api", "requested_limit", limit, "limit", maxListLimit)
 		return maxListLimit, nil
 	}
 	return int32(limit), nil

internal/api/handlers/limits_logging_test.go

@@ -0,0 +1,37 @@
+// Copyright 2026 Beacon Contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package handlers
+
+import (
+	"bytes"
+	"encoding/json"
+	"log"
+	"log/slog"
+	"net/http/httptest"
+	"strings"
+	"testing"
+)
+
+func TestLimitLogOmitsRequestValues(t *testing.T) {
+	var out bytes.Buffer
+	previous, writer, flags := slog.Default(), log.Writer(), log.Flags()
+	slog.SetDefault(slog.New(slog.NewJSONHandler(&out, nil)))
+	t.Cleanup(func() { slog.SetDefault(previous); log.SetOutput(writer); log.SetFlags(flags) })
+	req := httptest.NewRequest("GET", "/private-path%0Aforged?limit=1000", nil)
+	req.RemoteAddr = "private-peer"
+	limit, err := parseLimit(req, 50)
+	if err != nil || limit != 200 {
+		t.Fatalf("limit=%d error=%v", limit, err)
+	}
+	if strings.Contains(out.String(), "private-") || strings.Contains(out.String(), "forged") {
+		t.Fatal("limit log retained request identity")
+	}
+	var record map[string]any
+	if err := json.Unmarshal(out.Bytes(), &record); err != nil {
+		t.Fatal(err)
+	}
+	if record["requested_limit"] != float64(1000) || record["limit"] != float64(200) || record["component"] != "api" {
+		t.Fatalf("clamp diagnostics missing: %v", record)
+	}
+}

internal/api/handlers/responses.go

@@ -6,7 +6,8 @@ package handlers
 
 import (
 	"encoding/json"
-	"log"
+	"fmt"
+	"log/slog"
 	"net/http"
 	"strings"
 )
@@ -25,7 +26,7 @@ func respond(w http.ResponseWriter, status int, data any) {
 	w.Header().Set("Content-Type", "application/json")
 	w.WriteHeader(status)
 	if err := json.NewEncoder(w).Encode(data); err != nil {
-		log.Printf("api: failed to encode response: %v", err)
+		slog.Error("api: failed to encode response", "component", "api", "error", err)
 	}
 }
 
@@ -33,7 +34,7 @@ func respond(w http.ResponseWriter, status int, data any) {
 // The error code is derived automatically from the HTTP status text.
 func respondError(w http.ResponseWriter, status int, message string) {
 	if status >= 500 {
-		log.Printf("api: error %d: %s", status, message)
+		slog.Error(fmt.Sprintf("api: error %d: %s", status, message), "component", "api")
 	}
 	code := strings.ToLower(strings.ReplaceAll(http.StatusText(status), " ", "_"))
 	respond(w, status, map[string]APIError{"error": {Code: code, Message: message}})

internal/api/handlers/stats.go

@@ -4,7 +4,7 @@
 package handlers
 
 import (
-	"log"
+	"log/slog"
 	"net/http"
 	"strconv"
 	"time"
@@ -66,7 +66,7 @@ func getStatsOverview(reader api.Reader) http.HandlerFunc {
 		}
 		overview, err := reader.GetStatsOverview(r.Context(), iatas)
 		if err != nil {
-			log.Printf("api: GetStatsOverview failed: %v", err)
+			slog.Error("api: GetStatsOverview failed", "component", "api", "error", err)
 			respondError(w, http.StatusInternalServerError, "internal server error")
 			return
 		}
@@ -108,7 +108,7 @@ func getStatsObservations(reader api.Reader) http.HandlerFunc {
 		}
 		points, err := reader.GetStatsObservations(r.Context(), iatas, since)
 		if err != nil {
-			log.Printf("api: GetStatsObservations failed: %v", err)
+			slog.Error("api: GetStatsObservations failed", "component", "api", "error", err)
 			respondError(w, http.StatusInternalServerError, "internal server error")
 			return
 		}
@@ -150,7 +150,7 @@ func getStatsPayloadBreakdown(reader api.Reader) http.HandlerFunc {
 		}
 		breakdown, err := reader.GetStatsPayloadBreakdown(r.Context(), iatas, since)
 		if err != nil {
-			log.Printf("api: GetStatsPayloadBreakdown failed: %v", err)
+			slog.Error("api: GetStatsPayloadBreakdown failed", "component", "api", "error", err)
 			respondError(w, http.StatusInternalServerError, "internal server error")
 			return
 		}
@@ -188,7 +188,7 @@ func getStatsTopNodes(reader api.Reader) http.HandlerFunc {
 		}
 		nodes, err := reader.GetStatsTopNodes(r.Context(), iatas, limit)
 		if err != nil {
-			log.Printf("api: GetStatsTopNodes failed: %v", err)
+			slog.Error("api: GetStatsTopNodes failed", "component", "api", "error", err)
 			respondError(w, http.StatusInternalServerError, "internal server error")
 			return
 		}
@@ -236,7 +236,7 @@ func getStatsTopObservers(reader api.Reader) http.HandlerFunc {
 		}
 		observers, err := reader.GetStatsTopObservers(r.Context(), iatas, since, limit)
 		if err != nil {
-			log.Printf("api: GetStatsTopObservers failed: %v", err)
+			slog.Error("api: GetStatsTopObservers failed", "component", "api", "error", err)
 			respondError(w, http.StatusInternalServerError, "internal server error")
 			return
 		}
@@ -284,7 +284,7 @@ func getStatsTopAdvertisers(reader api.Reader) http.HandlerFunc {
 		}
 		advertisers, err := reader.GetStatsTopAdvertisers(r.Context(), iatas, since, limit)
 		if err != nil {
-			log.Printf("api: GetStatsTopAdvertisers failed: %v", err)
+			slog.Error("api: GetStatsTopAdvertisers failed", "component", "api", "error", err)
 			respondError(w, http.StatusInternalServerError, "internal server error")
 			return
 		}
@@ -322,7 +322,7 @@ func getStatsClockDrift(reader api.Reader) http.HandlerFunc {
 		}
 		entries, err := reader.GetStatsClockDrift(r.Context(), iatas, limit)
 		if err != nil {
-			log.Printf("api: GetStatsClockDrift failed: %v", err)
+			slog.Error("api: GetStatsClockDrift failed", "component", "api", "error", err)
 			respondError(w, http.StatusInternalServerError, "internal server error")
 			return
 		}
@@ -370,7 +370,7 @@ func getStatsTopTalkers(reader api.Reader) http.HandlerFunc {
 		}
 		talkers, err := reader.GetStatsTopTalkers(r.Context(), iatas, since, limit)
 		if err != nil {
-			log.Printf("api: GetStatsTopTalkers failed: %v", err)
+			slog.Error("api: GetStatsTopTalkers failed", "component", "api", "error", err)
 			respondError(w, http.StatusInternalServerError, "internal server error")
 			return
 		}

internal/api/middleware/logging.go

@@ -0,0 +1,63 @@
+// Copyright 2026 Beacon Contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package middleware
+
+import (
+	"fmt"
+	"log/slog"
+	"net/http"
+	"strings"
+	"time"
+
+	"github.com/go-chi/chi/v5"
+	chimw "github.com/go-chi/chi/v5/middleware"
+)
+
+// RequestLogger uses chi's response wrapper to preserve streaming/hijacking support.
+// Place it after TrustedProxyIP and RequestID, and before Recoverer.
+func RequestLogger(next http.Handler) http.Handler {
+	return chimw.RequestLogger(requestLogFormatter{})(next)
+}
+
+type requestLogFormatter struct{}
+type requestLogEntry struct{ request *http.Request }
+
+func (requestLogFormatter) NewLogEntry(r *http.Request) chimw.LogEntry {
+	return &requestLogEntry{request: r}
+}
+
+func (e *requestLogEntry) logger() *slog.Logger {
+	path := "unmatched"
+	if ctx := chi.RouteContext(e.request.Context()); ctx != nil && ctx.RoutePattern() != "" {
+		path = ctx.RoutePattern()
+	}
+	return slog.Default().With("component", "http", "method", singleLineLogValue(e.request.Method), "path", path,
+		"client_ip", singleLineLogValue(e.request.RemoteAddr), "request_id", singleLineLogValue(chimw.GetReqID(e.request.Context())))
+}
+
+// Strip line breaks before request fields reach a handler or downstream log consumer.
+func singleLineLogValue(value string) string {
+	value = strings.ReplaceAll(value, "\r", "")
+	return strings.ReplaceAll(value, "\n", "")
+}
+
+func (e *requestLogEntry) Write(status, bytes int, _ http.Header, elapsed time.Duration, _ any) {
+	if status == 0 {
+		status = http.StatusOK
+	}
+	level := slog.LevelInfo
+	if status >= 500 {
+		level = slog.LevelError
+	} else if status >= 400 {
+		level = slog.LevelWarn
+	}
+	if !slog.Default().Enabled(e.request.Context(), level) {
+		return
+	}
+	e.logger().Log(e.request.Context(), level, "request complete", "status", status, "bytes", bytes, "duration_ms", float64(elapsed)/float64(time.Millisecond))
+}
+
+func (e *requestLogEntry) Panic(value any, stack []byte) {
+	e.logger().ErrorContext(e.request.Context(), "request panic", "panic", fmt.Sprint(value), "stack", string(stack))
+}

internal/api/middleware/logging_test.go

@@ -0,0 +1,143 @@
+// Copyright 2026 Beacon Contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package middleware
+
+import (
+	"bytes"
+	"encoding/json"
+	"log"
+	"log/slog"
+	"net/http"
+	"net/http/httptest"
+	"net/netip"
+	"strings"
+	"testing"
+
+	"github.com/go-chi/chi/v5"
+	chimw "github.com/go-chi/chi/v5/middleware"
+)
+
+func TestRequestLogging(t *testing.T) {
+	for _, tc := range []struct {
+		status    int
+		panic     bool
+		level     slog.Level
+		wantLevel string
+	}{
+		{200, false, slog.LevelInfo, "INFO"}, {429, false, slog.LevelWarn, "WARN"},
+		{500, false, slog.LevelError, "ERROR"}, {200, false, slog.LevelWarn, ""}, {500, true, slog.LevelError, "ERROR"},
+	} {
+		t.Run(tc.level.String()+"/"+http.StatusText(tc.status), func(t *testing.T) {
+			var out bytes.Buffer
+			previous, writer, flags := slog.Default(), log.Writer(), log.Flags()
+			slog.SetDefault(slog.New(slog.NewJSONHandler(&out, &slog.HandlerOptions{Level: tc.level})))
+			t.Cleanup(func() { slog.SetDefault(previous); log.SetOutput(writer); log.SetFlags(flags) })
+			r := chi.NewRouter()
+			r.Use(chimw.RequestID)
+			r.Use(TrustedProxyIP([]netip.Prefix{netip.MustParsePrefix("192.0.2.0/24")}))
+			r.Use(RequestLogger)
+			r.Use(chimw.Recoverer)
+			r.Get("/items/{id}", func(w http.ResponseWriter, _ *http.Request) {
+				if tc.panic {
+					panic("test panic")
+				}
+				w.WriteHeader(tc.status)
+			})
+			req := httptest.NewRequest(http.MethodGet, "/items/private-id?token=private-token", nil)
+			req.RemoteAddr = "192.0.2.1:4444"
+			req.Header.Set("X-Real-IP", "203.0.113.7")
+			req.Header.Set("True-Client-IP", "198.51.100.8")
+			w := httptest.NewRecorder()
+			r.ServeHTTP(w, req)
+			if w.Code != tc.status {
+				t.Fatalf("status=%d", w.Code)
+			}
+			if tc.wantLevel == "" {
+				if out.Len() != 0 {
+					t.Fatal("info request was not filtered")
+				}
+				return
+			}
+			if strings.Contains(out.String(), "private-id") || strings.Contains(out.String(), "private-token") || strings.Contains(out.String(), "198.51.100.8") {
+				t.Fatal("request values leaked or spoofed identity used")
+			}
+			var completed bool
+			for _, line := range bytes.Split(bytes.TrimSpace(out.Bytes()), []byte("\n")) {
+				var record map[string]any
+				if err := json.Unmarshal(line, &record); err != nil {
+					t.Fatal(err)
+				}
+				if record["level"] != tc.wantLevel || record["component"] != "http" || record["client_ip"] != "203.0.113.7" || record["path"] != "/items/{id}" {
+					t.Fatalf("bad record: %v", record)
+				}
+				if record["msg"] == "request complete" {
+					completed = true
+					if record["status"] != float64(tc.status) {
+						t.Fatal("wrong logged status")
+					}
+				}
+			}
+			if !completed {
+				t.Fatal("request completion missing")
+			}
+		})
+	}
+}
+
+func TestRequestLogInputBoundary(t *testing.T) {
+	for _, format := range []string{"text", "json"} {
+		for _, path := range []string{"/items/private-path", "/unmatched-private-path%0D%0Aforged"} {
+			t.Run(format+path, func(t *testing.T) {
+				var out bytes.Buffer
+				previous, writer, flags := slog.Default(), log.Writer(), log.Flags()
+				var handler slog.Handler = slog.NewTextHandler(&out, nil)
+				if format == "json" {
+					handler = slog.NewJSONHandler(&out, nil)
+				}
+				slog.SetDefault(slog.New(handler))
+				t.Cleanup(func() { slog.SetDefault(previous); log.SetOutput(writer); log.SetFlags(flags) })
+				r := chi.NewRouter()
+				r.Use(chimw.RequestID, RequestLogger)
+				r.Get("/items/{id}", func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(200) })
+				req := httptest.NewRequest(http.MethodGet, path+"?token=private-token", nil)
+				req.RemoteAddr = "192.0.2.1:1234"
+				req.Header.Set("X-Request-ID", "client\r\nforged")
+				w := httptest.NewRecorder()
+				r.ServeHTTP(w, req)
+				wantPath, wantStatus := "/items/{id}", 200
+				if strings.HasPrefix(path, "/unmatched") {
+					wantPath, wantStatus = "unmatched", 404
+				}
+				if w.Code != wantStatus || strings.Count(out.String(), "\n") != 1 || strings.ContainsAny(strings.TrimSuffix(out.String(), "\n"), "\r\n") {
+					t.Fatalf("status or log boundary changed: %d %q", w.Code, out.String())
+				}
+				if strings.Contains(out.String(), "private-path") || strings.Contains(out.String(), "private-token") {
+					t.Fatal("raw path or query was retained")
+				}
+				if format == "json" {
+					var record map[string]any
+					if err := json.Unmarshal(out.Bytes(), &record); err != nil {
+						t.Fatal(err)
+					}
+					if record["path"] != wantPath || record["request_id"] != "clientforged" || record["client_ip"] != "192.0.2.1:1234" {
+						t.Fatalf("unexpected request fields: %v", record)
+					}
+				} else if !strings.Contains(out.String(), "path="+wantPath) || !strings.Contains(out.String(), "request_id=clientforged") {
+					t.Fatalf("unexpected text fields: %s", out.String())
+				}
+			})
+		}
+	}
+}
+
+func TestSingleLineLogValue(t *testing.T) {
+	for _, tc := range []struct{ input, want string }{
+		{"", ""}, {"GET", "GET"}, {"[2001:db8::1]:80", "[2001:db8::1]:80"},
+		{"client\r\nentry\n", "cliententry"}, {"café", "café"},
+	} {
+		if got := singleLineLogValue(tc.input); got != tc.want {
+			t.Errorf("got %q, want %q", got, tc.want)
+		}
+	}
+}

internal/api/router/rate_limit_test.go

@@ -7,6 +7,7 @@ import (
 	"bytes"
 	"encoding/json"
 	"log"
+	"log/slog"
 	"net/http"
 	"net/http/httptest"
 	"net/netip"
@@ -17,8 +18,6 @@ import (
 	"testing/synctest"
 	"time"
 
-	"github.com/go-chi/chi/v5/middleware"
-
 	"github.com/MeshCore-Beacon/beacon-server/internal/config"
 )
 
@@ -53,9 +52,9 @@ func rateRequest(handler http.Handler, path, peer, headerIP string) *httptest.Re
 
 func TestAPIRateLimitContractAndLogging(t *testing.T) {
 	var logs bytes.Buffer
-	previous := middleware.DefaultLogger
-	middleware.DefaultLogger = middleware.RequestLogger(&middleware.DefaultLogFormatter{Logger: log.New(&logs, "", 0), NoColor: true})
-	t.Cleanup(func() { middleware.DefaultLogger = previous })
+	previous, writer, flags := slog.Default(), log.Writer(), log.Flags()
+	slog.SetDefault(slog.New(slog.NewJSONHandler(&logs, nil)))
+	t.Cleanup(func() { slog.SetDefault(previous); log.SetOutput(writer); log.SetFlags(flags) })
 	proxy := config.ServerConfig{TrustedProxies: []netip.Prefix{netip.MustParsePrefix("127.0.0.1/32")}}
 	handler := New(nil, nil, nil, 5, config.CORSConfig{}, proxy, config.AuthConfig{}, config.ResolvedRateLimitConfig{Enabled: true, RequestsPerMinute: 2, Burst: 10})
 	for _, path := range []string{"/api/v1/brokers", "/api/v1/packets?limit=0"} {
@@ -77,8 +76,21 @@ func TestAPIRateLimitContractAndLogging(t *testing.T) {
 	if response.Header().Get("Content-Type") != "application/json" || response.Header().Get("Retry-After") != "60" || !strings.EqualFold(response.Header().Get("Access-Control-Expose-Headers"), "Retry-After") || response.Header().Get("Access-Control-Allow-Origin") == "" {
 		t.Fatalf("missing JSON/backoff/CORS headers: %v", response.Header())
 	}
-	if strings.Count(logs.String(), " - 429 ") != 1 || !strings.Contains(logs.String(), "from 198.51.100.25 - 429 ") || !strings.Contains(logs.String(), "/api/v1/brokers") {
-		t.Fatalf("expected one rejection log with resolved client and path: %s", logs.String())
+	rejections := 0
+	for _, line := range bytes.Split(bytes.TrimSpace(logs.Bytes()), []byte("\n")) {
+		var record map[string]any
+		if err := json.Unmarshal(line, &record); err != nil {
+			t.Fatal(err)
+		}
+		if record["status"] == float64(http.StatusTooManyRequests) {
+			rejections++
+			if record["level"] != "WARN" || record["client_ip"] != "198.51.100.25" || record["path"] != "/api/v1/*" {
+				t.Fatalf("wrong rejection log: %v", record)
+			}
+		}
+	}
+	if rejections != 1 {
+		t.Fatalf("got %d rejection logs, want 1", rejections)
 	}
 }
 

internal/api/router/router.go

@@ -71,7 +71,7 @@ func New(h *hub.Hub, reader api.Reader, workers []*ingest.Worker, maxConnsPerIP
 	// ── Global middleware ────────────────────────────────────────────────────
 	r.Use(middleware.RequestID)
 	r.Use(mw.TrustedProxyIP(serverCfg.TrustedProxies))
-	r.Use(middleware.Logger)
+	r.Use(mw.RequestLogger)
 	r.Use(middleware.Recoverer)
 	r.Use(middleware.CleanPath)
 	r.Use(middleware.StripSlashes)

internal/background/background.go

@@ -6,7 +6,7 @@ package background
 
 import (
 	"context"
-	"log"
+	"log/slog"
 	"time"
 )
 
@@ -36,11 +36,11 @@ func (s *Scheduler) Start(ctx context.Context) {
 			for {
 				select {
 				case <-ticker.C:
-					log.Printf("background[%s]: running", t.Name)
+					slog.Debug("task running", "component", "background", "task", t.Name)
 					if err := t.Run(ctx); err != nil {
-						log.Printf("background[%s]: failed: %v", t.Name, err)
+						slog.Error("task failed", "component", "background", "task", t.Name, "error", err)
 					} else {
-						log.Printf("background[%s]: complete", t.Name)
+						slog.Debug("task complete", "component", "background", "task", t.Name)
 					}
 				case <-ctx.Done():
 					return

internal/background/tasks_test.go

@@ -4,11 +4,15 @@
 package background
 
 import (
+	"bytes"
 	"context"
+	"encoding/json"
 	"errors"
 	"log"
+	"log/slog"
 	"strings"
 	"testing"
+	"testing/synctest"
 	"time"
 
 	"github.com/google/uuid"
@@ -155,33 +159,35 @@ func TestViewRefreshTaskCancelled(t *testing.T) {
 	}
 }
 
-type taskLog chan string
-
-func (w taskLog) Write(p []byte) (int, error) {
-	line := string(p)
-	if strings.Contains(line, "complete") || strings.Contains(line, "failed:") {
-		w <- line
-	}
-	return len(p), nil
-}
-
 func TestSchedulerFailureIsNotComplete(t *testing.T) {
-	lines := make(taskLog, 1)
-	previous := log.Writer()
-	log.SetOutput(lines)
-	defer log.SetOutput(previous)
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-	New([]Task{{Name: "fails", Interval: time.Millisecond, Run: func(context.Context) error {
-		cancel()
-		return errors.New("refresh unavailable")
-	}}}).Start(ctx)
-	select {
-	case line := <-lines:
-		if strings.Contains(line, "complete") || !strings.Contains(line, "refresh unavailable") {
-			t.Fatalf("incorrect failure status: %s", line)
+	synctest.Test(t, func(t *testing.T) {
+		var out bytes.Buffer
+		previous, writer, flags := slog.Default(), log.Writer(), log.Flags()
+		slog.SetDefault(slog.New(slog.NewJSONHandler(&out, &slog.HandlerOptions{Level: slog.LevelDebug})))
+		defer func() { slog.SetDefault(previous); log.SetOutput(writer); log.SetFlags(flags) }()
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+		New([]Task{{Name: "fails", Interval: time.Millisecond, Run: func(context.Context) error { cancel(); return errors.New("refresh unavailable") }}}).Start(ctx)
+		time.Sleep(2 * time.Millisecond) // advance the virtual clock to the first task tick
+		synctest.Wait()
+		failed := false
+		for _, line := range bytes.Split(bytes.TrimSpace(out.Bytes()), []byte("\n")) {
+			var record map[string]any
+			if err := json.Unmarshal(line, &record); err != nil {
+				t.Fatal(err)
+			}
+			if record["msg"] == "task complete" {
+				t.Fatal("failed task reported complete")
+			}
+			if record["msg"] == "task failed" {
+				failed = true
+				if record["level"] != "ERROR" || record["error"] != "refresh unavailable" || record["task"] != "fails" {
+					t.Fatalf("incorrect failure: %v", record)
+				}
+			}
 		}
-	case <-time.After(time.Second):
-		t.Fatal("scheduler did not report task outcome")
-	}
+		if !failed {
+			t.Fatal("scheduler did not report task failure")
+		}
+	})
 }

internal/config/config.go

@@ -18,6 +18,7 @@ import (
 // Config is the top-level structure of the Beacon config file.
 type Config struct {
 	Auth        AuthConfig            `yaml:"auth"`
+	Log         LogConfig             `yaml:"log"`
 	Server      ServerConfig          `yaml:"server"`
 	IATAs       map[string]IATAConfig `yaml:"iatas"`
 	Regions     []RegionConfig        `yaml:"regions"`
@@ -43,6 +44,12 @@ type AuthConfig struct {
 	APIKey string `yaml:"api_key" json:"-"`
 }
 
+// LogConfig controls application verbosity and stderr output format.
+type LogConfig struct {
+	Level  string `yaml:"level"`
+	Format string `yaml:"format"`
+}
+
 // ServerConfig controls which direct peers may supply the client address.
 type ServerConfig struct {
 	// TrustedProxies accepts IPv4/IPv6 CIDRs; an empty list trusts no proxy.

internal/config/config_test.go

@@ -11,6 +11,20 @@ import (
 	"time"
 )
 
+func TestLoadLogConfig(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "config.yaml")
+	if err := os.WriteFile(path, []byte("log:\n  level: warn\n  format: json\n"), 0600); err != nil {
+		t.Fatal(err)
+	}
+	cfg, err := Load(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if cfg.Log.Level != "warn" || cfg.Log.Format != "json" {
+		t.Fatalf("log config ignored: %+v", cfg.Log)
+	}
+}
+
 func TestLoadTrustedProxies(t *testing.T) {
 	for _, tc := range []struct {
 		name, yaml string

internal/config/seed.go

@@ -8,7 +8,7 @@ import (
 	"crypto/sha256"
 	"encoding/json"
 	"fmt"
-	"log"
+	"log/slog"
 	"os"
 	"strings"
 )
@@ -26,7 +26,7 @@ type Seeder interface {
 // Seed applies config-defined regions, IATA overrides to the database.
 // It is safe to call on every startup — all operations are upserts.
 func Seed(ctx context.Context, cfg *Config, db Seeder) error {
-	log.Printf("config: seeding %d IATAs, %d regions, %d scopes", len(cfg.IATAs), len(cfg.Regions), len(cfg.Scopes))
+	slog.Info(fmt.Sprintf("config: seeding %d IATAs, %d regions, %d scopes", len(cfg.IATAs), len(cfg.Regions), len(cfg.Scopes)), "component", "config")
 	// IATA overrides
 	for iata, details := range cfg.IATAs {
 		if err := db.UpsertIATADetails(ctx, iata, details.Name, details.Lat, details.Lng); err != nil {

internal/hub/hub.go

@@ -17,7 +17,8 @@ package hub
 
 import (
 	"encoding/json"
-	"log"
+	"fmt"
+	"log/slog"
 	"slices"
 )
 
@@ -211,7 +212,7 @@ func (h *Hub) Broadcast(e Event) {
 	select {
 	case h.broadcast <- e:
 	default:
-		log.Println("hub: broadcast channel full, dropping event")
+		slog.Warn("hub: broadcast channel full, dropping event", "component", "hub")
 	}
 }
 
@@ -280,7 +281,7 @@ func (h *Hub) Run() {
 					default:
 						// laggedCh itself full; write pump will catch up on next drain
 					}
-					log.Printf("hub: client send buffer full, dropped event type=%s", evt.Type)
+					slog.Warn(fmt.Sprintf("hub: client send buffer full, dropped event type=%s", evt.Type), "component", "hub")
 				}
 			}
 		}

internal/ingest/backfill.go

@@ -7,7 +7,7 @@ import (
 	"context"
 	"encoding/hex"
 	"fmt"
-	"log"
+	"log/slog"
 	"strings"
 	"time"
 
@@ -81,7 +81,7 @@ func DecryptGroupText(ctx context.Context, db DB, keys ChannelKeyStore, packetHa
 		// Non-fatal: the message is stored either way, so just log and continue -- matches
 		// the live ingest path's existing behavior of not treating this as a hard failure.
 		if err := db.SetPacketDecrypted(ctx, packetHash); err != nil {
-			log.Printf("ingest: failed to set packet decrypted for %s: %v", hex.EncodeToString(packetHash), err)
+			slog.Error(fmt.Sprintf("ingest: failed to set packet decrypted for %s", hex.EncodeToString(packetHash)), "component", "ingest", "error", err)
 		}
 	}
 	return &DecryptGroupTextResult{
@@ -112,7 +112,7 @@ func BackfillChannelMessages(ctx context.Context, db DB, keys ChannelKeyStore) (
 	for _, p := range packets {
 		result, err := DecryptGroupText(ctx, db, keys, p.PacketHash, p.RawPayload)
 		if err != nil {
-			log.Printf("ingest: backfill: decrypt failed for packet %s: %v", hex.EncodeToString(p.PacketHash), err)
+			slog.Error(fmt.Sprintf("ingest: backfill: decrypt failed for packet %s", hex.EncodeToString(p.PacketHash)), "component", "ingest", "error", err)
 			continue
 		}
 		if result != nil && result.NewMessage {

internal/ingest/ingest.go

@@ -40,7 +40,8 @@ import (
 	"crypto/rand"
 	"encoding/binary"
 	"encoding/json"
-	"log"
+	"fmt"
+	"log/slog"
 	"strconv"
 	"strings"
 	"time"
@@ -216,6 +217,7 @@ type ScopeStore interface {
 
 // Worker holds the dependencies for one broker's ingest loop.
 type Worker struct {
+	log              *slog.Logger
 	cfg              Config
 	db               DB
 	hub              *hub.Hub
@@ -228,7 +230,8 @@ type Worker struct {
 
 // New creates an ingest Worker. Call Start() to connect and begin processing.
 func New(cfg Config, db DB, h *hub.Hub, keys ChannelKeyStore, scopes ScopeStore) *Worker {
-	return &Worker{cfg: cfg, db: db, hub: h, keys: keys, scopes: scopes}
+	return &Worker{cfg: cfg, db: db, hub: h, keys: keys, scopes: scopes,
+		log: slog.Default().With("component", "ingest", "broker", cfg.BrokerName)}
 }
 
 // Start connects to the broker and blocks until ctx is cancelled. It
@@ -253,22 +256,22 @@ func (w *Worker) Start(ctx context.Context) {
 		SetConnectRetry(true).
 		SetConnectRetryInterval(5 * time.Second).
 		SetOnConnectHandler(func(c mqtt.Client) {
-			log.Printf("ingest[%s]: connected to %s", w.cfg.BrokerName, w.cfg.URL)
+			w.log.Info("connected")
 			w.subscribe(c)
 		}).
 		SetConnectionLostHandler(func(_ mqtt.Client, err error) {
-			log.Printf("ingest[%s]: connection lost, will reconnect: %v", w.cfg.BrokerName, err)
+			w.log.Warn("connection lost, will reconnect", "error", err)
 		})
 
 	w.client = mqtt.NewClient(opts)
 	if tok := w.client.Connect(); tok.Wait() && tok.Error() != nil {
-		log.Printf("ingest[%s]: initial connect failed: %v", w.cfg.BrokerName, tok.Error())
+		w.log.Error("initial connect failed", "error", tok.Error())
 		// paho will retry; we fall through and wait for ctx
 	}
 
 	<-ctx.Done()
 	w.client.Disconnect(500)
-	log.Printf("ingest[%s]: stopped", w.cfg.BrokerName)
+	w.log.Info("stopped")
 }
 
 func (w *Worker) BrokerName() string {
@@ -296,7 +299,7 @@ func (w *Worker) subscribe(client mqtt.Client) {
 		w.handleMessage(msg)
 	})
 	if tok.Wait() && tok.Error() != nil {
-		log.Printf("ingest[%s]: subscribe error: %v", w.cfg.BrokerName, tok.Error())
+		w.log.Error("subscribe error", "error", tok.Error())
 	}
 }
 
@@ -329,14 +332,16 @@ func (w *Worker) handleMessage(msg mqtt.Message) {
 	// iata_codes.iata is CHAR(3); anything else would fail the DB insert
 	// downstream, so reject malformed topic segments here instead.
 	if !isValidIATA(iata) {
-		log.Printf("ingest[%s]: dropped packet with malformed IATA %q on topic %s", w.cfg.BrokerName, iata, msg.Topic())
+		w.log.Warn("dropped packet with malformed IATA", "iata", iata, "topic", msg.Topic())
 		return
 	}
 
 	// Drop packets from IATAs outside the configured geographic filter.
 	if w.cfg.AllowedIATAs != nil {
 		if _, ok := w.cfg.AllowedIATAs[iata]; !ok {
-			log.Printf("ingest[%s]: dropped packet from %s (not in allowed IATAs)", w.cfg.BrokerName, iata)
+			if w.log.Enabled(context.Background(), slog.LevelDebug) {
+				w.log.Debug(fmt.Sprintf("dropped packet from %s (not in allowed IATAs)", iata))
+			}
 			return
 		}
 	}
@@ -358,7 +363,7 @@ func (w *Worker) handleMessage(msg mqtt.Message) {
 func (w *Worker) broadcast(eventType hub.EventType, iata string, payloadType uint8, channelHash string, payload any) {
 	b, err := json.Marshal(payload)
 	if err != nil {
-		log.Printf("ingest[%s]: failed to marshal %s event: %v", w.cfg.BrokerName, eventType, err)
+		w.log.Error(fmt.Sprintf("failed to marshal %s event", eventType), "error", err)
 		return
 	}
 	w.hub.Broadcast(hub.Event{
@@ -380,13 +385,13 @@ func (w *Worker) broadcast(eventType hub.EventType, iata string, payloadType uin
 func (w *Worker) broadcastPacketObservation(iata string, payloadType uint8, evt packetObservationEvent, resolvedPath []api.ResolvedHop) {
 	base, err := json.Marshal(evt)
 	if err != nil {
-		log.Printf("ingest[%s]: failed to marshal packetObservation event: %v", w.cfg.BrokerName, err)
+		w.log.Error("failed to marshal packetObservation event", "error", err)
 		return
 	}
 	evt.Observation.ResolvedPath = resolvedPath
 	resolved, err := json.Marshal(evt)
 	if err != nil {
-		log.Printf("ingest[%s]: failed to marshal packetObservation event (resolved variant): %v", w.cfg.BrokerName, err)
+		w.log.Error("failed to marshal packetObservation event (resolved variant)", "error", err)
 		resolved = nil // fall back to base-only; not fatal
 	}
 	w.hub.Broadcast(hub.Event{

internal/ingest/ingest_test.go

@@ -8,6 +8,7 @@ import (
 	"encoding/binary"
 	"encoding/json"
 	"errors"
+	"log/slog"
 	"testing"
 	"time"
 
@@ -303,6 +304,7 @@ func (s *stubDB) UpdateObserverRegionScope(_ context.Context, _ uuid.UUID, _ str
 func newTestWorker() (*Worker, *stubDB) {
 	db := &stubDB{}
 	w := &Worker{
+		log:    slog.Default().With("component", "ingest", "broker", "test"),
 		cfg:    Config{BrokerName: "test"},
 		db:     db,
 		hub:    hub.New(),

internal/ingest/logging_test.go

@@ -0,0 +1,63 @@
+// Copyright 2026 Beacon Contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package ingest
+
+import (
+	"bytes"
+	"encoding/json"
+	"log/slog"
+	"strconv"
+	"strings"
+	"testing"
+)
+
+type malformedTopicMessage struct{ topic string }
+
+func (m malformedTopicMessage) Duplicate() bool   { return false }
+func (m malformedTopicMessage) Qos() byte         { return 0 }
+func (m malformedTopicMessage) Retained() bool    { return false }
+func (m malformedTopicMessage) Topic() string     { return m.topic }
+func (m malformedTopicMessage) MessageID() uint16 { return 0 }
+func (m malformedTopicMessage) Payload() []byte   { return nil }
+func (m malformedTopicMessage) Ack()              {}
+
+func TestMalformedIATAWarnNamesSource(t *testing.T) {
+	for _, format := range []string{"json", "text"} {
+		for _, iata := range []string{"XX", "XX\r\nforged=true"} {
+			t.Run(format+"/"+strconv.Quote(iata), func(t *testing.T) {
+				var output bytes.Buffer
+				var handler slog.Handler = slog.NewJSONHandler(&output, nil)
+				if format == "text" {
+					handler = slog.NewTextHandler(&output, nil)
+				}
+				previous := slog.Default()
+				slog.SetDefault(slog.New(handler))
+				t.Cleanup(func() { slog.SetDefault(previous) })
+				worker, _ := newTestWorker()
+				topic := "meshcore/" + iata + "/0011223344556677/packets"
+				worker.handleMessage(malformedTopicMessage{topic: topic})
+				if bytes.Count(output.Bytes(), []byte("\n")) != 1 || bytes.ContainsRune(output.Bytes(), '\r') {
+					t.Fatalf("warning did not remain one escaped record: %q", output.String())
+				}
+				if format == "json" {
+					var record map[string]any
+					if err := json.Unmarshal(output.Bytes(), &record); err != nil {
+						t.Fatal(err)
+					}
+					if record["iata"] != iata || record["topic"] != topic || record["msg"] != "dropped packet with malformed IATA" {
+						t.Fatalf("warning lost its source: %v", record)
+					}
+				} else {
+					expectedIATA, expectedTopic := iata, topic
+					if strings.ContainsAny(iata, "\r\n") {
+						expectedIATA, expectedTopic = strconv.Quote(iata), strconv.Quote(topic)
+					}
+					if !strings.Contains(output.String(), "iata="+expectedIATA) || !strings.Contains(output.String(), "topic="+expectedTopic) {
+						t.Fatalf("warning lost its source: %q", output.String())
+					}
+				}
+			})
+		}
+	}
+}

internal/ingest/neighbors.go

@@ -7,7 +7,7 @@ import (
 	"context"
 	"encoding/hex"
 	"encoding/json"
-	"log"
+	"fmt"
 )
 
 // neighborReportEntry is one entry in the "neighbors" array of a /neighbors report.
@@ -33,30 +33,30 @@ type neighborReport struct {
 func (w *Worker) handleNeighbors(ctx context.Context, iata, pubkeyHex string, raw []byte) {
 	var report neighborReport
 	if err := json.Unmarshal(raw, &report); err != nil {
-		log.Printf("ingest[%s]: malformed neighbors envelope from %s: %v", w.cfg.BrokerName, pubkeyHex, err)
+		w.log.Warn(fmt.Sprintf("malformed neighbors envelope from %s", pubkeyHex), "error", err)
 		return
 	}
 
 	pubkey, err := hex.DecodeString(pubkeyHex)
 	if err != nil {
-		log.Printf("ingest[%s]: invalid pubkey hex in neighbors from %s: %v", w.cfg.BrokerName, pubkeyHex, err)
+		w.log.Warn(fmt.Sprintf("invalid pubkey hex in neighbors from %s", pubkeyHex), "error", err)
 		return
 	}
 
 	observerID, _, err := w.db.UpsertObserver(ctx, pubkey)
 	if err != nil {
-		log.Printf("ingest[%s]: db: upsert observer failed in neighbors from %s: %v", w.cfg.BrokerName, pubkeyHex, err)
+		w.log.Error(fmt.Sprintf("db: upsert observer failed in neighbors from %s", pubkeyHex), "error", err)
 		return
 	}
 	if err := w.db.UpsertObserverBroker(ctx, observerID, w.cfg.BrokerName); err != nil {
-		log.Printf("ingest[%s]: db: upsert observer broker failed in neighbors from %s: %v", w.cfg.BrokerName, pubkeyHex, err)
+		w.log.Error(fmt.Sprintf("db: upsert observer broker failed in neighbors from %s", pubkeyHex), "error", err)
 	}
 
 	// self.scopes is always known (it's the observer's own config, not an OTA
 	// query), so this is an unconditional write -- unlike the per-neighbor
 	// scopes below, there's no "query failed" case to protect against here.
 	if err := w.db.UpdateObserverRegionScope(ctx, observerID, report.Self.Scopes); err != nil {
-		log.Printf("ingest[%s]: db: update observer region scope failed for %s: %v", w.cfg.BrokerName, pubkeyHex, err)
+		w.log.Error(fmt.Sprintf("db: update observer region scope failed for %s", pubkeyHex), "error", err)
 	}
 
 	// The observer's own node row (as opposed to its observers row above) is
@@ -72,7 +72,7 @@ func (w *Worker) handleNeighbors(ctx context.Context, iata, pubkeyHex string, ra
 	for _, n := range report.Neighbors {
 		neighborPubkey, err := hex.DecodeString(n.PubKey)
 		if err != nil {
-			log.Printf("ingest[%s]: invalid neighbor pubkey hex %q from %s: %v", w.cfg.BrokerName, n.PubKey, pubkeyHex, err)
+			w.log.Warn(fmt.Sprintf("invalid neighbor pubkey hex %q from %s", n.PubKey, pubkeyHex), "error", err)
 			continue
 		}
 		neighborNodeID, err := w.db.GetNodeByPubkey(ctx, neighborPubkey)
@@ -98,7 +98,7 @@ func (w *Worker) handleNeighbors(ctx context.Context, iata, pubkeyHex string, ra
 		}
 
 		if err := w.db.UpsertNodeNeighbor(ctx, observerNodeID, neighborNodeID, iata, &snr, regionScope); err != nil {
-			log.Printf("ingest[%s]: db: upsert neighbor failed for %s -> %s: %v", w.cfg.BrokerName, pubkeyHex, n.PubKey, err)
+			w.log.Error(fmt.Sprintf("db: upsert neighbor failed for %s -> %s", pubkeyHex, n.PubKey), "error", err)
 		}
 	}
 }

internal/ingest/packet.go

@@ -11,7 +11,6 @@ import (
 	"encoding/hex"
 	"encoding/json"
 	"fmt"
-	"log"
 	"strings"
 	"time"
 
@@ -290,46 +289,46 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [
 	}
 	err := json.Unmarshal(raw, &envelope)
 	if err != nil {
-		log.Printf("ingest[%s]: malformed packet envelope from %s/%s", w.cfg.BrokerName, iata, pubkeyHex)
+		w.log.Warn(fmt.Sprintf("malformed packet envelope from %s/%s", iata, pubkeyHex))
 		return
 	}
 	if envelope.Raw == "" {
 		if envelope.Len == "0" || envelope.Len == "" {
 			return // observer keepalive with no packet data
 		}
-		log.Printf("ingest[%s]: malformed packet envelope from %s/%s", w.cfg.BrokerName, iata, pubkeyHex)
+		w.log.Warn(fmt.Sprintf("malformed packet envelope from %s/%s", iata, pubkeyHex))
 		return
 	}
 	hexBytes, err := hex.DecodeString(strings.ReplaceAll(envelope.Raw, " ", ""))
 	if err != nil {
-		log.Printf("ingest[%s]: invalid hex from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err)
+		w.log.Warn(fmt.Sprintf("invalid hex from %s/%s", iata, pubkeyHex), "error", err)
 		return
 	}
 	packet, err := meshcore.PacketFromBytes(hexBytes)
 	if err != nil {
-		log.Printf("ingest[%s]: error decoding packet from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err)
+		w.log.Warn(fmt.Sprintf("error decoding packet from %s/%s", iata, pubkeyHex), "error", err)
 		return
 	}
 
 	pubkeyBytes, err := hex.DecodeString(pubkeyHex)
 	if err != nil {
-		log.Printf("ingest[%s]: invalid pubkey hex from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err)
+		w.log.Warn(fmt.Sprintf("invalid pubkey hex from %s/%s", iata, pubkeyHex), "error", err)
 		return
 	}
 
 	id, observerName, err := w.db.UpsertObserver(ctx, pubkeyBytes)
 	if err != nil {
-		log.Printf("ingest[%s]: db: upsert observer failed with packet from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err)
+		w.log.Error(fmt.Sprintf("db: upsert observer failed with packet from %s/%s", iata, pubkeyHex), "error", err)
 		return
 	}
 	err = w.db.UpsertObserverBroker(ctx, id, w.cfg.BrokerName)
 	if err != nil {
-		log.Printf("ingest[%s]: db: update observer broker failed with packet from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err)
+		w.log.Error(fmt.Sprintf("db: update observer broker failed with packet from %s/%s", iata, pubkeyHex), "error", err)
 		return
 	}
 	err = w.db.UpsertIATA(ctx, iata)
 	if err != nil {
-		log.Printf("ingest[%s]: db: upsert IATA failed with packet from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err)
+		w.log.Error(fmt.Sprintf("db: upsert IATA failed with packet from %s/%s", iata, pubkeyHex), "error", err)
 		return
 	}
 	packetHash := packet.PacketHash()
@@ -584,7 +583,7 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [
 						if len(prevEntries) == 1 && len(currEntries) == 1 {
 							snr := snrValues[i]
 							if err := w.db.UpsertNodeNeighbor(ctx, currEntries[0].NodeID, prevEntries[0].NodeID, iata, &snr, nil); err != nil {
-								log.Printf("ingest[%s]: failed to upsert trace neighbor: %v", w.cfg.BrokerName, err)
+								w.log.Error("failed to upsert trace neighbor", "error", err)
 							}
 						}
 					}
@@ -664,7 +663,7 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [
 						if oErr == nil && rErr == nil && observerNodeID != responderNodeID {
 							rxSNR := float32(parseNumber(envelope.SNR))
 							if err := w.db.UpsertNodeNeighbor(ctx, observerNodeID, responderNodeID, iata, &rxSNR, nil); err != nil {
-								log.Printf("ingest[%s]: failed to upsert observer-discover neighbor: %v", w.cfg.BrokerName, err)
+								w.log.Error("failed to upsert observer-discover neighbor", "error", err)
 							}
 						}
 					}
@@ -703,7 +702,7 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [
 	if matchedScope != nil {
 		id, err := w.db.GetTransportScopeByName(ctx, *matchedScope)
 		if err != nil {
-			log.Printf("ingest[%s]: failed to get scope ID for %s: %v", w.cfg.BrokerName, *matchedScope, err)
+			w.log.Error(fmt.Sprintf("failed to get scope ID for %s", *matchedScope), "error", err)
 		} else {
 			scopeID = &id
 		}
@@ -728,7 +727,7 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [
 	}
 	isNew, err := w.db.UpsertPacket(ctx, pParams)
 	if err != nil {
-		log.Printf("ingest[%s]: db: upsert packet failed from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err)
+		w.log.Error(fmt.Sprintf("db: upsert packet failed from %s/%s", iata, pubkeyHex), "error", err)
 		return
 	}
 	// Try parsing with timezone offset first
@@ -746,21 +745,21 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [
 		heardAt, err = time.Parse("2006-01-02T15:04:05Z", envelope.Timestamp)
 	}
 	if err != nil {
-		log.Printf("ingest[%s]: failed to parse timestamp %q: %v", w.cfg.BrokerName, envelope.Timestamp, err)
+		w.log.Warn(fmt.Sprintf("failed to parse timestamp %q", envelope.Timestamp), "error", err)
 		heardAt = time.Now().UTC()
 	} else {
 		// clamp to server time if offset is suspicious (> 30 min drift)
 		now := time.Now().UTC()
 		diff := heardAt.UTC().Sub(now)
 		if diff > 30*time.Minute || diff < -30*time.Minute {
-			log.Printf("ingest[%s]: clamping suspicious timestamp %s (diff %v) for pubkey %s", w.cfg.BrokerName, envelope.Timestamp, diff, pubkeyHex[:8])
+			w.log.Warn(fmt.Sprintf("clamping suspicious timestamp %s (diff %v) for pubkey %s", envelope.Timestamp, diff, pubkeyHex[:8]))
 			heardAt = now
 		}
 	}
 
 	radio, err := w.db.GetObserverRadio(ctx, id)
 	if err != nil {
-		log.Printf("ingest[%s]: db: get observer radio failed for %s: %v", w.cfg.BrokerName, pubkeyHex, err)
+		w.log.Error(fmt.Sprintf("db: get observer radio failed for %s", pubkeyHex), "error", err)
 	}
 	// Save the same endpoint resolution used by the live event in the observation INSERT.
 	var resolvedSource, resolvedDestination *api.ResolvedHop
@@ -790,7 +789,7 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [
 	if snapshot.HasResolvedNodes() {
 		resolvedEndpoints, err = json.Marshal(snapshot)
 		if err != nil {
-			log.Printf("ingest[%s]: endpoint snapshot encoding failed: %v", w.cfg.BrokerName, err)
+			w.log.Error("endpoint snapshot encoding failed", "error", err)
 			resolvedEndpoints = nil // optional enrichment must not discard the observation
 		}
 	}
@@ -824,27 +823,27 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [
 	}
 	inserted, err := w.db.InsertObservation(ctx, oParams)
 	if err != nil {
-		log.Printf("ingest[%s]: db: insert observation failed from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err)
+		w.log.Error(fmt.Sprintf("db: insert observation failed from %s/%s", iata, pubkeyHex), "error", err)
 		return
 	}
 
 	if scopeID != nil && inserted {
 		if err := w.db.UpsertObserverScope(ctx, id, *scopeID); err != nil {
-			log.Printf("ingest[%s]: failed to upsert observer scope for %s: %v", w.cfg.BrokerName, id, err)
+			w.log.Error(fmt.Sprintf("failed to upsert observer scope for %s", id), "error", err)
 		}
 	}
 
 	// Runs on duplicate observations too; the upsert only writes when the row is >1h stale.
 	if channelHash != nil {
 		if err := w.db.UpsertChannelIATA(ctx, channelHash, iata, heardAt); err != nil {
-			log.Printf("ingest[%s]: db: upsert channel IATA failed from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err)
+			w.log.Error(fmt.Sprintf("db: upsert channel IATA failed from %s/%s", iata, pubkeyHex), "error", err)
 		}
 	}
 
 	// Runs on duplicate observations too; the upsert only writes when the row is >1h stale.
 	if traceTag != nil {
 		if err := w.db.UpsertTraceIATA(ctx, traceTag, iata, heardAt); err != nil {
-			log.Printf("ingest[%s]: db: upsert trace IATA failed from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err)
+			w.log.Error(fmt.Sprintf("db: upsert trace IATA failed from %s/%s", iata, pubkeyHex), "error", err)
 		}
 	}
 
@@ -858,7 +857,7 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [
 	}
 	resolved, err := w.db.ResolvePathHashes(ctx, iata, hashes)
 	if err != nil {
-		log.Printf("ingest[%s]: path resolution failed: %v", w.cfg.BrokerName, err)
+		w.log.Error("path resolution failed", "error", err)
 	}
 	var resolvedIDs []uuid.UUID
 	for _, entries := range resolved {
@@ -882,7 +881,7 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [
 		}
 		if allHigh && len(nodeIDs) > 1 {
 			if err := w.db.UpsertKnownRoute(ctx, nodeIDs, hashPrefixes, iata, int32(len(nodeIDs))); err != nil {
-				log.Printf("ingest[%s]: failed to upsert known route: %v", w.cfg.BrokerName, err)
+				w.log.Error("failed to upsert known route", "error", err)
 			}
 		}
 	}
@@ -912,7 +911,7 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [
 		evt.Observation.PropagationTimeMs = 0 // not yet calculated
 		count, err := w.db.GetPacketObservationCount(ctx, packetHash[:])
 		if err != nil {
-			log.Printf("ingest[%s]: failed to get observation count: %v", w.cfg.BrokerName, err)
+			w.log.Error("failed to get observation count", "error", err)
 			count = 0
 		}
 		evt.Packet.ObservationCount = count

internal/ingest/side_effects.go

@@ -7,7 +7,6 @@ import (
 	"context"
 	"encoding/hex"
 	"fmt"
-	"log"
 	"strings"
 	"time"
 
@@ -72,11 +71,11 @@ func (w *Worker) handlePayloadTypeSideEffects(ctx context.Context, packet *meshc
 	if packet.PayloadType() == meshcore.PayloadTypeAdvert {
 		advert, err := meshcore.AdvertFromBytes(packet.Payload)
 		if err != nil {
-			log.Printf("ingest[%s]: error decoding advert payload: %v", w.cfg.BrokerName, err)
+			w.log.Warn("error decoding advert payload", "error", err)
 			return
 		}
 		if !advert.Verify() {
-			log.Printf("ingest[%s]: dropped advert with invalid signature from pubkey %s", w.cfg.BrokerName, hex.EncodeToString(advert.PublicKey.PublicKeyBytes()))
+			w.log.Warn(fmt.Sprintf("dropped advert with invalid signature from pubkey %s", hex.EncodeToString(advert.PublicKey.PublicKeyBytes())))
 			return
 		}
 		var lat, lon *float64
@@ -101,7 +100,7 @@ func (w *Worker) handlePayloadTypeSideEffects(ctx context.Context, packet *meshc
 		}
 		nodeID, err := w.db.UpsertNode(ctx, params, nodeRadio)
 		if err != nil {
-			log.Printf("ingest[%s]: db: upsert node failed: %v", w.cfg.BrokerName, err)
+			w.log.Error("db: upsert node failed", "error", err)
 			return
 		}
 		// invalidate cache for this node
@@ -118,7 +117,7 @@ func (w *Worker) handlePayloadTypeSideEffects(ctx context.Context, packet *meshc
 					key := hex.EncodeToString(firstHop[0])
 					if entries := resolved[key]; len(entries) == 1 {
 						if err := w.db.UpsertNodeNeighbor(ctx, nodeID, entries[0].NodeID, iata, nil, nil); err != nil {
-							log.Printf("ingest[%s]: failed to upsert node neighbor: %v", w.cfg.BrokerName, err)
+							w.log.Error("failed to upsert node neighbor", "error", err)
 						}
 					}
 				}
@@ -133,21 +132,21 @@ func (w *Worker) handlePayloadTypeSideEffects(ctx context.Context, packet *meshc
 			if oErr == nil && observerNodeID != nodeID {
 				snr := rxSNR
 				if err := w.db.UpsertNodeNeighbor(ctx, observerNodeID, nodeID, iata, &snr, nil); err != nil {
-					log.Printf("ingest[%s]: failed to upsert observer-advert neighbor: %v", w.cfg.BrokerName, err)
+					w.log.Error("failed to upsert observer-advert neighbor", "error", err)
 				}
 			}
 		}
 		if err := w.db.UpsertNodeIATA(ctx, nodeID, iata); err != nil {
-			log.Printf("ingest[%s]: db: upsert node IATA failed: %v", w.cfg.BrokerName, err)
+			w.log.Error("db: upsert node IATA failed", "error", err)
 		}
 		if scopeID != nil && (packet.RouteType() == meshcore.RouteTypeTransportFlood || packet.RouteType() == meshcore.RouteTypeTransportDirect) {
 			if err := w.db.SetNodeDefaultScope(ctx, nodeID, *scopeID); err != nil {
-				log.Printf("ingest[%s]: failed to set default scope for node %s: %v", w.cfg.BrokerName, hex.EncodeToString(advert.PublicKey.PublicKeyBytes()), err)
+				w.log.Error(fmt.Sprintf("failed to set default scope for node %s", hex.EncodeToString(advert.PublicKey.PublicKeyBytes())), "error", err)
 			}
 		}
 		prefix4 := advert.PublicKey.PublicKeyBytes()[:4]
 		if err := w.db.UpsertNodeShortID(ctx, nodeID, iata, prefix4); err != nil {
-			log.Printf("ingest[%s]: failed to upsert node short ID for %s: %v", w.cfg.BrokerName, hex.EncodeToString(prefix4), err)
+			w.log.Error(fmt.Sprintf("failed to upsert node short ID for %s", hex.EncodeToString(prefix4)), "error", err)
 		}
 		pubkeyHex := hex.EncodeToString(advert.PublicKey.PublicKeyBytes())
 		isObserver := w.db.IsObserverByPubkey(ctx, advert.PublicKey.PublicKeyBytes())
@@ -180,14 +179,14 @@ func (w *Worker) handlePayloadTypeSideEffects(ctx context.Context, packet *meshc
 	if packet.PayloadType() == meshcore.PayloadTypeGrpTxt {
 		grpTxt, err := meshcore.GroupTextFromBytes(packet.Payload)
 		if err != nil {
-			log.Printf("ingest[%s]: error decoding group text payload: %v", w.cfg.BrokerName, err)
+			w.log.Warn("error decoding group text payload", "error", err)
 			return
 		}
 		channelHashBytes := []byte{grpTxt.ChannelHash}
 
 		result, err := DecryptGroupText(ctx, w.db, w.keys, packetHash, packet.Payload)
 		if err != nil {
-			log.Printf("ingest[%s]: decrypt group text failed: %v", w.cfg.BrokerName, err)
+			w.log.Error("decrypt group text failed", "error", err)
 			return
 		}
 		if result == nil {

internal/ingest/status.go

@@ -8,7 +8,6 @@ import (
 	"encoding/hex"
 	"encoding/json"
 	"fmt"
-	"log"
 	"strconv"
 	"strings"
 	"time"
@@ -71,17 +70,17 @@ func (w *Worker) handleStatus(ctx context.Context, pubkeyHex string, raw []byte)
 		} `json:"stats"`
 	}
 	if err := json.Unmarshal(raw, &envelope); err != nil {
-		log.Printf("ingest[%s]: malformed status envelope from %s: %v", w.cfg.BrokerName, pubkeyHex, err)
+		w.log.Warn(fmt.Sprintf("malformed status envelope from %s", pubkeyHex), "error", err)
 		return
 	}
 	pubkey, err := hex.DecodeString(pubkeyHex)
 	if err != nil {
-		log.Printf("ingest[%s]: invalid pubkey hex in status from %s: %v", w.cfg.BrokerName, pubkeyHex, err)
+		w.log.Warn(fmt.Sprintf("invalid pubkey hex in status from %s", pubkeyHex), "error", err)
 		return
 	}
 	id, _, err := w.db.UpsertObserver(ctx, pubkey)
 	if err != nil {
-		log.Printf("ingest[%s]: db: upsert observer failed in status from %s: %v", w.cfg.BrokerName, pubkeyHex, err)
+		w.log.Error(fmt.Sprintf("db: upsert observer failed in status from %s", pubkeyHex), "error", err)
 		return
 	}
 	// invalidate cache for observer details
@@ -89,7 +88,7 @@ func (w *Worker) handleStatus(ctx context.Context, pubkeyHex string, raw []byte)
 		w.onObserverUpsert(ctx, id)
 	}
 	if err := w.db.UpsertObserverBroker(ctx, id, w.cfg.BrokerName); err != nil {
-		log.Printf("ingest[%s]: db: upsert observer broker failed in status from %s: %v", w.cfg.BrokerName, pubkeyHex, err)
+		w.log.Error(fmt.Sprintf("db: upsert observer broker failed in status from %s", pubkeyHex), "error", err)
 	}
 	params := UpdateObserverStatusParams{
 		PublicKey:      pubkey,
@@ -122,29 +121,29 @@ func (w *Worker) handleStatus(ctx context.Context, pubkeyHex string, raw []byte)
 
 	radio := strings.Split(strings.TrimSpace(envelope.RadioString), ",")
 	if len(radio) != 4 {
-		log.Printf("ingest[%s]: missing or malformed radio params in status from %s, skipping radio fields", w.cfg.BrokerName, pubkeyHex)
+		w.log.Warn(fmt.Sprintf("missing or malformed radio params in status from %s, skipping radio fields", pubkeyHex))
 	} else {
 		freq, err := strconv.ParseFloat(radio[0], 32)
 		if err != nil {
-			log.Printf("ingest[%s]: error parsing radio freq in status from %s: %v", w.cfg.BrokerName, pubkeyHex, err)
+			w.log.Warn(fmt.Sprintf("error parsing radio freq in status from %s", pubkeyHex), "error", err)
 		} else {
 			params.RadioFreqMHz = float32(freq)
 		}
 		bw, err := strconv.ParseFloat(radio[1], 32)
 		if err != nil {
-			log.Printf("ingest[%s]: error parsing radio bw in status from %s: %v", w.cfg.BrokerName, pubkeyHex, err)
+			w.log.Warn(fmt.Sprintf("error parsing radio bw in status from %s", pubkeyHex), "error", err)
 		} else {
 			params.RadioBWKHz = float32(bw)
 		}
 		sf, err := strconv.ParseInt(radio[2], 10, 16)
 		if err != nil {
-			log.Printf("ingest[%s]: error parsing radio sf in status from %s: %v", w.cfg.BrokerName, pubkeyHex, err)
+			w.log.Warn(fmt.Sprintf("error parsing radio sf in status from %s", pubkeyHex), "error", err)
 		} else {
 			params.RadioSF = int16(sf)
 		}
 		cr, err := strconv.ParseInt(radio[3], 10, 16)
 		if err != nil {
-			log.Printf("ingest[%s]: error parsing radio cr in status from %s: %v", w.cfg.BrokerName, pubkeyHex, err)
+			w.log.Warn(fmt.Sprintf("error parsing radio cr in status from %s", pubkeyHex), "error", err)
 		} else {
 			params.RadioCR = int16(cr)
 		}
@@ -152,7 +151,7 @@ func (w *Worker) handleStatus(ctx context.Context, pubkeyHex string, raw []byte)
 
 	observerID, err := w.db.UpdateObserverStatus(ctx, params)
 	if err != nil {
-		log.Printf("ingest[%s]: db: update observer status failed for %s: %v", w.cfg.BrokerName, pubkeyHex, err)
+		w.log.Error(fmt.Sprintf("db: update observer status failed for %s", pubkeyHex), "error", err)
 		return
 	}
 	// Store a telemetry snapshot at the configured resolution.
@@ -161,7 +160,7 @@ func (w *Worker) handleStatus(ctx context.Context, pubkeyHex string, raw []byte)
 	// stats — skip the insert rather than writing an all-zero row that would win
 	// the hourly dedup and pollute the telemetry aggregates.
 	if envelope.Stats.UptimeSeconds == 0 {
-		log.Printf("ingest[%s]: status from %s has no usable stats (uptime_secs missing or zero), skipping telemetry insert", w.cfg.BrokerName, pubkeyHex)
+		w.log.Debug("status has no usable stats; skipping telemetry insert")
 	} else {
 		resolution := w.cfg.TelemetryResolution
 		if resolution == 0 {
@@ -180,7 +179,7 @@ func (w *Worker) handleStatus(ctx context.Context, pubkeyHex string, raw []byte)
 			envelope.Stats.NoiseFloor, envelope.Stats.UptimeSeconds,
 			&queueLen, &debugFlags, &recvErrors,
 		); err != nil {
-			log.Printf("ingest[%s]: db: insert telemetry failed for %s: %v", w.cfg.BrokerName, pubkeyHex, err)
+			w.log.Error(fmt.Sprintf("db: insert telemetry failed for %s", pubkeyHex), "error", err)
 		}
 	}
 
@@ -190,7 +189,7 @@ func (w *Worker) handleStatus(ctx context.Context, pubkeyHex string, raw []byte)
 	}
 	scopes, err := w.db.GetObserverScopes(ctx, observerID)
 	if err != nil {
-		log.Printf("ingest[%s]: failed to get observer scopes for %s: %v", w.cfg.BrokerName, pubkeyHex, err)
+		w.log.Error(fmt.Sprintf("failed to get observer scopes for %s", pubkeyHex), "error", err)
 		scopes = []string{}
 	}
 	var radioStr *string
@@ -216,7 +215,7 @@ func (w *Worker) handleStatus(ctx context.Context, pubkeyHex string, raw []byte)
 	}
 	payload, err := json.Marshal(evt)
 	if err != nil {
-		log.Printf("ingest[%s]: failed to marshal status event payload for %s: %v", w.cfg.BrokerName, pubkeyHex, err)
+		w.log.Error(fmt.Sprintf("failed to marshal status event payload for %s", pubkeyHex), "error", err)
 		return
 	}
 	w.hub.Broadcast(hub.Event{Type: hub.EventObserverStatus, Payload: payload, IATA: iata})

internal/logging/logging.go

@@ -0,0 +1,47 @@
+// Copyright 2026 Beacon Contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package logging
+
+import (
+	"fmt"
+	"io"
+	"log/slog"
+	"os"
+	"strings"
+
+	"github.com/MeshCore-Beacon/beacon-server/internal/config"
+)
+
+// New resolves environment overrides, validates options and writes to out.
+// Empty settings default to info/text; callers install the logger before starting workers.
+func New(out io.Writer, cfg config.LogConfig) (*slog.Logger, error) {
+	if value := os.Getenv("LOG_LEVEL"); value != "" {
+		cfg.Level = value
+	}
+	if value := os.Getenv("LOG_FORMAT"); value != "" {
+		cfg.Format = value
+	}
+	var level slog.Level
+	switch strings.ToLower(strings.TrimSpace(cfg.Level)) {
+	case "", "info":
+		level = slog.LevelInfo
+	case "debug":
+		level = slog.LevelDebug
+	case "warn":
+		level = slog.LevelWarn
+	case "error":
+		level = slog.LevelError
+	default:
+		return nil, fmt.Errorf("log level must be debug, info, warn or error")
+	}
+	opts := &slog.HandlerOptions{Level: level}
+	switch strings.ToLower(strings.TrimSpace(cfg.Format)) {
+	case "", "text":
+		return slog.New(slog.NewTextHandler(out, opts)), nil
+	case "json":
+		return slog.New(slog.NewJSONHandler(out, opts)), nil
+	default:
+		return nil, fmt.Errorf("log format must be text or json")
+	}
+}

internal/logging/logging_test.go

@@ -0,0 +1,97 @@
+// Copyright 2026 Beacon Contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package logging
+
+import (
+	"bytes"
+	"encoding/json"
+	"log/slog"
+	"strings"
+	"testing"
+
+	"github.com/MeshCore-Beacon/beacon-server/internal/config"
+)
+
+func TestLevelsAndFormats(t *testing.T) {
+	for _, format := range []string{"text", "json"} {
+		for i, level := range []string{"debug", "info", "warn", "error"} {
+			t.Run(format+"/"+level, func(t *testing.T) {
+				t.Setenv("LOG_LEVEL", "")
+				t.Setenv("LOG_FORMAT", "")
+				var out bytes.Buffer
+				logger, err := New(&out, config.LogConfig{Level: level, Format: format})
+				if err != nil {
+					t.Fatal(err)
+				}
+				logger = logger.With("component", "ingest", "broker", "test")
+				logger.Debug("debug message")
+				logger.Info("info message")
+				logger.Warn("warn message")
+				logger.Error("error message")
+				lines := strings.Split(strings.TrimSpace(out.String()), "\n")
+				if len(lines) != 4-i {
+					t.Fatalf("got %d records, want %d: %s", len(lines), 4-i, out.String())
+				}
+				for _, line := range lines {
+					if format == "json" {
+						var record map[string]any
+						if err := json.Unmarshal([]byte(line), &record); err != nil {
+							t.Fatal(err)
+						}
+						if record["component"] != "ingest" || record["broker"] != "test" || record["time"] == nil {
+							t.Fatalf("missing fields: %v", record)
+						}
+					} else if !strings.Contains(line, "component=ingest") || !strings.Contains(line, "broker=test") {
+						t.Fatalf("missing fields: %s", line)
+					}
+				}
+				if !strings.Contains(out.String(), "error message") {
+					t.Fatal("error suppressed")
+				}
+			})
+		}
+	}
+}
+
+func TestDefaultsAndEnvironment(t *testing.T) {
+	t.Setenv("LOG_LEVEL", "")
+	t.Setenv("LOG_FORMAT", "")
+	var out bytes.Buffer
+	logger, err := New(&out, config.LogConfig{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if logger.Enabled(t.Context(), slog.LevelDebug) || !logger.Enabled(t.Context(), slog.LevelInfo) {
+		t.Fatal("default must be info")
+	}
+	t.Setenv("LOG_LEVEL", " WARN ")
+	t.Setenv("LOG_FORMAT", "JSON")
+	logger, err = New(&out, config.LogConfig{Level: "debug", Format: "text"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	logger.Info("hidden")
+	logger.Error("visible")
+	var record map[string]any
+	if err = json.Unmarshal(bytes.TrimSpace(out.Bytes()), &record); err != nil {
+		t.Fatal(err)
+	}
+	if record["level"] != "ERROR" || record["msg"] != "visible" {
+		t.Fatalf("env overrides ignored: %v", record)
+	}
+}
+
+func TestInvalidOptions(t *testing.T) {
+	for _, cfg := range []config.LogConfig{{Level: "verbose"}, {Level: "off"}, {Level: "INFO+2"}, {Format: "xml"}} {
+		t.Setenv("LOG_LEVEL", "")
+		t.Setenv("LOG_FORMAT", "")
+		if _, err := New(&bytes.Buffer{}, cfg); err == nil {
+			t.Fatalf("accepted invalid options: %+v", cfg)
+		}
+	}
+	t.Setenv("LOG_LEVEL", "invalid")
+	if _, err := New(&bytes.Buffer{}, config.LogConfig{Level: "info"}); err == nil {
+		t.Fatal("invalid environment override accepted")
+	}
+}

internal/presence/coalescer.go

@@ -11,7 +11,8 @@ package presence
 
 import (
 	"context"
-	"log"
+	"fmt"
+	"log/slog"
 	"sync"
 	"time"
 
@@ -258,7 +259,7 @@ func (c *Coalescer) Flush(ctx context.Context) {
 	c.mu.Unlock()
 
 	if err := c.flushObservers(ctx, observers); err != nil {
-		log.Printf("presence: flush observers failed (%d rows dropped): %v", len(observers), err)
+		slog.Error(fmt.Sprintf("presence: flush observers failed (%d rows dropped)", len(observers)), "component", "presence", "error", err)
 	}
 
 	if len(brokers) > 0 {
@@ -271,7 +272,7 @@ func (c *Coalescer) Flush(ctx context.Context) {
 			seen = append(seen, ts)
 		}
 		if err := c.Store.TouchObserverBrokers(ctx, ids, names, seen); err != nil {
-			log.Printf("presence: flush observer brokers failed (%d rows dropped): %v", len(ids), err)
+			slog.Error(fmt.Sprintf("presence: flush observer brokers failed (%d rows dropped)", len(ids)), "component", "presence", "error", err)
 		}
 	}
 
@@ -283,7 +284,7 @@ func (c *Coalescer) Flush(ctx context.Context) {
 			heard = append(heard, ts)
 		}
 		if err := c.Store.TouchPackets(ctx, hashes, heard); err != nil {
-			log.Printf("presence: flush packets failed (%d rows dropped): %v", len(hashes), err)
+			slog.Error(fmt.Sprintf("presence: flush packets failed (%d rows dropped)", len(hashes)), "component", "presence", "error", err)
 		}
 	}
 }

internal/ws/handler.go

@@ -31,7 +31,8 @@ package ws
 import (
 	"context"
 	"encoding/json"
-	"log"
+	"fmt"
+	"log/slog"
 	"net"
 	"net/http"
 	"strconv"
@@ -59,14 +60,14 @@ func Handler(h *hub.Hub, reader api.Reader, maxConnsPerIP int) http.HandlerFunc
 			ip = host
 		}
 		if !limiter.acquire(ip) {
-			log.Printf("ws: connection limit reached for IP %s", ip)
+			slog.Warn(fmt.Sprintf("ws: connection limit reached for IP %s", ip), "component", "ws")
 			http.Error(w, "too many connections from this IP", http.StatusTooManyRequests)
 			return
 		}
 		defer limiter.release(ip)
 		conn, err := websocket.Accept(w, r, nil)
 		if err != nil {
-			log.Printf("ws: failed to accept connection: %v", err)
+			slog.Warn("ws: failed to accept connection", "component", "ws", "error", err)
 			return
 		}
 
@@ -85,10 +86,10 @@ func Handler(h *hub.Hub, reader api.Reader, maxConnsPerIP int) http.HandlerFunc
 			"connectionId": connID,
 		}
 		helloBytes, _ := json.Marshal(hello)
-		log.Printf("ws[%s]: connected, hello: %s", connID, helloBytes)
+		slog.Debug("connected", "component", "ws", "connection_id", connID)
 		err = conn.Write(ctx, websocket.MessageText, helloBytes)
 		if err != nil {
-			log.Printf("ws[%s]: failed to send hello: %v", connID, err)
+			slog.Warn(fmt.Sprintf("ws[%s]: failed to send hello", connID), "component", "ws", "error", err)
 			return
 		}
 
@@ -109,7 +110,7 @@ func Handler(h *hub.Hub, reader api.Reader, maxConnsPerIP int) http.HandlerFunc
 					msgBytes, _ := json.Marshal(msg)
 					err = conn.Write(ctx, websocket.MessageText, msgBytes)
 					if err != nil {
-						log.Printf("ws[%s]: failed to write hub event: %v", connID, err)
+						slog.Warn(fmt.Sprintf("ws[%s]: failed to write hub event", connID), "component", "ws", "error", err)
 						cancel()
 						return
 					}
@@ -126,7 +127,7 @@ func Handler(h *hub.Hub, reader api.Reader, maxConnsPerIP int) http.HandlerFunc
 					}
 					lagBytes, _ := json.Marshal(lagged)
 					if err := conn.Write(ctx, websocket.MessageText, lagBytes); err != nil {
-						log.Printf("ws[%s]: failed to write lagged notice: %v", connID, err)
+						slog.Warn(fmt.Sprintf("ws[%s]: failed to write lagged notice", connID), "component", "ws", "error", err)
 						cancel()
 						return
 					}
@@ -144,7 +145,7 @@ func Handler(h *hub.Hub, reader api.Reader, maxConnsPerIP int) http.HandlerFunc
 			_, msgBytes, err := conn.Read(readCtx)
 			readCancel()
 			if err != nil {
-				log.Printf("ws[%s]: read error: %v", connID, err)
+				slog.Warn(fmt.Sprintf("ws[%s]: read error", connID), "component", "ws", "error", err)
 				return
 			}
 			handleClientMessage(ctx, client, reader, h, conn, connID, msgBytes)
@@ -182,7 +183,7 @@ type subscribeScope struct {
 func handleClientMessage(ctx context.Context, client *hub.Client, reader api.Reader, h *hub.Hub, conn *websocket.Conn, connID string, raw []byte) {
 	var msg clientMessage
 	if err := json.Unmarshal(raw, &msg); err != nil {
-		log.Printf("ws[%s]: bad message: %v", connID, err)
+		slog.Warn("bad message", "component", "ws", "connection_id", connID, "error", err)
 		return
 	}
 
@@ -195,12 +196,12 @@ func handleClientMessage(ctx context.Context, client *hub.Client, reader api.Rea
 		for _, ridStr := range msg.Scope.RegionIDs {
 			rid, err := strconv.ParseInt(ridStr, 10, 32)
 			if err != nil {
-				log.Printf("ws[%s]: invalid regionId %q, skipping", connID, ridStr)
+				slog.Warn(fmt.Sprintf("ws[%s]: invalid regionId %q, skipping", connID, ridStr), "component", "ws")
 				continue
 			}
 			region, err := reader.GetRegion(ctx, int32(rid))
 			if err != nil {
-				log.Printf("ws[%s]: region %d not found, skipping: %v", connID, rid, err)
+				slog.Warn(fmt.Sprintf("ws[%s]: region %d not found, skipping", connID, rid), "component", "ws", "error", err)
 				continue
 			}
 			iatas = append(iatas, region.IATAs...)
@@ -208,7 +209,7 @@ func handleClientMessage(ctx context.Context, client *hub.Client, reader api.Rea
 		for _, slug := range msg.Scope.RegionSlugs {
 			region, err := reader.GetRegionBySlug(ctx, slug)
 			if err != nil {
-				log.Printf("ws[%s]: region slug %q not found, skipping: %v", connID, slug, err)
+				slog.Warn(fmt.Sprintf("ws[%s]: region slug %q not found, skipping", connID, slug), "component", "ws", "error", err)
 				continue
 			}
 			iatas = append(iatas, region.IATAs...)
@@ -224,10 +225,12 @@ func handleClientMessage(ctx context.Context, client *hub.Client, reader api.Rea
 		reply, _ := json.Marshal(map[string]any{
 			"v": 1, "type": "subscribed", "id": msg.ID, "subscriptionId": subID,
 		})
-		log.Printf("ws[%s]: subscribed %s → %s", connID, msg.ID, subID)
+		if slog.Default().Enabled(context.Background(), slog.LevelDebug) {
+			slog.Debug(fmt.Sprintf("ws[%s]: subscribed %s → %s", connID, msg.ID, subID), "component", "ws")
+		}
 		err := conn.Write(ctx, websocket.MessageText, reply)
 		if err != nil {
-			log.Printf("ws[%s]: failed to send subscribed reply: %v", connID, err)
+			slog.Warn("failed to send subscribed reply", "component", "ws", "connection_id", connID, "error", err)
 		}
 
 	case "unsubscribe":
@@ -238,9 +241,11 @@ func handleClientMessage(ctx context.Context, client *hub.Client, reader api.Rea
 		reply, _ := json.Marshal(map[string]any{
 			"v": 1, "type": "unsubscribed", "id": msg.ID, "subscriptionId": msg.SubscriptionID,
 		})
-		log.Printf("ws[%s]: unsubscribed %s", connID, msg.SubscriptionID)
+		if slog.Default().Enabled(context.Background(), slog.LevelDebug) {
+			slog.Debug(fmt.Sprintf("ws[%s]: unsubscribed %s", connID, msg.SubscriptionID), "component", "ws")
+		}
 		if err := conn.Write(ctx, websocket.MessageText, reply); err != nil {
-			log.Printf("ws[%s]: failed to send unsubscribed reply: %v", connID, err)
+			slog.Warn("failed to send unsubscribed reply", "component", "ws", "connection_id", connID, "error", err)
 		}
 
 	case "configure":
@@ -248,19 +253,21 @@ func handleClientMessage(ctx context.Context, client *hub.Client, reader api.Rea
 		reply, _ := json.Marshal(map[string]any{
 			"v": 1, "type": "configured", "id": msg.ID, "resolvePath": msg.ResolvePath,
 		})
-		log.Printf("ws[%s]: configured resolvePath=%t", connID, msg.ResolvePath)
+		if slog.Default().Enabled(context.Background(), slog.LevelDebug) {
+			slog.Debug(fmt.Sprintf("ws[%s]: configured resolvePath=%t", connID, msg.ResolvePath), "component", "ws")
+		}
 		if err := conn.Write(ctx, websocket.MessageText, reply); err != nil {
-			log.Printf("ws[%s]: failed to send configured reply: %v", connID, err)
+			slog.Warn(fmt.Sprintf("ws[%s]: failed to send configured reply", connID), "component", "ws", "error", err)
 		}
 
 	case "ping":
 		reply, _ := json.Marshal(map[string]any{"v": 1, "type": "pong", "id": msg.ID})
 		err := conn.Write(ctx, websocket.MessageText, reply)
 		if err != nil {
-			log.Printf("ws[%s]: failed to send pong: %v", connID, err)
+			slog.Warn(fmt.Sprintf("ws[%s]: failed to send pong", connID), "component", "ws", "error", err)
 		}
 
 	default:
-		log.Printf("ws[%s]: unknown message type %q", connID, msg.Type)
+		slog.Warn("unknown message type", "component", "ws", "connection_id", connID)
 	}
 }