Serve /api/v1/stats counts from realtime in-memory counters (closes #27)

Realtime in-memory counters seeded at startup and adjusted on every insert, update and delete; no periodic recompute. Independent review passed: #29 (comment)

model: claude-opus-4-8 (implementation and review); merged by claude-fable-5

internal/database/database.go

@@ -72,6 +72,7 @@ type Database struct {
 	mu       sync.Mutex
 	lockedAt time.Time
 	lockedBy string
+	counts   *liveCounts
 }
 
 // New creates a new database connection and initializes the schema.
@@ -115,12 +116,19 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) {
 	db.SetMaxIdleConns(maxConns)
 	db.SetConnMaxLifetime(0)
 
-	database := &Database{db: db, logger: logger, path: dbPath}
+	database := &Database{db: db, logger: logger, path: dbPath, counts: &liveCounts{}}
 
 	if err := database.Initialize(); err != nil {
 		return nil, fmt.Errorf("failed to initialize database: %w", err)
 	}
 
+	// Seed the in-memory statistics counters from the tables once, before the
+	// streamer starts writing. From here on every write keeps them current, so
+	// the stats endpoints never scan the tables to report counts.
+	if err := database.seedCounts(context.Background()); err != nil {
+		return nil, fmt.Errorf("failed to seed statistics counters: %w", err)
+	}
+
 	return database, nil
 }
 
@@ -208,6 +216,81 @@ func (d *Database) beginTx() (*loggingTx, error) {
 	return &loggingTx{Tx: tx, logger: d.logger}, nil
 }
 
