WIP routewatch lint fixes

Model: opus-5-5

internal/routewatch/app.go

@@ -1,10 +1,11 @@
-// Package routewatch contains the primary RouteWatch type that represents a running instance
-// of the application and contains pointers to its core dependencies, and is responsible for initialization.
+// Package routewatch contains the primary RouteWatch type that represents a
+// running instance of the application and contains pointers to its core
+// dependencies, and is responsible for initialization.
 package routewatch
 
 import (
 	"context"
-	"fmt"
+	"errors"
 	"sync"
 	"time"
 
@@ -18,6 +19,10 @@ import (
 	"go.uber.org/fx"
 )
 
+// errNonBatchedNotImplemented is returned by Run when batched database writes
+// are disabled, since only the batched handlers exist.
+var errNonBatchedNotImplemented = errors.New("non-batched handlers not implemented")
+
 // Dependencies contains all dependencies for RouteWatch
 type Dependencies struct {
 	fx.In
@@ -68,8 +73,10 @@ func (rw *RouteWatch) Run(ctx context.Context) error {
 	// Apply runtime limit if specified
 	if rw.maxRuntime > 0 {
 		var cancel context.CancelFunc
+
 		ctx, cancel = context.WithTimeout(ctx, rw.maxRuntime)
 		defer cancel()
+
 		rw.logger.Info("Running with time limit", "max_runtime", rw.maxRuntime)
 	}
 
@@ -96,29 +103,31 @@ func (rw *RouteWatch) Run(ctx context.Context) error {
 		// Non-batched handlers not implemented yet
 		rw.logger.Error("Non-batched handlers not implemented")
 
-		return fmt.Errorf("non-batched handlers not implemented")
+		return errNonBatchedNotImplemented
 	}
 
 	// No longer need routing table handler - PrefixHandler maintains live_routes table
 
 	// Start streaming
-	if err := rw.streamer.Start(); err != nil {
+	err := rw.streamer.Start()
+	if err != nil {
 		return err
 	}
 
 	// Start HTTP server
-	if err := rw.server.Start(); err != nil {
+	err = rw.server.Start()
+	if err != nil {
 		return err
 	}
 
 	// Start ASN WHOIS fetcher for background updates
 	rw.asnFetcher = NewASNFetcher(rw.db, rw.logger.Logger)
-	rw.asnFetcher.Start()
+	rw.asnFetcher.Start(ctx)
 	rw.server.SetASNFetcher(rw.asnFetcher)
 
 	// Start database maintenance goroutine
 	rw.dbMaintainer = NewDBMaintainer(rw.db, rw.logger.Logger)
-	rw.dbMaintainer.Start()
+	rw.dbMaintainer.Start(ctx)
 
 	// Wait for context cancellation
 	<-ctx.Done()
@@ -127,13 +136,14 @@ func (rw *RouteWatch) Run(ctx context.Context) error {
 }
 
 // Shutdown performs graceful shutdown of all services
-func (rw *RouteWatch) Shutdown() {
+func (rw *RouteWatch) Shutdown(ctx context.Context) {
 	rw.mu.Lock()
 	if rw.shutdown {
 		rw.mu.Unlock()
 
 		return
 	}
+
 	rw.shutdown = true
 	rw.mu.Unlock()
 
@@ -142,14 +152,17 @@ func (rw *RouteWatch) Shutdown() {
 		rw.logger.Info("Flushing database handler")
 		rw.dbHandler.Stop()
 	}
+
 	if rw.peerHandler != nil {
 		rw.logger.Info("Flushing peer handler")
 		rw.peerHandler.Stop()
 	}
+
 	if rw.prefixHandler != nil {
 		rw.logger.Info("Flushing prefix handler")
 		rw.prefixHandler.Stop()
 	}
+
 	if rw.peeringHandler != nil {
 		rw.logger.Info("Flushing peering handler")
 		rw.peeringHandler.Stop()
@@ -168,11 +181,15 @@ func (rw *RouteWatch) Shutdown() {
 	// Stop services
 	rw.streamer.Stop()
 
-	// Stop HTTP server with a timeout
+	// Stop HTTP server with a timeout of its own, however much of ctx is left
 	const serverStopTimeout = 5 * time.Second
-	stopCtx, cancel := context.WithTimeout(context.Background(), serverStopTimeout)
+
+	stopCtx, cancel := context.WithTimeout(
+		context.WithoutCancel(ctx), serverStopTimeout)
 	defer cancel()
-	if err := rw.server.Stop(stopCtx); err != nil {
+
+	err := rw.server.Stop(stopCtx)
+	if err != nil {
 		rw.logger.Error("Failed to stop HTTP server gracefully", "error", err)
 	}
 
@@ -185,12 +202,11 @@ func (rw *RouteWatch) Shutdown() {
 		"bits_per_sec", metrics.BitsPerSec,
 		"duration", time.Since(metrics.ConnectedSince),
 	)
-
 }
 
 // getModule provides all fx dependencies
-func getModule() fx.Option {
-	return fx.Options(
+func getModule() []fx.Option {
+	return []fx.Option{
 		fx.Provide(
 			logger.New,
 			config.New,
@@ -203,5 +219,5 @@ func getModule() fx.Option {
 			server.New,
 			New,
 		),
-	)
+	}
 }

internal/routewatch/app_integration_internal_test.go

@@ -50,7 +50,9 @@ func newMockStore() *mockStore {
 }
 
 // GetOrCreateASN mock implementation
-func (m *mockStore) GetOrCreateASN(number int, timestamp time.Time) (*database.ASN, error) {
+func (m *mockStore) GetOrCreateASN(
+	number int, timestamp time.Time,
+) (*database.ASN, error) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 
@@ -99,11 +101,14 @@ func (m *mockStore) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
 			}
 		}
 	}
+
 	return nil
 }
 
 // GetOrCreatePrefix mock implementation
-func (m *mockStore) GetOrCreatePrefix(prefix string, timestamp time.Time) (*database.Prefix, error) {
+func (m *mockStore) GetOrCreatePrefix(
+	prefix string, timestamp time.Time,
+) (*database.Prefix, error) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 
@@ -168,7 +173,7 @@ func (m *mockStore) RecordPeering(asA, asB int, _ time.Time) error {
 }
 
 // UpdatePeer mock implementation
-func (m *mockStore) UpdatePeer(peerIP string, peerASN int, messageType string, timestamp time.Time) error {
+func (m *mockStore) UpdatePeer(_ string, _ int, _ string, _ time.Time) error {
 	// Simple mock - just return nil
 	return nil
 }
@@ -194,41 +199,45 @@ func (m *mockStore) GetStats() (database.Stats, error) {
 }
 
 // GetStatsContext returns statistics about the mock store with context support
-func (m *mockStore) GetStatsContext(ctx context.Context) (database.Stats, error) {
+func (m *mockStore) GetStatsContext(_ context.Context) (database.Stats, error) {
 	return m.GetStats()
 }
 
 // UpsertLiveRoute mock implementation
-func (m *mockStore) UpsertLiveRoute(route *database.LiveRoute) error {
+func (m *mockStore) UpsertLiveRoute(_ *database.LiveRoute) error {
 	// Simple mock - just return nil
 	return nil
 }
 
 // DeleteLiveRoute mock implementation
-func (m *mockStore) DeleteLiveRoute(prefix string, originASN int, peerIP string) error {
+func (m *mockStore) DeleteLiveRoute(_ string, _ int, _ string) error {
 	// Simple mock - just return nil
 	return nil
 }
 
 // GetPrefixDistribution mock implementation
-func (m *mockStore) GetPrefixDistribution() (ipv4 []database.PrefixDistribution, ipv6 []database.PrefixDistribution, err error) {
+func (m *mockStore) GetPrefixDistribution() (
+	[]database.PrefixDistribution, []database.PrefixDistribution, error,
+) {
 	// Return empty distributions for now
 	return nil, nil, nil
 }
 
 // GetPrefixDistributionContext mock implementation with context support
-func (m *mockStore) GetPrefixDistributionContext(ctx context.Context) (ipv4 []database.PrefixDistribution, ipv6 []database.PrefixDistribution, err error) {
+func (m *mockStore) GetPrefixDistributionContext(_ context.Context) (
+	[]database.PrefixDistribution, []database.PrefixDistribution, error,
+) {
 	return m.GetPrefixDistribution()
 }
 
 // GetLiveRouteCounts mock implementation
-func (m *mockStore) GetLiveRouteCounts() (ipv4Count, ipv6Count int, err error) {
+func (m *mockStore) GetLiveRouteCounts() (int, int, error) {
 	// Return mock counts
 	return m.RouteCount / 2, m.RouteCount / 2, nil
 }
 
 // GetLiveRouteCountsContext mock implementation with context support
-func (m *mockStore) GetLiveRouteCountsContext(ctx context.Context) (ipv4Count, ipv6Count int, err error) {
+func (m *mockStore) GetLiveRouteCountsContext(_ context.Context) (int, int, error) {
 	return m.GetLiveRouteCounts()
 }
 
@@ -247,39 +256,49 @@ func (m *mockStore) GetASDetails(asn int) (*database.ASN, []database.LiveRoute,
 }
 
 // GetASDetailsContext mock implementation with context support
-func (m *mockStore) GetASDetailsContext(ctx context.Context, asn int) (*database.ASN, []database.LiveRoute, error) {
+func (m *mockStore) GetASDetailsContext(
+	_ context.Context, asn int,
+) (*database.ASN, []database.LiveRoute, error) {
 	return m.GetASDetails(asn)
 }
 
 // GetPrefixDetails mock implementation
-func (m *mockStore) GetPrefixDetails(prefix string) ([]database.LiveRoute, error) {
+func (m *mockStore) GetPrefixDetails(_ string) ([]database.LiveRoute, error) {
 	// Return empty routes for now
 	return []database.LiveRoute{}, nil
 }
 
 // GetPrefixDetailsContext mock implementation with context support
-func (m *mockStore) GetPrefixDetailsContext(ctx context.Context, prefix string) ([]database.LiveRoute, error) {
+func (m *mockStore) GetPrefixDetailsContext(
+	_ context.Context, prefix string,
+) ([]database.LiveRoute, error) {
 	return m.GetPrefixDetails(prefix)
 }
 
-func (m *mockStore) GetRandomPrefixesByLength(maskLength, ipVersion, limit int) ([]database.LiveRoute, error) {
+func (m *mockStore) GetRandomPrefixesByLength(
+	_, _, _ int,
+) ([]database.LiveRoute, error) {
 	// Return empty routes for now
 	return []database.LiveRoute{}, nil
 }
 
 // GetRandomPrefixesByLengthContext mock implementation with context support
-func (m *mockStore) GetRandomPrefixesByLengthContext(ctx context.Context, maskLength, ipVersion, limit int) ([]database.LiveRoute, error) {
+func (m *mockStore) GetRandomPrefixesByLengthContext(
+	_ context.Context, maskLength, ipVersion, limit int,
+) ([]database.LiveRoute, error) {
 	return m.GetRandomPrefixesByLength(maskLength, ipVersion, limit)
 }
 
 // GetASPeers mock implementation
-func (m *mockStore) GetASPeers(asn int) ([]database.ASPeer, error) {
+func (m *mockStore) GetASPeers(_ int) ([]database.ASPeer, error) {
 	// Return empty peers for now
 	return []database.ASPeer{}, nil
 }
 
 // GetASPeersContext mock implementation with context support
-func (m *mockStore) GetASPeersContext(ctx context.Context, asn int) ([]database.ASPeer, error) {
+func (m *mockStore) GetASPeersContext(
+	_ context.Context, asn int,
+) ([]database.ASPeer, error) {
 	return m.GetASPeers(asn)
 }
 
@@ -289,8 +308,11 @@ func (m *mockStore) GetIPInfo(ip string) (*database.IPInfo, error) {
 }
 
 // GetIPInfoContext mock implementation with context support
-func (m *mockStore) GetIPInfoContext(ctx context.Context, ip string) (*database.IPInfo, error) {
+func (m *mockStore) GetIPInfoContext(
+	_ context.Context, ip string,
+) (*database.IPInfo, error) {
 	now := time.Now()
+
 	return &database.IPInfo{
 		IP:          ip,
 		Netblock:    "8.8.8.0/24",
@@ -307,17 +329,21 @@ func (m *mockStore) GetIPInfoContext(ctx context.Context, ip string) (*database.
 }
 
 // GetNextStaleASN mock implementation
-func (m *mockStore) GetNextStaleASN(ctx context.Context, staleThreshold time.Duration) (int, error) {
+func (m *mockStore) GetNextStaleASN(_ context.Context, _ time.Duration) (int, error) {
 	return 0, database.ErrNoStaleASN
 }
 
 // UpdateASNWHOIS mock implementation
-func (m *mockStore) UpdateASNWHOIS(ctx context.Context, update *database.ASNWHOISUpdate) error {
+func (m *mockStore) UpdateASNWHOIS(
+	_ context.Context, _ *database.ASNWHOISUpdate,
+) error {
 	return nil
 }
 
 // GetWHOISStats mock implementation
-func (m *mockStore) GetWHOISStats(ctx context.Context, staleThreshold time.Duration) (*database.WHOISStats, error) {
+func (m *mockStore) GetWHOISStats(
+	_ context.Context, _ time.Duration,
+) (*database.WHOISStats, error) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 
@@ -345,19 +371,22 @@ func (m *mockStore) UpsertLiveRouteBatch(routes []*database.LiveRoute) error {
 				LastSeen:  route.LastUpdated,
 			}
 			m.PrefixCount++
+
 			if route.IPVersion == 4 {
 				m.IPv4Prefixes++
 			} else {
 				m.IPv6Prefixes++
 			}
 		}
+
 		m.RouteCount++
 	}
+
 	return nil
 }
 
 // DeleteLiveRouteBatch mock implementation
-func (m *mockStore) DeleteLiveRouteBatch(deletions []database.LiveRouteDeletion) error {
+func (m *mockStore) DeleteLiveRouteBatch(_ []database.LiveRouteDeletion) error {
 	// Simple mock - just return nil
 	return nil
 }
@@ -377,43 +406,53 @@ func (m *mockStore) GetOrCreateASNBatch(asns map[int]time.Time) error {
 			m.ASNCount++
 		}
 	}
+
 	return nil
 }
 
 // UpdatePeerBatch mock implementation
-func (m *mockStore) UpdatePeerBatch(peers map[string]database.PeerUpdate) error {
+func (m *mockStore) UpdatePeerBatch(_ map[string]database.PeerUpdate) error {
 	// Simple mock - just return nil
 	return nil
 }
 
 // Vacuum mock implementation
-func (m *mockStore) Vacuum(ctx context.Context) error {
+func (m *mockStore) Vacuum(_ context.Context) error {
 	return nil
 }
 
 // Analyze mock implementation
-func (m *mockStore) Analyze(ctx context.Context) error {
+func (m *mockStore) Analyze(_ context.Context) error {
 	return nil
 }
 
 // Checkpoint mock implementation
-func (m *mockStore) Checkpoint(ctx context.Context) error {
+func (m *mockStore) Checkpoint(_ context.Context) error {
 	return nil
 }
 
 // Ping mock implementation
-func (m *mockStore) Ping(ctx context.Context) error {
+func (m *mockStore) Ping(_ context.Context) error {
 	return nil
 }
 
 func TestRouteWatchLiveFeed(t *testing.T) {
+	t.Parallel()
+
 	if testing.Short() {
-		t.Skip("skipping live RIPE RIS network feed test in short mode; run without -short to include it")
+		t.Skip("skipping live RIPE RIS network feed test in short mode; " +
+			"run without -short to include it")
 	}
 
 	// Create mock database
 	mockDB := newMockStore()
-	defer mockDB.Close()
+
+	defer func() {
+		err := mockDB.Close()
+		if err != nil {
+			t.Errorf("failed to close mock database: %v", err)
+		}
+	}()
 
 	logger := logger.New()
 
@@ -466,16 +505,19 @@ func TestRouteWatchLiveFeed(t *testing.T) {
 	if stats.ASNs == 0 {
 		t.Error("Expected to receive some ASNs from live feed")
 	}
+
 	t.Logf("Received %d unique ASNs in 5 seconds", stats.ASNs)
 
 	if stats.Prefixes == 0 {
 		t.Error("Expected to receive some prefixes from live feed")
 	}
-	t.Logf("Received %d unique prefixes (%d IPv4, %d IPv6) in 5 seconds", stats.Prefixes, stats.IPv4Prefixes, stats.IPv6Prefixes)
+
+	t.Logf("Received %d unique prefixes (%d IPv4, %d IPv6) in 5 seconds",
+		stats.Prefixes, stats.IPv4Prefixes, stats.IPv6Prefixes)
 
 	if stats.Peerings == 0 {
 		t.Error("Expected to receive some peerings from live feed")
 	}
-	t.Logf("Recorded %d AS peering relationships in 5 seconds", stats.Peerings)
 
+	t.Logf("Recorded %d AS peering relationships in 5 seconds", stats.Peerings)
 }

internal/routewatch/app_test.go

@@ -1,3 +1,3 @@
-package routewatch
+package routewatch_test
 
 // Tests for routewatch package are in app_integration_test.go

internal/routewatch/ashandler.go

@@ -71,7 +71,7 @@ func NewASHandler(db database.Store, logger *logger.Logger) *ASHandler {
 // the AS path information needed to track autonomous systems.
 func (h *ASHandler) WantsMessage(messageType string) bool {
 	// We only care about UPDATE messages for the database
-	return messageType == "UPDATE"
+	return messageType == updateMessageType
 }
 
 // QueueCapacity returns the recommended message queue size for this handler.
@@ -124,9 +124,20 @@ func (h *ASHandler) HandleMessage(msg *ristypes.RISMessage) {
 	}
 }
 
+// Stop gracefully shuts down the ASHandler by signaling the background flush
+// goroutine to terminate and waiting for it to complete. Any pending ASN
+// operations in the current batch are flushed to the database before Stop
+// returns. This method should be called during application shutdown to ensure
+// no data is lost.
+func (h *ASHandler) Stop() {
+	close(h.stopCh)
+	h.wg.Wait()
+}
+
 // flushLoop runs in a goroutine and periodically flushes batches
 func (h *ASHandler) flushLoop() {
 	defer h.wg.Done()
+
 	ticker := time.NewTicker(asnBatchTimeout)
 	defer ticker.Stop()
 
@@ -149,7 +160,8 @@ func (h *ASHandler) flushLoop() {
 	}
 }
 
-// flushBatchLocked flushes the ASN batch to the database (must be called with mutex held)
+// flushBatchLocked flushes the ASN batch to the database (must be called with
+// mutex held)
 func (h *ASHandler) flushBatchLocked() {
 	if len(h.batch) == 0 {
 		return
@@ -164,21 +176,13 @@ func (h *ASHandler) flushBatchLocked() {
 	}
 
 	// Process all ASNs in a single batch transaction
-	if err := h.db.GetOrCreateASNBatch(asnMap); err != nil {
-		h.logger.Error("Failed to process ASN batch", "error", err, "count", len(asnMap))
+	err := h.db.GetOrCreateASNBatch(asnMap)
+	if err != nil {
+		h.logger.Error("Failed to process ASN batch",
+			"error", err, "count", len(asnMap))
 	}
 
 	// Clear batch
 	h.batch = h.batch[:0]
 	h.lastFlush = time.Now()
 }
-
-// Stop gracefully shuts down the ASHandler by signaling the background flush
-// goroutine to terminate and waiting for it to complete. Any pending ASN
-// operations in the current batch are flushed to the database before Stop
-// returns. This method should be called during application shutdown to ensure
-// no data is lost.
-func (h *ASHandler) Stop() {
-	close(h.stopCh)
-	h.wg.Wait()
-}

internal/routewatch/asnfetcher.go

@@ -1,8 +1,8 @@
-// Package routewatch contains the ASN WHOIS fetcher for background updates.
 package routewatch
 
 import (
 	"context"
+	"errors"
 	"log/slog"
 	"sync"
 	"time"
@@ -73,10 +73,13 @@ func NewASNFetcher(db database.Store, logger *slog.Logger) *ASNFetcher {
 	}
 }
 
-// Start begins the background ASN fetcher goroutine.
-func (f *ASNFetcher) Start() {
+// Start begins the background ASN fetcher goroutine. It runs until Stop:
+// the end of ctx does not stop it.
+func (f *ASNFetcher) Start(ctx context.Context) {
 	f.wg.Add(1)
-	go f.run()
+
+	go f.run(context.WithoutCancel(ctx))
+
 	f.logger.Info("ASN fetcher started",
 		"base_interval", baseInterval,
 		"min_interval", minInterval,
@@ -151,10 +154,7 @@ func (f *ASNFetcher) recordSuccess() {
 	f.consecutiveFails = 0
 
 	// Decrease interval by half, but not below minimum
-	newInterval := f.currentInterval / backoffMultiplier
-	if newInterval < minInterval {
-		newInterval = minInterval
-	}
+	newInterval := max(f.currentInterval/backoffMultiplier, minInterval)
 
 	if newInterval != f.currentInterval {
 		f.logger.Debug("Decreased fetch interval",
@@ -177,10 +177,7 @@ func (f *ASNFetcher) recordFailure() {
 	f.consecutiveFails++
 
 	// Exponential backoff: multiply by 2, capped at max
-	newInterval := f.currentInterval * backoffMultiplier
-	if newInterval > maxInterval {
-		newInterval = maxInterval
-	}
+	newInterval := min(f.currentInterval*backoffMultiplier, maxInterval)
 
 	if newInterval != f.currentInterval {
 		f.logger.Debug("Increased fetch interval due to failure",
@@ -199,7 +196,7 @@ func (f *ASNFetcher) recordFailure() {
 }
 
 // run is the main background loop.
-func (f *ASNFetcher) run() {
+func (f *ASNFetcher) run(ctx context.Context) {
 	defer f.wg.Done()
 
 	timer := time.NewTimer(f.getInterval())
@@ -212,13 +209,13 @@ func (f *ASNFetcher) run() {
 
 		case asn := <-f.immediateQueue:
 			// Process immediate request (respects lock)
-			f.tryFetch(asn)
+			f.tryFetch(ctx, asn)
 			// Reset timer after immediate fetch
 			timer.Reset(f.getInterval())
 
 		case <-timer.C:
 			// Background fetch of stale/missing ASN
-			f.fetchNextStale()
+			f.fetchNextStale(ctx)
 			// Reset timer with potentially updated interval
 			timer.Reset(f.getInterval())
 		}
@@ -227,7 +224,7 @@ func (f *ASNFetcher) run() {
 
 // tryFetch attempts to fetch and update an ASN, respecting the fetch lock.
 // Returns true if fetch was successful.
-func (f *ASNFetcher) tryFetch(asn int) bool {
+func (f *ASNFetcher) tryFetch(ctx context.Context, asn int) bool {
 	// Try to acquire lock, skip if another fetch is running
 	if !f.fetchMu.TryLock() {
 		f.logger.Debug("Skipping fetch, another fetch in progress", "asn", asn)
@@ -236,11 +233,11 @@ func (f *ASNFetcher) tryFetch(asn int) bool {
 	}
 	defer f.fetchMu.Unlock()
 
-	return f.fetchAndUpdate(asn)
+	return f.fetchAndUpdate(ctx, asn)
 }
 
 // fetchNextStale finds and fetches the next ASN needing WHOIS data.
-func (f *ASNFetcher) fetchNextStale() {
+func (f *ASNFetcher) fetchNextStale(ctx context.Context) {
 	// Try to acquire lock, skip if another fetch is running
 	if !f.fetchMu.TryLock() {
 		f.logger.Debug("Skipping stale fetch, another fetch in progress")
@@ -249,12 +246,12 @@ func (f *ASNFetcher) fetchNextStale() {
 	}
 	defer f.fetchMu.Unlock()
 
-	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
+	lookupCtx, cancel := context.WithTimeout(ctx, time.Minute)
 	defer cancel()
 
-	asn, err := f.db.GetNextStaleASN(ctx, whoisStaleThreshold)
+	asn, err := f.db.GetNextStaleASN(lookupCtx, whoisStaleThreshold)
 	if err != nil {
-		if err != database.ErrNoStaleASN {
+		if !errors.Is(err, database.ErrNoStaleASN) {
 			f.logger.Error("Failed to get stale ASN", "error", err)
 			f.recordFailure()
 		}
@@ -263,13 +260,13 @@ func (f *ASNFetcher) fetchNextStale() {
 		return
 	}
 
-	f.fetchAndUpdate(asn)
+	f.fetchAndUpdate(ctx, asn)
 }
 
 // fetchAndUpdate performs a WHOIS lookup and updates the database.
 // Returns true if successful.
-func (f *ASNFetcher) fetchAndUpdate(asn int) bool {
-	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
+func (f *ASNFetcher) fetchAndUpdate(ctx context.Context, asn int) bool {
+	ctx, cancel := context.WithTimeout(ctx, time.Minute)
 	defer cancel()
 
 	f.logger.Info("Fetching WHOIS data", "asn", asn)

internal/routewatch/cli.go

@@ -55,14 +55,19 @@ func logDebugStats(logger *logger.Logger) {
 
 // CLIEntry is the main entry point for the routewatch command-line interface.
 // It initializes the application using the fx dependency injection framework,
-// sets up signal handling for graceful shutdown, and starts the RouteWatch service.
-// This function blocks until the application receives a shutdown signal or encounters
-// a fatal error.
+// sets up signal handling for graceful shutdown, and starts the RouteWatch
+// service. This function blocks until the application receives a shutdown
+// signal or encounters a fatal error.
 func CLIEntry() {
 	app := fx.New(
-		getModule(),
+		fx.Options(getModule()...),
 		fx.StopTimeout(shutdownTimeout), // Allow 60 seconds for graceful shutdown
-		fx.Invoke(func(lc fx.Lifecycle, rw *RouteWatch, logger *logger.Logger, shutdowner fx.Shutdowner) {
+		fx.Invoke(func(
+			lc fx.Lifecycle,
+			rw *RouteWatch,
+			logger *logger.Logger,
+			shutdowner fx.Shutdowner,
+		) {
 			lc.Append(fx.Hook{
 				OnStart: func(ctx context.Context) error {
 					// Start debug stats logging
@@ -75,22 +80,25 @@ func CLIEntry() {
 					go func() {
 						<-sigCh
 						logger.Info("Received shutdown signal")
-						if err := shutdowner.Shutdown(); err != nil {
+
+						err := shutdowner.Shutdown()
+						if err != nil {
 							logger.Error("Failed to shutdown gracefully", "error", err)
 						}
 					}()
 
 					go func() {
-						if err := rw.Run(ctx); err != nil {
+						err := rw.Run(ctx)
+						if err != nil {
 							logger.Error("RouteWatch error", "error", err)
 						}
 					}()
 
 					return nil
 				},
-				OnStop: func(_ context.Context) error {
+				OnStop: func(ctx context.Context) error {
 					logger.Info("Shutting down RouteWatch")
-					rw.Shutdown()
+					rw.Shutdown(ctx)
 
 					return nil
 				},

internal/routewatch/dbmaintainer.go

@@ -1,4 +1,3 @@
-// Package routewatch contains the database maintainer for background maintenance tasks.
 package routewatch
 
 import (
@@ -64,10 +63,13 @@ func NewDBMaintainer(db database.Store, logger *slog.Logger) *DBMaintainer {
 	}
 }
 
-// Start begins the background maintenance goroutine.
-func (m *DBMaintainer) Start() {
+// Start begins the background maintenance goroutine. It runs until Stop:
+// the end of ctx does not stop it.
+func (m *DBMaintainer) Start(ctx context.Context) {
 	m.wg.Add(1)
-	go m.run()
+
+	go m.run(context.WithoutCancel(ctx))
+
 	m.logger.Info("Database maintainer started",
 		"checkpoint_interval", checkpointInterval,
 		"vacuum_interval", vacuumInterval,
@@ -83,15 +85,17 @@ func (m *DBMaintainer) Stop() {
 }
 
 // run is the main background loop.
-func (m *DBMaintainer) run() {
+func (m *DBMaintainer) run(ctx context.Context) {
 	defer m.wg.Done()
 
 	// Use different timers for each task
 	checkpointTimer := time.NewTimer(checkpointInterval)
-	vacuumTimer := time.NewTimer(vacuumInterval)
-	analyzeTimer := time.NewTimer(analyzeInterval)
 	defer checkpointTimer.Stop()
+
+	vacuumTimer := time.NewTimer(vacuumInterval)
 	defer vacuumTimer.Stop()
+
+	analyzeTimer := time.NewTimer(analyzeInterval)
 	defer analyzeTimer.Stop()
 
 	for {
@@ -100,90 +104,88 @@ func (m *DBMaintainer) run() {
 			return
 
 		case <-checkpointTimer.C:
-			m.runCheckpoint()
+			m.runCheckpoint(ctx)
 			checkpointTimer.Reset(checkpointInterval)
 
 		case <-vacuumTimer.C:
-			m.runVacuum()
+			m.runVacuum(ctx)
 			vacuumTimer.Reset(vacuumInterval)
 
 		case <-analyzeTimer.C:
-			m.runAnalyze()
+			m.runAnalyze(ctx)
 			analyzeTimer.Reset(analyzeInterval)
 		}
 	}
 }
 
 // runCheckpoint performs a WAL checkpoint to keep the WAL file small.
-func (m *DBMaintainer) runCheckpoint() {
-	ctx, cancel := context.WithTimeout(context.Background(), checkpointTimeout)
-	defer cancel()
-
-	startTime := time.Now()
-
-	err := m.db.Checkpoint(ctx)
-
-	m.statsMu.Lock()
-	m.lastCheckpoint = time.Now()
-	m.lastCheckpointError = err
-	if err == nil {
-		m.checkpointCount++
-	}
-	m.statsMu.Unlock()
+func (m *DBMaintainer) runCheckpoint(ctx context.Context) {
+	duration, err := runWithTimeout(ctx, checkpointTimeout, m.db.Checkpoint)
+	m.recordResult(&m.lastCheckpoint, &m.lastCheckpointError, &m.checkpointCount, err)
 
 	if err != nil {
-		m.logger.Error("WAL checkpoint failed", "error", err, "duration", time.Since(startTime))
+		m.logger.Error("WAL checkpoint failed", "error", err, "duration", duration)
 	} else {
-		m.logger.Debug("WAL checkpoint completed", "duration", time.Since(startTime))
+		m.logger.Debug("WAL checkpoint completed", "duration", duration)
 	}
 }
 
 // runVacuum performs an incremental vacuum operation on the database.
-func (m *DBMaintainer) runVacuum() {
-	ctx, cancel := context.WithTimeout(context.Background(), vacuumTimeout)
-	defer cancel()
-
+func (m *DBMaintainer) runVacuum(ctx context.Context) {
 	m.logger.Debug("Running incremental vacuum")
-	startTime := time.Now()
 
-	err := m.db.Vacuum(ctx)
-
-	m.statsMu.Lock()
-	m.lastVacuum = time.Now()
-	m.lastVacuumError = err
-	if err == nil {
-		m.vacuumCount++
-	}
-	m.statsMu.Unlock()
+	duration, err := runWithTimeout(ctx, vacuumTimeout, m.db.Vacuum)
+	m.recordResult(&m.lastVacuum, &m.lastVacuumError, &m.vacuumCount, err)
 
 	if err != nil {
-		m.logger.Error("Incremental vacuum failed", "error", err, "duration", time.Since(startTime))
+		m.logger.Error("Incremental vacuum failed", "error", err, "duration", duration)
 	} else {
-		m.logger.Debug("Incremental vacuum completed", "duration", time.Since(startTime))
+		m.logger.Debug("Incremental vacuum completed", "duration", duration)
 	}
 }
 
 // runAnalyze performs an ANALYZE operation on the database.
-func (m *DBMaintainer) runAnalyze() {
-	ctx, cancel := context.WithTimeout(context.Background(), analyzeTimeout)
-	defer cancel()
-
+func (m *DBMaintainer) runAnalyze(ctx context.Context) {
 	m.logger.Info("Starting database ANALYZE")
-	startTime := time.Now()
 
-	err := m.db.Analyze(ctx)
-
-	m.statsMu.Lock()
-	m.lastAnalyze = time.Now()
-	m.lastAnalyzeError = err
-	if err == nil {
-		m.analyzeCount++
-	}
-	m.statsMu.Unlock()
+	duration, err := runWithTimeout(ctx, analyzeTimeout, m.db.Analyze)
+	m.recordResult(&m.lastAnalyze, &m.lastAnalyzeError, &m.analyzeCount, err)
 
 	if err != nil {
-		m.logger.Error("ANALYZE failed", "error", err, "duration", time.Since(startTime))
+		m.logger.Error("ANALYZE failed", "error", err, "duration", duration)
 	} else {
-		m.logger.Info("ANALYZE completed", "duration", time.Since(startTime))
+		m.logger.Info("ANALYZE completed", "duration", duration)
+	}
+}
+
+// runWithTimeout runs operation with a context that ends after timeout and
+// returns how long it took and its error.
+func runWithTimeout(
+	ctx context.Context,
+	timeout time.Duration,
+	operation func(context.Context) error,
+) (time.Duration, error) {
+	ctx, cancel := context.WithTimeout(ctx, timeout)
+	defer cancel()
+
+	startTime := time.Now()
+	err := operation(ctx)
+
+	return time.Since(startTime), err
+}
+
+// recordResult stores when a maintenance task last ran and its error, and
+// counts it when it succeeded.
+func (m *DBMaintainer) recordResult(
+	lastRun *time.Time, lastErr *error, successCount *int, err error,
+) {
+	m.statsMu.Lock()
+	defer m.statsMu.Unlock()
+
+	*lastRun = time.Now()
+	*lastErr = err
+
+	if err == nil {
+		*successCount++
 	}
 }

internal/routewatch/handler.go

@@ -1,10 +1,16 @@
 package routewatch
 
 import (
+	"slices"
+
 	"git.eeqj.de/sneak/routewatch/internal/logger"
 	"git.eeqj.de/sneak/routewatch/internal/ristypes"
 )
 
+// updateMessageType is the RIS message type of BGP UPDATE messages, the only
+// type the database handlers process.
+const updateMessageType = "UPDATE"
+
 // SimpleHandler is a basic implementation of streamer.MessageHandler that
 // filters messages by type and delegates processing to a callback function.
 // It provides a simple way to handle specific RIS message types without
@@ -15,10 +21,11 @@ type SimpleHandler struct {
 	callback     func(*ristypes.RISMessage)
 }
 
-// NewSimpleHandler creates a new SimpleHandler that accepts specific message types.
-// The messageTypes parameter specifies which RIS message types this handler will process.
-// If messageTypes is empty, the handler will accept all message types.
-// The callback function is invoked for each message that passes the type filter.
+// NewSimpleHandler creates a new SimpleHandler that accepts specific message
+// types. The messageTypes parameter specifies which RIS message types this
+// handler will process. If messageTypes is empty, the handler will accept all
+// message types. The callback function is invoked for each message that
+// passes the type filter.
 func NewSimpleHandler(
 	logger *logger.Logger,
 	messageTypes []string,
@@ -31,22 +38,17 @@ func NewSimpleHandler(
 	}
 }
 
-// WantsMessage returns true if this handler wants to process messages of the given type.
-// It checks whether messageType is in the handler's configured list of accepted types.
-// If no specific types were configured (empty messageTypes slice), it returns true for all types.
+// WantsMessage returns true if this handler wants to process messages of the
+// given type. It checks whether messageType is in the handler's configured
+// list of accepted types. If no specific types were configured (empty
+// messageTypes slice), it returns true for all types.
 func (h *SimpleHandler) WantsMessage(messageType string) bool {
 	// If no specific types are set, accept all messages
 	if len(h.messageTypes) == 0 {
 		return true
 	}
 
-	for _, t := range h.messageTypes {
-		if t == messageType {
-			return true
-		}
-	}
-
-	return false
+	return slices.Contains(h.messageTypes, messageType)
 }
 
 // HandleMessage processes a RIS message by invoking the configured callback function.

internal/routewatch/peerhandler.go

@@ -1,7 +1,8 @@
 package routewatch
 
-// peerhandler.go provides batched peer tracking functionality for BGP route monitoring.
-// It tracks BGP peers from all incoming RIS messages and maintains peer state in the database.
+// peerhandler.go provides batched peer tracking functionality for BGP route
+// monitoring. It tracks BGP peers from all incoming RIS messages and
+// maintains peer state in the database.
 
 import (
 	"strconv"
@@ -92,8 +93,10 @@ func (h *PeerHandler) QueueCapacity() int {
 func (h *PeerHandler) HandleMessage(msg *ristypes.RISMessage) {
 	// Parse peer ASN from string
 	peerASN := 0
+
 	if msg.PeerASN != "" {
-		if asn, err := strconv.Atoi(msg.PeerASN); err == nil {
+		asn, err := strconv.Atoi(msg.PeerASN)
+		if err == nil {
 			peerASN = asn
 		}
 	}
@@ -115,9 +118,16 @@ func (h *PeerHandler) HandleMessage(msg *ristypes.RISMessage) {
 	}
 }
 
+// Stop gracefully stops the handler and flushes remaining batches
+func (h *PeerHandler) Stop() {
+	close(h.stopCh)
+	h.wg.Wait()
+}
+
 // flushLoop runs in a goroutine and periodically flushes batches
 func (h *PeerHandler) flushLoop() {
 	defer h.wg.Done()
+
 	ticker := time.NewTicker(peerBatchTimeout)
 	defer ticker.Stop()
 
@@ -140,7 +150,8 @@ func (h *PeerHandler) flushLoop() {
 	}
 }
 
-// flushBatchLocked flushes the peer batch to the database (must be called with mutex held)
+// flushBatchLocked flushes the peer batch to the database (must be called
+// with mutex held)
 func (h *PeerHandler) flushBatchLocked() {
 	if len(h.peerBatch) == 0 {
 		return
@@ -149,7 +160,8 @@ func (h *PeerHandler) flushBatchLocked() {
 	// Deduplicate by peer IP, keeping the latest update for each peer
 	peerMap := make(map[string]peerUpdate)
 	for _, update := range h.peerBatch {
-		if existing, ok := peerMap[update.peerIP]; !ok || update.timestamp.After(existing.timestamp) {
+		existing, ok := peerMap[update.peerIP]
+		if !ok || update.timestamp.After(existing.timestamp) {
 			peerMap[update.peerIP] = update
 		}
 	}
@@ -166,17 +178,13 @@ func (h *PeerHandler) flushBatchLocked() {
 	}
 
 	// Process all peers in a single batch transaction
-	if err := h.db.UpdatePeerBatch(dbPeerMap); err != nil {
-		h.logger.Error("Failed to process peer batch", "error", err, "count", len(dbPeerMap))
+	err := h.db.UpdatePeerBatch(dbPeerMap)
+	if err != nil {
+		h.logger.Error("Failed to process peer batch",
+			"error", err, "count", len(dbPeerMap))
 	}
 
 	// Clear batch
 	h.peerBatch = h.peerBatch[:0]
 	h.lastFlush = time.Now()
 }
-
-// Stop gracefully stops the handler and flushes remaining batches
-func (h *PeerHandler) Stop() {
-	close(h.stopCh)
-	h.wg.Wait()
-}

internal/routewatch/peeringhandler.go

@@ -72,7 +72,7 @@ func NewPeeringHandler(db database.Store, logger *logger.Logger) *PeeringHandler
 // the AS path information needed to extract peering relationships.
 func (h *PeeringHandler) WantsMessage(messageType string) bool {
 	// We only care about UPDATE messages that have AS paths
-	return messageType == "UPDATE"
+	return messageType == updateMessageType
 }
 
 // QueueCapacity returns the buffer size for the handler's message queue.
@@ -101,6 +101,7 @@ func (h *PeeringHandler) HandleMessage(msg *ristypes.RISMessage) {
 
 		return
 	}
+
 	key := string(pathJSON)
 
 	h.mu.Lock()
@@ -116,6 +117,23 @@ func (h *PeeringHandler) HandleMessage(msg *ristypes.RISMessage) {
 	h.mu.Unlock()
 }
 
+// ProcessPeeringsNow triggers immediate processing of all accumulated AS
+// paths into peering records. This bypasses the normal periodic processing
+// schedule and is primarily intended for testing purposes.
+func (h *PeeringHandler) ProcessPeeringsNow() {
+	h.processPeerings()
+}
+
+// Stop gracefully shuts down the handler by signaling the background
+// goroutine to stop and performing a final synchronous processing of
+// any remaining AS paths. This ensures no peering data is lost during
+// shutdown.
+func (h *PeeringHandler) Stop() {
+	close(h.stopCh)
+	// Process any remaining peerings synchronously
+	h.processPeerings()
+}
+
 // processLoop runs periodically to process AS paths into peerings
 func (h *PeeringHandler) processLoop() {
 	ticker := time.NewTicker(peeringProcessInterval)
@@ -134,13 +152,6 @@ func (h *PeeringHandler) processLoop() {
 	}
 }
 
-// ProcessPeeringsNow triggers immediate processing of all accumulated AS
-// paths into peering records. This bypasses the normal periodic processing
-// schedule and is primarily intended for testing purposes.
-func (h *PeeringHandler) ProcessPeeringsNow() {
-	h.processPeerings()
-}
-
 // processPeerings extracts peerings from AS paths and writes to database
 func (h *PeeringHandler) processPeerings() {
 	// Take the accumulated paths and replace the map with a fresh empty one
@@ -160,11 +171,14 @@ func (h *PeeringHandler) processPeerings() {
 	type peeringKey struct {
 		low, high int
 	}
+
 	peerings := make(map[peeringKey]time.Time)
 
 	for pathJSON, timestamp := range paths {
 		var path []int
-		if err := json.Unmarshal([]byte(pathJSON), &path); err != nil {
+
+		err := json.Unmarshal([]byte(pathJSON), &path)
+		if err != nil {
 			h.logger.Error("Failed to decode AS path", "error", err)
 
 			continue
@@ -197,6 +211,7 @@ func (h *PeeringHandler) processPeerings() {
 	// Record peerings in database
 	start := time.Now()
 	successCount := 0
+
 	for key, ts := range peerings {
 		err := h.db.RecordPeering(key.low, key.high, ts)
 		if err != nil {
@@ -218,13 +233,3 @@ func (h *PeeringHandler) processPeerings() {
 		"duration", time.Since(start),
 	)
 }
-
-// Stop gracefully shuts down the handler by signaling the background
-// goroutine to stop and performing a final synchronous processing of
-// any remaining AS paths. This ensures no peering data is lost during
-// shutdown.
-func (h *PeeringHandler) Stop() {
-	close(h.stopCh)
-	// Process any remaining peerings synchronously
-	h.processPeerings()
-}

internal/routewatch/peeringhandler_internal_test.go

@@ -53,7 +53,8 @@ func newTestHandler(db database.Store) *PeeringHandler {
 	}
 }
 
-func snapshot(h *PeeringHandler) (tracked, dropped int) {
+// snapshot returns how many paths the handler tracks and how many it dropped.
+func snapshot(h *PeeringHandler) (int, int) {
 	h.mu.Lock()
 	defer h.mu.Unlock()
 
@@ -82,6 +83,8 @@ func handle(h *PeeringHandler, ts time.Time, asns ...int) {
 // empties the path map (the swap) and that a path seen again after a run is
 // recorded in the next run too.
 func TestPeeringHandlerProcessesEachRunAndEmpties(t *testing.T) {
+	t.Parallel()
+
 	store := &recordingStore{mockStore: newMockStore()}
 	h := newTestHandler(store)
 
@@ -94,6 +97,7 @@ func TestPeeringHandlerProcessesEachRunAndEmpties(t *testing.T) {
 	if tracked, _ := snapshot(h); tracked != 0 {
 		t.Fatalf("map not empty after first run: %d paths remain", tracked)
 	}
+
 	if got := store.callCount(); got != 1 {
 		t.Fatalf("want 1 RecordPeering call after first run, got %d", got)
 	}
@@ -105,6 +109,7 @@ func TestPeeringHandlerProcessesEachRunAndEmpties(t *testing.T) {
 	if tracked, _ := snapshot(h); tracked != 0 {
 		t.Fatalf("map not empty after second run: %d paths remain", tracked)
 	}
+
 	if got := store.callCount(); got != 2 {
 		t.Fatalf("want 2 RecordPeering calls after second run, got %d", got)
 	}
@@ -114,6 +119,8 @@ func TestPeeringHandlerProcessesEachRunAndEmpties(t *testing.T) {
 // and counts them, while a path already tracked is refreshed rather than
 // dropped.
 func TestPeeringHandlerCapDropsAndCounts(t *testing.T) {
+	t.Parallel()
+
 	store := &recordingStore{mockStore: newMockStore()}
 	h := newTestHandler(store)
 
@@ -126,6 +133,7 @@ func TestPeeringHandlerCapDropsAndCounts(t *testing.T) {
 
 	h.mu.Lock()
 	h.asPaths[existing] = now
+
 	for i := 0; len(h.asPaths) < maxTrackedPaths; i++ {
 		h.asPaths[strconv.Itoa(i)] = now
 	}
@@ -138,6 +146,7 @@ func TestPeeringHandlerCapDropsAndCounts(t *testing.T) {
 	if tracked != maxTrackedPaths {
 		t.Fatalf("want map size %d after drop, got %d", maxTrackedPaths, tracked)
 	}
+
 	if dropped != 1 {
 		t.Fatalf("want dropped count 1, got %d", dropped)
 	}
@@ -150,6 +159,7 @@ func TestPeeringHandlerCapDropsAndCounts(t *testing.T) {
 	if tracked != maxTrackedPaths {
 		t.Fatalf("want map size %d after refresh, got %d", maxTrackedPaths, tracked)
 	}
+
 	if dropped != 1 {
 		t.Fatalf("want dropped count still 1 after refresh, got %d", dropped)
 	}
@@ -157,7 +167,9 @@ func TestPeeringHandlerCapDropsAndCounts(t *testing.T) {
 	h.mu.Lock()
 	gotTS := h.asPaths[existing]
 	h.mu.Unlock()
+
 	if !gotTS.Equal(refreshed) {
-		t.Fatalf("existing path timestamp not refreshed: want %v, got %v", refreshed, gotTS)
+		t.Fatalf("existing path timestamp not refreshed: want %v, got %v",
+			refreshed, gotTS)
 	}
 }

internal/routewatch/prefixhandler.go

@@ -33,8 +33,8 @@ const (
 	ipv6Version = 6
 )
 
-// PrefixHandler tracks BGP prefixes and maintains a live routing table in the database.
-// Routes are added on announcement and deleted on withdrawal.
+// PrefixHandler tracks BGP prefixes and maintains a live routing table in the
+// database. Routes are added on announcement and deleted on withdrawal.
 type PrefixHandler struct {
 	db      database.Store
 	logger  *logger.Logger
@@ -79,10 +79,11 @@ func (h *PrefixHandler) SetMetricsTracker(metrics *metrics.Tracker) {
 	h.metrics = metrics
 }
 
-// WantsMessage returns true if this handler wants to process messages of the given type
+// WantsMessage returns true if this handler wants to process messages of the
+// given type
 func (h *PrefixHandler) WantsMessage(messageType string) bool {
 	// We only care about UPDATE messages for the routing table
-	return messageType == "UPDATE"
+	return messageType == updateMessageType
 }
 
 // QueueCapacity returns the desired queue capacity for this handler
@@ -145,9 +146,16 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
 	}
 }
 
+// Stop gracefully stops the handler and flushes remaining batches
+func (h *PrefixHandler) Stop() {
+	close(h.stopCh)
+	h.wg.Wait()
+}
+
 // flushLoop runs in a goroutine and periodically flushes batches
 func (h *PrefixHandler) flushLoop() {
 	defer h.wg.Done()
+
 	ticker := time.NewTicker(prefixBatchTimeout)
 	defer ticker.Stop()
 
@@ -170,7 +178,8 @@ func (h *PrefixHandler) flushLoop() {
 	}
 }
 
-// flushBatchLocked flushes the prefix batch to the database (must be called with mutex held)
+// flushBatchLocked flushes the prefix batch to the database (must be called
+// with mutex held)
 func (h *PrefixHandler) flushBatchLocked() {
 	if len(h.batch) == 0 {
 		return
@@ -182,23 +191,29 @@ func (h *PrefixHandler) flushBatchLocked() {
 	// Group updates by prefix to deduplicate
 	// For each prefix, keep the latest update
 	prefixMap := make(map[string]prefixUpdate)
+
 	for _, update := range h.batch {
 		key := update.prefix
-		if existing, ok := prefixMap[key]; !ok || update.timestamp.After(existing.timestamp) {
+
+		existing, ok := prefixMap[key]
+		if !ok || update.timestamp.After(existing.timestamp) {
 			prefixMap[key] = update
 		}
 	}
 
 	// Collect routes to upsert and delete
-	var routesToUpsert []*database.LiveRoute
-	var routesToDelete []database.LiveRouteDeletion
+	var (
+		routesToUpsert []*database.LiveRoute
+		routesToDelete []database.LiveRouteDeletion
+	)
 
 	// Collect unique prefixes to update
 	prefixesToUpdate := make(map[string]time.Time)
 
 	for _, update := range prefixMap {
 		// Track prefix for both announcements and withdrawals
-		if _, exists := prefixesToUpdate[update.prefix]; !exists || update.timestamp.After(prefixesToUpdate[update.prefix]) {
+		seen, exists := prefixesToUpdate[update.prefix]
+		if !exists || update.timestamp.After(seen) {
 			prefixesToUpdate[update.prefix] = update.timestamp
 		}
 
@@ -212,7 +227,8 @@ func (h *PrefixHandler) flushBatchLocked() {
 			// Parse CIDR to get IP version
 			_, ipVersion, err := parseCIDR(update.prefix)
 			if err != nil {
-				h.logger.Error("Failed to parse CIDR for withdrawal", "prefix", update.prefix, "error", err)
+				h.logger.Error("Failed to parse CIDR for withdrawal",
+					"prefix", update.prefix, "error", err)
 
 				continue
 			}
@@ -227,30 +243,7 @@ func (h *PrefixHandler) flushBatchLocked() {
 		}
 	}
 
-	// Process batch operations
-	successCount := 0
-	if len(routesToUpsert) > 0 {
-		if err := h.db.UpsertLiveRouteBatch(routesToUpsert); err != nil {
-			h.logger.Error("Failed to upsert route batch", "error", err, "count", len(routesToUpsert))
-		} else {
-			successCount += len(routesToUpsert)
-		}
-	}
-
-	if len(routesToDelete) > 0 {
-		if err := h.db.DeleteLiveRouteBatch(routesToDelete); err != nil {
-			h.logger.Error("Failed to delete route batch", "error", err, "count", len(routesToDelete))
-		} else {
-			successCount += len(routesToDelete)
-		}
-	}
-
-	// Update prefix tables
-	if len(prefixesToUpdate) > 0 {
-		if err := h.db.UpdatePrefixesBatch(prefixesToUpdate); err != nil {
-			h.logger.Error("Failed to update prefix batch", "error", err, "count", len(prefixesToUpdate))
-		}
-	}
+	successCount := h.writeBatch(routesToUpsert, routesToDelete, prefixesToUpdate)
 
 	elapsed := time.Since(startTime)
 	h.logger.Debug("Flushed prefix batch",
@@ -265,6 +258,47 @@ func (h *PrefixHandler) flushBatchLocked() {
 	h.lastFlush = time.Now()
 }
 
+// writeBatch writes the collected route upserts, route deletions and prefix
+// updates to the database and returns how many routes it wrote.
+func (h *PrefixHandler) writeBatch(
+	routesToUpsert []*database.LiveRoute,
+	routesToDelete []database.LiveRouteDeletion,
+	prefixesToUpdate map[string]time.Time,
+) int {
+	successCount := 0
+
+	if len(routesToUpsert) > 0 {
+		err := h.db.UpsertLiveRouteBatch(routesToUpsert)
+		if err != nil {
+			h.logger.Error("Failed to upsert route batch",
+				"error", err, "count", len(routesToUpsert))
+		} else {
+			successCount += len(routesToUpsert)
+		}
+	}
+
+	if len(routesToDelete) > 0 {
+		err := h.db.DeleteLiveRouteBatch(routesToDelete)
+		if err != nil {
+			h.logger.Error("Failed to delete route batch",
+				"error", err, "count", len(routesToDelete))
+		} else {
+			successCount += len(routesToDelete)
+		}
+	}
+
+	// Update prefix tables
+	if len(prefixesToUpdate) > 0 {
+		err := h.db.UpdatePrefixesBatch(prefixesToUpdate)
+		if err != nil {
+			h.logger.Error("Failed to update prefix batch",
+				"error", err, "count", len(prefixesToUpdate))
+		}
+	}
+
+	return successCount
+}
+
 // canonicalPrefix returns prefix in the text form net/netip prints, the form
 // the IP lookup builds when it looks a prefix up. The feed sends IPv6
 // announcements compressed ("2001:db8::/32") but IPv6 withdrawals uncompressed
@@ -280,8 +314,8 @@ func canonicalPrefix(prefix string) string {
 	return p.Masked().String()
 }
 
-// parseCIDR extracts the mask length and IP version from a prefix string
-func parseCIDR(prefix string) (maskLength int, ipVersion int, err error) {
+// parseCIDR returns the mask length and the IP version of a prefix string
+func parseCIDR(prefix string) (int, int, error) {
 	_, ipNet, err := net.ParseCIDR(prefix)
 	if err != nil {
 		return 0, 0, err
@@ -295,50 +329,6 @@ func parseCIDR(prefix string) (maskLength int, ipVersion int, err error) {
 	return ones, ipv4Version, nil
 }
 
-// processAnnouncement handles storing an announcement in the database
-// nolint:unused // kept for potential future use
-func (h *PrefixHandler) processAnnouncement(_ *database.Prefix, update prefixUpdate) {
-	// Parse CIDR to get mask length
-	maskLength, ipVersion, err := parseCIDR(update.prefix)
-	if err != nil {
-		h.logger.Error("Failed to parse CIDR",
-			"prefix", update.prefix,
-			"error", err,
-		)
-
-		return
-	}
-
-	// Track route update metrics
-	if h.metrics != nil {
-		if ipVersion == ipv4Version {
-			h.metrics.RecordIPv4Update()
-		} else {
-			h.metrics.RecordIPv6Update()
-		}
-	}
-
-	// Create live route record
-	liveRoute := &database.LiveRoute{
-		ID:          uuid.New(),
-		Prefix:      update.prefix,
-		MaskLength:  maskLength,
-		IPVersion:   ipVersion,
-		OriginASN:   update.originASN,
-		PeerIP:      update.peer,
-		ASPath:      update.path,
-		NextHop:     update.peer, // Using peer as next hop
-		LastUpdated: update.timestamp,
-	}
-
-	if err := h.db.UpsertLiveRoute(liveRoute); err != nil {
-		h.logger.Error("Failed to upsert live route",
-			"prefix", update.prefix,
-			"error", err,
-		)
-	}
-}
-
 // createLiveRoute creates a LiveRoute from a prefix update
 func (h *PrefixHandler) createLiveRoute(update prefixUpdate) *database.LiveRoute {
 	// Parse CIDR to get mask length
@@ -376,104 +366,3 @@ func (h *PrefixHandler) createLiveRoute(update prefixUpdate) *database.LiveRoute
 
 	return liveRoute
 }
-
-// processAnnouncementDirect handles storing an announcement directly without prefix table
-// nolint:unused // kept for potential future use
-func (h *PrefixHandler) processAnnouncementDirect(update prefixUpdate) {
-	// Parse CIDR to get mask length
-	maskLength, ipVersion, err := parseCIDR(update.prefix)
-	if err != nil {
-		h.logger.Error("Failed to parse CIDR",
-			"prefix", update.prefix,
-			"error", err,
-		)
-
-		return
-	}
-
-	// Track route update metrics
-	if h.metrics != nil {
-		if ipVersion == ipv4Version {
-			h.metrics.RecordIPv4Update()
-		} else {
-			h.metrics.RecordIPv6Update()
-		}
-	}
-
-	// Create live route record
-	liveRoute := &database.LiveRoute{
-		ID:          uuid.New(),
-		Prefix:      update.prefix,
-		MaskLength:  maskLength,
-		IPVersion:   ipVersion,
-		OriginASN:   update.originASN,
-		PeerIP:      update.peer,
-		ASPath:      update.path,
-		NextHop:     update.peer, // Using peer as next hop
-		LastUpdated: update.timestamp,
-	}
-
-	if err := h.db.UpsertLiveRoute(liveRoute); err != nil {
-		h.logger.Error("Failed to upsert live route",
-			"prefix", update.prefix,
-			"error", err,
-		)
-	}
-}
-
-// processWithdrawalDirect handles removing a route directly without prefix table
-// nolint:unused // kept for potential future use
-func (h *PrefixHandler) processWithdrawalDirect(update prefixUpdate) {
-	// For withdrawals, we need to delete the route from live_routes
-	if update.originASN > 0 {
-		if err := h.db.DeleteLiveRoute(update.prefix, update.originASN, update.peer); err != nil {
-			h.logger.Error("Failed to delete live route",
-				"prefix", update.prefix,
-				"origin_asn", update.originASN,
-				"peer", update.peer,
-				"error", err,
-			)
-		}
-	} else {
-		// If no origin ASN, just delete all routes for this prefix from this peer
-		if err := h.db.DeleteLiveRoute(update.prefix, 0, update.peer); err != nil {
-			h.logger.Error("Failed to delete live route (no origin ASN)",
-				"prefix", update.prefix,
-				"peer", update.peer,
-				"error", err,
-			)
-		}
-	}
-}
-
-// processWithdrawal handles removing a route from the live routing table
-// nolint:unused // kept for potential future use
-func (h *PrefixHandler) processWithdrawal(_ *database.Prefix, update prefixUpdate) {
-	// For withdrawals, we need to delete the route from live_routes
-	// Since we have the origin ASN from the update, we can delete the specific route
-	if update.originASN > 0 {
-		if err := h.db.DeleteLiveRoute(update.prefix, update.originASN, update.peer); err != nil {
-			h.logger.Error("Failed to delete live route",
-				"prefix", update.prefix,
-				"origin_asn", update.originASN,
-				"peer", update.peer,
-				"error", err,
-			)
-		}
-	} else {
-		// If no origin ASN, just delete all routes for this prefix from this peer
-		if err := h.db.DeleteLiveRoute(update.prefix, 0, update.peer); err != nil {
-			h.logger.Error("Failed to delete live route (no origin ASN)",
-				"prefix", update.prefix,
-				"peer", update.peer,
-				"error", err,
-			)
-		}
-	}
-}
-
-// Stop gracefully stops the handler and flushes remaining batches
-func (h *PrefixHandler) Stop() {
-	close(h.stopCh)
-	h.wg.Wait()
-}

internal/routewatch/prefixhandler_internal_test.go

@@ -11,7 +11,10 @@ import (
 	"git.eeqj.de/sneak/routewatch/internal/ristypes"
 )
 
-const testPeerIP = "2001:db8:ffff::1"
+const (
+	testPeerIP   = "2001:db8:ffff::1"
+	testV4Prefix = "192.0.2.0/24"
+)
 
 // TestPrefixHandlerStoresPrefixesTheIPLookupFinds runs announcements and
 // withdrawals through the prefix handler into a real database and looks the
@@ -20,6 +23,8 @@ const testPeerIP = "2001:db8:ffff::1"
 // remove the route the announcement stored, and the IP lookup must find the
 // stored prefix.
 func TestPrefixHandlerStoresPrefixesTheIPLookupFinds(t *testing.T) {
+	t.Parallel()
+
 	db, err := database.New(&config.Config{StateDir: t.TempDir()}, logger.New())
 	if err != nil {
 		t.Fatalf("failed to create database: %v", err)
@@ -31,6 +36,7 @@ func TestPrefixHandlerStoresPrefixesTheIPLookupFinds(t *testing.T) {
 	flush := func() {
 		h.mu.Lock()
 		defer h.mu.Unlock()
+
 		h.flushBatchLocked()
 	}
 
@@ -41,34 +47,38 @@ func TestPrefixHandlerStoresPrefixesTheIPLookupFinds(t *testing.T) {
 		ParsedTimestamp: ts,
 		Announcements: []ristypes.RISAnnouncement{{
 			NextHop:  testPeerIP,
-			Prefixes: []string{"2001:db8:1::/48", "192.0.2.0/24"},
+			Prefixes: []string{"2001:db8:1::/48", testV4Prefix},
 		}},
 	})
 	flush()
 
 	for ip, want := range map[string]string{
 		"2001:db8:1::1": "2001:db8:1::/48",
-		"192.0.2.1":     "192.0.2.0/24",
+		"192.0.2.1":     testV4Prefix,
 	} {
 		info, err := db.GetIPInfo(ip)
 		if err != nil {
 			t.Fatalf("GetIPInfo(%s) after announcement: %v", ip, err)
 		}
+
 		if info.Netblock != want || info.ASN != testASNB {
-			t.Errorf("GetIPInfo(%s) = %s AS%d, want %s AS%d", ip, info.Netblock, info.ASN, want, testASNB)
+			t.Errorf("GetIPInfo(%s) = %s AS%d, want %s AS%d",
+				ip, info.Netblock, info.ASN, want, testASNB)
 		}
 	}
 
 	h.HandleMessage(&ristypes.RISMessage{
 		Peer:            testPeerIP,
 		ParsedTimestamp: ts.Add(time.Minute),
-		Withdrawals:     []string{"2001:db8:1:0:0:0:0:0/48", "192.0.2.0/24"},
+		Withdrawals:     []string{"2001:db8:1:0:0:0:0:0/48", testV4Prefix},
 	})
 	flush()
 
 	for _, ip := range []string{"2001:db8:1::1", "192.0.2.1"} {
-		if info, err := db.GetIPInfo(ip); !errors.Is(err, database.ErrNoRoute) {
-			t.Errorf("GetIPInfo(%s) after withdrawal = %+v, %v; want ErrNoRoute", ip, info, err)
+		info, err := db.GetIPInfo(ip)
+		if !errors.Is(err, database.ErrNoRoute) {
+			t.Errorf("GetIPInfo(%s) after withdrawal = %+v, %v; want ErrNoRoute",
+				ip, info, err)
 		}
 	}
 }