+// A live-route upsert is an UPDATE followed, only when no row matched, by an
+// INSERT. The UPDATE's rows-affected count (1 for an existing key, 0 for a new
+// one) is what lets the in-memory route counters stay exact without a COUNT(*).
+// Callers hold the database write lock, so no other writer can insert the same
+// key between the two statements. The id column is set only on INSERT, so an
+// updated route keeps its original id, exactly as the previous ON CONFLICT
+// upsert did.
+const (
+	updateLiveRouteV4SQL = `UPDATE live_routes_v4 SET mask_length = ?, as_path = ?, next_hop = ?,
+		last_updated = ?, ip_start = ?, ip_end = ? WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`
+
+	insertLiveRouteV4SQL = `INSERT INTO live_routes_v4 (id, prefix, mask_length, origin_asn, peer_ip,
+		as_path, next_hop, last_updated, ip_start, ip_end) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`
+
+	updateLiveRouteV6SQL = `UPDATE live_routes_v6 SET mask_length = ?, as_path = ?, next_hop = ?,
+		last_updated = ? WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`
+
+	insertLiveRouteV6SQL = `INSERT INTO live_routes_v6 (id, prefix, mask_length, origin_asn, peer_ip,
+		as_path, next_hop, last_updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`
+)
+
+// upsertRouteRowV4 updates an IPv4 live route, inserting it when no row matched,
+// and reports whether a new row was inserted.
+func upsertRouteRowV4(upd, ins *sql.Stmt, route *LiveRoute, pathJSON string) (inserted bool, err error) {
+	if route.V4IPStart == nil || route.V4IPEnd == nil {
+		return false, fmt.Errorf("IPv4 route %s missing range values", route.Prefix)
+	}
+
+	res, err := upd.Exec(route.MaskLength, pathJSON, route.NextHop, route.LastUpdated,
+		*route.V4IPStart, *route.V4IPEnd, route.Prefix, route.OriginASN, route.PeerIP)
+	if err != nil {
+		return false, err
+	}
+	affected, err := res.RowsAffected()
+	if err != nil {
+		return false, err
+	}
+	if affected > 0 {
+		return false, nil
+	}
+
+	_, err = ins.Exec(route.ID.String(), route.Prefix, route.MaskLength, route.OriginASN,
+		route.PeerIP, pathJSON, route.NextHop, route.LastUpdated, *route.V4IPStart, *route.V4IPEnd)
+	if err != nil {
+		return false, err
+	}
+
+	return true, nil
+}
+
+// upsertRouteRowV6 updates an IPv6 live route, inserting it when no row matched,
+// and reports whether a new row was inserted.
+func upsertRouteRowV6(upd, ins *sql.Stmt, route *LiveRoute, pathJSON string) (inserted bool, err error) {
+	res, err := upd.Exec(route.MaskLength, pathJSON, route.NextHop, route.LastUpdated,
+		route.Prefix, route.OriginASN, route.PeerIP)
+	if err != nil {
+		return false, err
+	}
+	affected, err := res.RowsAffected()
+	if err != nil {
+		return false, err
+	}
+	if affected > 0 {
+		return false, nil
+	}
+
+	_, err = ins.Exec(route.ID.String(), route.Prefix, route.MaskLength, route.OriginASN,
+		route.PeerIP, pathJSON, route.NextHop, route.LastUpdated)
+	if err != nil {
+		return false, err
+	}
+
+	return true, nil
+}
+
 // UpsertLiveRouteBatch inserts or updates multiple live routes in a single transaction
 func (d *Database) UpsertLiveRouteBatch(routes []*LiveRoute) error {
 	if len(routes) == 0 {
@@ -227,91 +310,65 @@ func (d *Database) UpsertLiveRouteBatch(routes []*LiveRoute) error {
 		}
 	}()
 
-	// Prepare statements for both IPv4 and IPv6
-	queryV4 := `
-		INSERT INTO live_routes_v4 (id, prefix, mask_length, origin_asn, peer_ip, as_path, next_hop, 
-			last_updated, ip_start, ip_end)
-		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
-		ON CONFLICT(prefix, origin_asn, peer_ip) DO UPDATE SET
-			mask_length = excluded.mask_length,
-			as_path = excluded.as_path,
-			next_hop = excluded.next_hop,
-			last_updated = excluded.last_updated,
-			ip_start = excluded.ip_start,
-			ip_end = excluded.ip_end
-	`
-
-	queryV6 := `
-		INSERT INTO live_routes_v6 (id, prefix, mask_length, origin_asn, peer_ip, as_path, next_hop, 
-			last_updated)
-		VALUES (?, ?, ?, ?, ?, ?, ?, ?)
-		ON CONFLICT(prefix, origin_asn, peer_ip) DO UPDATE SET
-			mask_length = excluded.mask_length,
-			as_path = excluded.as_path,
-			next_hop = excluded.next_hop,
-			last_updated = excluded.last_updated
-	`
-
-	stmtV4, err := tx.Prepare(queryV4)
+	// Prepare the update and insert statements for both tables.
+	updV4, err := tx.Prepare(updateLiveRouteV4SQL)
 	if err != nil {
-		return fmt.Errorf("failed to prepare IPv4 statement: %w", err)
+		return fmt.Errorf("failed to prepare IPv4 update statement: %w", err)
 	}
-	defer func() { _ = stmtV4.Close() }()
+	defer func() { _ = updV4.Close() }()
 
-	stmtV6, err := tx.Prepare(queryV6)
+	insV4, err := tx.Prepare(insertLiveRouteV4SQL)
 	if err != nil {
-		return fmt.Errorf("failed to prepare IPv6 statement: %w", err)
+		return fmt.Errorf("failed to prepare IPv4 insert statement: %w", err)
 	}
-	defer func() { _ = stmtV6.Close() }()
+	defer func() { _ = insV4.Close() }()
 
+	updV6, err := tx.Prepare(updateLiveRouteV6SQL)
+	if err != nil {
+		return fmt.Errorf("failed to prepare IPv6 update statement: %w", err)
+	}
+	defer func() { _ = updV6.Close() }()
+
+	insV6, err := tx.Prepare(insertLiveRouteV6SQL)
+	if err != nil {
+		return fmt.Errorf("failed to prepare IPv6 insert statement: %w", err)
+	}
+	defer func() { _ = insV6.Close() }()
+
+	var newV4, newV6 int
 	for _, route := range routes {
-		// Encode AS path as JSON
 		pathJSON, err := json.Marshal(route.ASPath)
 		if err != nil {
 			return fmt.Errorf("failed to encode AS path: %w", err)
 		}
 
-		// Use appropriate statement based on IP version
 		if route.IPVersion == ipVersionV4 {
-			// IPv4 routes must have range values
-			if route.V4IPStart == nil || route.V4IPEnd == nil {
-				return fmt.Errorf("IPv4 route %s missing range values", route.Prefix)
+			inserted, err := upsertRouteRowV4(updV4, insV4, route, string(pathJSON))
+			if err != nil {
+				return fmt.Errorf("failed to upsert route %s: %w", route.Prefix, err)
+			}
+			if inserted {
+				newV4++
 			}
 
-			_, err = stmtV4.Exec(
-				route.ID.String(),
-				route.Prefix,
-				route.MaskLength,
-				route.OriginASN,
-				route.PeerIP,
-				string(pathJSON),
-				route.NextHop,
-				route.LastUpdated,
-				*route.V4IPStart,
-				*route.V4IPEnd,
-			)
-		} else {
-			// IPv6 routes
-			_, err = stmtV6.Exec(
-				route.ID.String(),
-				route.Prefix,
-				route.MaskLength,
-				route.OriginASN,
-				route.PeerIP,
-				string(pathJSON),
-				route.NextHop,
-				route.LastUpdated,
-			)
+			continue
 		}
+
+		inserted, err := upsertRouteRowV6(updV6, insV6, route, string(pathJSON))
 		if err != nil {
 			return fmt.Errorf("failed to upsert route %s: %w", route.Prefix, err)
 		}
+		if inserted {
+			newV6++
+		}
 	}
 
 	if err = tx.Commit(); err != nil {
 		return fmt.Errorf("failed to commit transaction: %w", err)
 	}
 
+	d.counts.addRoutes(newV4, newV6)
+
 	return nil
 }
 
@@ -362,6 +419,7 @@ func (d *Database) DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error {
 	defer func() { _ = stmtV6WithoutOrigin.Close() }()
 
 	// Process deletions
+	var deletedV4, deletedV6 int64
 	for _, del := range deletions {
 		var stmt *sql.Stmt
 
@@ -382,21 +440,36 @@ func (d *Database) DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error {
 		}
 
 		// Execute deletion
+		var res sql.Result
 		if del.OriginASN == 0 {
-			_, err = stmt.Exec(del.Prefix, del.PeerIP)
+			res, err = stmt.Exec(del.Prefix, del.PeerIP)
 		} else {
-			_, err = stmt.Exec(del.Prefix, del.OriginASN, del.PeerIP)
+			res, err = stmt.Exec(del.Prefix, del.OriginASN, del.PeerIP)
 		}
 
 		if err != nil {
 			return fmt.Errorf("failed to delete route %s: %w", del.Prefix, err)
 		}
+
+		// A deletion with no origin ASN can remove several rows, so use the
+		// exact rows-affected count to keep the in-memory route counters right.
+		affected, err := res.RowsAffected()
+		if err != nil {
+			return fmt.Errorf("failed to count deleted route %s: %w", del.Prefix, err)
+		}
+		if del.IPVersion == ipVersionV4 {
+			deletedV4 += affected
+		} else {
+			deletedV6 += affected
+		}
 	}
 
 	if err = tx.Commit(); err != nil {
 		return fmt.Errorf("failed to commit transaction: %w", err)
 	}
 
+	d.counts.addRoutes(-int(deletedV4), -int(deletedV6))
+
 	return nil
 }
 
@@ -456,6 +529,7 @@ func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
 	}
 	defer func() { _ = insertV6Stmt.Close() }()
 
+	var newV4, newV6 int
 	for prefix, timestamp := range prefixes {
 		ipVersion := detectIPVersion(prefix)
 
@@ -482,6 +556,11 @@ func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
 			if err != nil {
 				return fmt.Errorf("failed to insert prefix %s: %w", prefix, err)
 			}
+			if ipVersion == ipVersionV4 {
+				newV4++
+			} else {
+				newV6++
+			}
 		default:
 			return fmt.Errorf("failed to query prefix %s: %w", prefix, err)
 		}
@@ -491,6 +570,8 @@ func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
 		return fmt.Errorf("failed to commit transaction: %w", err)
 	}
 
+	d.counts.addPrefixes(newV4, newV6)
+
 	return nil
 }
 
@@ -534,6 +615,7 @@ func (d *Database) GetOrCreateASNBatch(asns map[int]time.Time) error {
 	}
 	defer func() { _ = insertStmt.Close() }()
 
+	var newASNs int
 	for number, timestamp := range asns {
 		var asn ASN
 		var handle, description sql.NullString
@@ -568,6 +650,7 @@ func (d *Database) GetOrCreateASNBatch(asns map[int]time.Time) error {
 			if err != nil {
 				return fmt.Errorf("failed to insert ASN %d: %w", number, err)
 			}
+			newASNs++
 
 			continue
 		}
@@ -581,6 +664,8 @@ func (d *Database) GetOrCreateASNBatch(asns map[int]time.Time) error {
 		return fmt.Errorf("failed to commit transaction: %w", err)
 	}
 
+	d.counts.addASNs(newASNs)
+
 	return nil
 }
 
@@ -652,6 +737,8 @@ func (d *Database) GetOrCreateASN(number int, timestamp time.Time) (*ASN, error)
 		return nil, err
 	}
 
+	d.counts.addASNs(1)
+
 	return &asn, nil
 }
 
@@ -727,6 +814,12 @@ func (d *Database) GetOrCreatePrefix(prefix string, timestamp time.Time) (*Prefi
 		return nil, err
 	}
 
+	if ipVersion == ipVersionV4 {
+		d.counts.addPrefixes(1, 0)
+	} else {
+		d.counts.addPrefixes(0, 1)
+	}
+
 	return &p, nil
 }
 
@@ -803,6 +896,10 @@ func (d *Database) RecordPeering(asA, asB int, timestamp time.Time) error {
 		return err
 	}
 
+	if !exists {
+		d.counts.addPeerings(1)
+	}
+
 	return nil
 }
 
@@ -846,6 +943,7 @@ func (d *Database) UpdatePeerBatch(peers map[string]PeerUpdate) error {
 	}
 	defer func() { _ = insertStmt.Close() }()
 
+	var newPeers int
 	for _, update := range peers {
 		var exists bool
 		err = checkStmt.QueryRow(update.PeerIP).Scan(&exists)
@@ -863,12 +961,17 @@ func (d *Database) UpdatePeerBatch(peers map[string]PeerUpdate) error {
 		if err != nil {
 			return fmt.Errorf("failed to update peer %s: %w", update.PeerIP, err)
 		}
+		if !exists {
+			newPeers++
+		}
 	}
 
 	if err = tx.Commit(); err != nil {
 		return fmt.Errorf("failed to commit transaction: %w", err)
 	}
 
+	d.counts.addPeers(newPeers)
+
 	return nil
 }
 
@@ -918,6 +1021,10 @@ func (d *Database) UpdatePeer(peerIP string, peerASN int, messageType string, ti
 		return err
 	}
 
+	if !exists {
+		d.counts.addPeers(1)
+	}
+
 	return nil
 }
 
@@ -926,169 +1033,161 @@ func (d *Database) GetStats() (Stats, error) {
 	return d.GetStatsContext(context.Background())
 }
 
-// GetStatsContext returns database statistics with context support
+// GetStatsContext returns database statistics with context support.
+//
+// The row counts (ASNs, prefixes, peerings, peers, live routes) come from the
+// in-memory counters, seeded at startup and kept current on every write, so a
+// read runs no COUNT(*) over the tables. The oldest/newest route timestamps are
+// read from the ends of the last_updated index, and the file size from a
+// stat(); neither is a table scan. The only remaining query is the prefix
+// distribution.
 func (d *Database) GetStatsContext(ctx context.Context) (Stats, error) {
 	var stats Stats
 
-	// Count ASNs
-	err := d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM asns").Scan(&stats.ASNs)
-	if err != nil {
-		return stats, err
-	}
+	// Row counts from memory, as a single consistent snapshot.
+	d.counts.fill(&stats)
 
-	// Count prefixes from both tables
-	err = d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM prefixes_v4").Scan(&stats.IPv4Prefixes)
-	if err != nil {
-		return stats, err
-	}
-
-	err = d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM prefixes_v6").Scan(&stats.IPv6Prefixes)
-	if err != nil {
-		return stats, err
-	}
-
-	stats.Prefixes = stats.IPv4Prefixes + stats.IPv6Prefixes
-
-	// Count peerings
-	err = d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM peerings").Scan(&stats.Peerings)
-	if err != nil {
-		return stats, err
-	}
-
-	// Count peers
-	err = d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM bgp_peers").Scan(&stats.Peers)
-	if err != nil {
-		return stats, err
-	}
-
-	// Get database file size
-	fileInfo, err := os.Stat(d.path)
-	if err != nil {
+	// Database file size is a cheap stat() on the file.
+	if fileInfo, err := os.Stat(d.path); err != nil {
 		d.logger.Warn("Failed to get database file size", "error", err)
-		stats.FileSizeBytes = 0
 	} else {
 		stats.FileSizeBytes = fileInfo.Size()
 	}
 
-	// Get live routes count from both tables
-	var v4Count, v6Count int
-	err = d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM live_routes_v4").Scan(&v4Count)
-	if err != nil {
-		return stats, fmt.Errorf("failed to count IPv4 routes: %w", err)
-	}
-	err = d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM live_routes_v6").Scan(&v6Count)
-	if err != nil {
-		return stats, fmt.Errorf("failed to count IPv6 routes: %w", err)
-	}
-	stats.LiveRoutes = v4Count + v6Count
-
-	// Get oldest and newest route timestamps
-	routeTimestampQuery := `
-		SELECT MIN(last_updated), MAX(last_updated) FROM (
-			SELECT last_updated FROM live_routes_v4
-			UNION ALL
-			SELECT last_updated FROM live_routes_v6
-		)
-	`
-	var oldestRoute, newestRoute *time.Time
-	err = d.db.QueryRowContext(ctx, routeTimestampQuery).Scan(&oldestRoute, &newestRoute)
+	// Oldest and newest route timestamps read one row from each end of the
+	// last_updated index (a log-time lookup, not a scan). Selecting the column
+	// directly lets the driver parse the DATETIME into time.Time; the old
+	// MIN/MAX union scan returned an untyped string that failed to scan and
+	// logged a warning on every call.
+	oldest, newest, err := d.routeTimestampRange(ctx)
 	if err != nil {
+		// Display-only fields; log but keep the rest of the stats.
 		d.logger.Warn("Failed to get route timestamps", "error", err)
 	} else {
-		stats.OldestRoute = oldestRoute
-		stats.NewestRoute = newestRoute
+		stats.OldestRoute = oldest
+		stats.NewestRoute = newest
 	}
 
-	// Get prefix distribution
+	// Prefix distribution counts distinct prefixes per mask length. It stays a
+	// query over the covering (mask_length, prefix) index rather than an
+	// in-memory counter: maintaining distinct-prefix-per-mask in memory would
+	// need a per-prefix table of roughly a million entries, memory this service
+	// is tuned to avoid.
 	stats.IPv4PrefixDistribution, stats.IPv6PrefixDistribution, err = d.GetPrefixDistributionContext(ctx)
 	if err != nil {
-		// Log but don't fail
+		// Log but don't fail.
 		d.logger.Warn("Failed to get prefix distribution", "error", err)
 	}
 
 	return stats, nil
 }
 
+// routeTimestampRange returns the earliest and latest last_updated across both
+// live route tables, or nil values when both tables are empty. Each query reads
+// one row from an end of the last_updated index rather than scanning the tables.
+func (d *Database) routeTimestampRange(ctx context.Context) (oldest, newest *time.Time, err error) {
+	oldestV4, ok, err := d.scanRouteTimestamp(ctx,
+		"SELECT last_updated FROM live_routes_v4 ORDER BY last_updated ASC LIMIT 1")
+	if err != nil {
+		return nil, nil, err
+	}
+	if ok {
+		oldest = &oldestV4
+	}
+
+	oldestV6, ok, err := d.scanRouteTimestamp(ctx,
+		"SELECT last_updated FROM live_routes_v6 ORDER BY last_updated ASC LIMIT 1")
+	if err != nil {
+		return nil, nil, err
+	}
+	if ok && (oldest == nil || oldestV6.Before(*oldest)) {
+		oldest = &oldestV6
+	}
+
+	newestV4, ok, err := d.scanRouteTimestamp(ctx,
+		"SELECT last_updated FROM live_routes_v4 ORDER BY last_updated DESC LIMIT 1")
+	if err != nil {
+		return nil, nil, err
+	}
+	if ok {
+		newest = &newestV4
+	}
+
+	newestV6, ok, err := d.scanRouteTimestamp(ctx,
+		"SELECT last_updated FROM live_routes_v6 ORDER BY last_updated DESC LIMIT 1")
+	if err != nil {
+		return nil, nil, err
+	}
+	if ok && (newest == nil || newestV6.After(*newest)) {
+		newest = &newestV6
+	}
+
+	return oldest, newest, nil
+}
+
+// scanRouteTimestamp runs a single-row timestamp query. ok is false when the
+// table is empty. The query selects the last_updated column directly so the
+// driver parses the DATETIME value into a time.Time.
+func (d *Database) scanRouteTimestamp(ctx context.Context, query string) (ts time.Time, ok bool, err error) {
+	err = d.db.QueryRowContext(ctx, query).Scan(&ts)
+	switch {
+	case errors.Is(err, sql.ErrNoRows):
+		return time.Time{}, false, nil
+	case err != nil:
+		return time.Time{}, false, err
+	default:
+		return ts, true, nil
+	}
+}
+
 // UpsertLiveRoute inserts or updates a live route
 func (d *Database) UpsertLiveRoute(route *LiveRoute) error {
 	d.lock("UpsertLiveRoute")
 	defer d.unlock()
 
-	// Choose table based on IP version
-	tableName := "live_routes_v4"
-	if route.IPVersion == ipVersionV6 {
-		tableName = "live_routes_v6"
-	}
-
-	var query string
-	if route.IPVersion == ipVersionV4 {
-		query = fmt.Sprintf(`
-			INSERT INTO %s (id, prefix, mask_length, origin_asn, peer_ip, as_path, next_hop, 
-				last_updated, ip_start, ip_end)
-			VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
-			ON CONFLICT(prefix, origin_asn, peer_ip) DO UPDATE SET
-				mask_length = excluded.mask_length,
-				as_path = excluded.as_path,
-				next_hop = excluded.next_hop,
-				last_updated = excluded.last_updated,
-				ip_start = excluded.ip_start,
-				ip_end = excluded.ip_end
-		`, tableName)
-	} else {
-		query = fmt.Sprintf(`
-			INSERT INTO %s (id, prefix, mask_length, origin_asn, peer_ip, as_path, next_hop, 
-				last_updated)
-			VALUES (?, ?, ?, ?, ?, ?, ?, ?)
-			ON CONFLICT(prefix, origin_asn, peer_ip) DO UPDATE SET
-				mask_length = excluded.mask_length,
-				as_path = excluded.as_path,
-				next_hop = excluded.next_hop,
-				last_updated = excluded.last_updated
-		`, tableName)
-	}
-
-	// Encode AS path as JSON
 	pathJSON, err := json.Marshal(route.ASPath)
 	if err != nil {
 		return fmt.Errorf("failed to encode AS path: %w", err)
 	}
 
-	if route.IPVersion == ipVersionV4 {
-		// Convert v4_ip_start and v4_ip_end to interface{} for SQL NULL handling
-		var v4Start, v4End interface{}
-		if route.V4IPStart != nil {
-			v4Start = *route.V4IPStart
-		}
-		if route.V4IPEnd != nil {
-			v4End = *route.V4IPEnd
-		}
-
-		_, err = d.db.Exec(query,
-			route.ID.String(),
-			route.Prefix,
-			route.MaskLength,
-			route.OriginASN,
-			route.PeerIP,
-			string(pathJSON),
-			route.NextHop,
-			route.LastUpdated,
-			v4Start,
-			v4End,
-		)
-	} else {
-		_, err = d.db.Exec(query,
-			route.ID.String(),
-			route.Prefix,
-			route.MaskLength,
-			route.OriginASN,
-			route.PeerIP,
-			string(pathJSON),
-			route.NextHop,
-			route.LastUpdated,
-		)
+	updateSQL, insertSQL := updateLiveRouteV4SQL, insertLiveRouteV4SQL
+	if route.IPVersion == ipVersionV6 {
+		updateSQL, insertSQL = updateLiveRouteV6SQL, insertLiveRouteV6SQL
 	}
 
-	return err
+	// The write lock is held, so no other writer can insert this key between the
+	// update and the insert even though they are separate autocommit statements.
+	upd, err := d.db.Prepare(updateSQL)
+	if err != nil {
+		return fmt.Errorf("failed to prepare update statement: %w", err)
+	}
+	defer func() { _ = upd.Close() }()
+
+	ins, err := d.db.Prepare(insertSQL)
+	if err != nil {
+		return fmt.Errorf("failed to prepare insert statement: %w", err)
+	}
+	defer func() { _ = ins.Close() }()
+
+	var inserted bool
+	if route.IPVersion == ipVersionV4 {
+		inserted, err = upsertRouteRowV4(upd, ins, route, string(pathJSON))
+	} else {
+		inserted, err = upsertRouteRowV6(upd, ins, route, string(pathJSON))
+	}
+	if err != nil {
+		return fmt.Errorf("failed to upsert route %s: %w", route.Prefix, err)
+	}
+
+	if inserted {
+		if route.IPVersion == ipVersionV4 {
+			d.counts.addRoutes(1, 0)
+		} else {
+			d.counts.addRoutes(0, 1)
+		}
+	}
+
+	return nil
 }
 
 // DeleteLiveRoute deletes a live route
@@ -1103,23 +1202,41 @@ func (d *Database) DeleteLiveRoute(prefix string, originASN int, peerIP string)
 		return fmt.Errorf("invalid prefix format: %w", err)
 	}
 
-	tableName := "live_routes_v4"
-	if ipnet.IP.To4() == nil {
-		tableName = "live_routes_v6"
+	isV4 := ipnet.IP.To4() != nil
+
+	// Literal per-table queries (rather than one formatted with the table name)
+	// so the delete carries no dynamically built SQL. A delete with no origin
+	// ASN can remove several rows.
+	var res sql.Result
+	switch {
+	case isV4 && originASN == 0:
+		res, err = d.db.Exec(`DELETE FROM live_routes_v4 WHERE prefix = ? AND peer_ip = ?`, prefix, peerIP)
+	case isV4:
+		res, err = d.db.Exec(
+			`DELETE FROM live_routes_v4 WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`,
+			prefix, originASN, peerIP)
+	case originASN == 0:
+		res, err = d.db.Exec(`DELETE FROM live_routes_v6 WHERE prefix = ? AND peer_ip = ?`, prefix, peerIP)
+	default:
+		res, err = d.db.Exec(
+			`DELETE FROM live_routes_v6 WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`,
+			prefix, originASN, peerIP)
+	}
+	if err != nil {
+		return err
 	}
 
-	var query string
-	if originASN == 0 {
-		// Delete all routes for this prefix from this peer
-		query = fmt.Sprintf(`DELETE FROM %s WHERE prefix = ? AND peer_ip = ?`, tableName)
-		_, err = d.db.Exec(query, prefix, peerIP)
+	affected, err := res.RowsAffected()
+	if err != nil {
+		return err
+	}
+	if isV4 {
+		d.counts.addRoutes(-int(affected), 0)
 	} else {
-		// Delete specific route
-		query = fmt.Sprintf(`DELETE FROM %s WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`, tableName)
-		_, err = d.db.Exec(query, prefix, originASN, peerIP)
+		d.counts.addRoutes(0, -int(affected))
 	}
 
-	return err
+	return nil
 }
 
 // GetPrefixDistribution returns the distribution of unique prefixes by mask length