Store the routing table in the database (part of #122)

A new routes table holds one row per prefix and origin AS: the peer set
as the bitset's bytes, and the two kept routes, each with its peer's
number, AS path and time in seconds. A new route_peers table holds each
peer's number and address.

WriteRouteChanges writes one TakeChanges result in a transaction.
ReadRouteTable reads both tables into a new routing table, through the
new Table.Load, which records no change. It fails on a row that names a
peer number missing from route_peers. Nothing uses them yet.

Model: opus-5-5

internal/database/routes.go

@@ -0,0 +1,339 @@
+package database
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net/netip"
+	"time"
+
+	"github.com/bits-and-blooms/bitset"
+
+	"git.eeqj.de/sneak/routewatch/internal/routetable"
+)
+
+// keptRoutesPerRow is how many kept routes a row of the routes table has
+// columns for: the newest and the older one.
+const keptRoutesPerRow = 2
+
+// errUnknownRoutePeer is returned for a peer that route_peers lacks: on a
+// write, the peer of a kept route; on a read, a peer number in a row.
+var errUnknownRoutePeer = errors.New("peer not in route_peers")
+
+// The statements of the routes and route_peers tables (schema.sql).
+const (
+	insertRoutePeerSQL = `INSERT INTO route_peers (number, address)
+		VALUES (?, ?)`
+	routePeerNumberSQL = `SELECT number FROM route_peers WHERE address = ?`
+	routePeersSQL      = `SELECT number, address FROM route_peers`
+
+	upsertRouteSQL = `INSERT INTO routes (prefix, origin_asn, peers,
+			newest_peer, newest_as_path, newest_time,
+			older_peer, older_as_path, older_time)
+		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+		ON CONFLICT (prefix, origin_asn) DO UPDATE SET peers = excluded.peers,
+			newest_peer = excluded.newest_peer,
+			newest_as_path = excluded.newest_as_path,
+			newest_time = excluded.newest_time,
+			older_peer = excluded.older_peer,
+			older_as_path = excluded.older_as_path,
+			older_time = excluded.older_time`
+	deleteRouteSQL = `DELETE FROM routes WHERE prefix = ? AND origin_asn = ?`
+	routesSQL      = `SELECT prefix, origin_asn, peers,
+		newest_peer, newest_as_path, newest_time,
+		older_peer, older_as_path, older_time FROM routes`
+)
+
+// WriteRouteChanges writes, in one transaction, what one call of the routing
+// table's TakeChanges returned. It stores each new peer with its number, writes
+// the row of each changed prefix and origin AS pair, and deletes the row of
+// each removed one. Route times are stored to the second, as the feed gives
+// them.
+//
+// When it fails, it writes nothing. The caller then writes these changes and
+// peers again, because later changes can name the peers it did not store.
+func (d *Database) WriteRouteChanges(
+	changes []routetable.Change, newPeers map[string]uint,
+) error {
+	if len(changes) == 0 && len(newPeers) == 0 {
+		return nil
+	}
+
+	d.lock("WriteRouteChanges")
+	defer d.unlock()
+
+	tx, err := d.beginTx()
+	if err != nil {
+		return fmt.Errorf("failed to begin transaction: %w", err)
+	}
+	defer d.rollback(tx)
+
+	ctx := context.Background()
+
+	// The new peers first, so that the rows can look up their numbers.
+	for address, number := range newPeers {
+		_, err := tx.ExecContext(ctx, insertRoutePeerSQL, number, address)
+		if err != nil {
+			return fmt.Errorf("failed to store peer %s: %w", address, err)
+		}
+	}
+
+	err = writeRouteRows(ctx, tx, changes)
+	if err != nil {
+		return err
+	}
+
+	err = tx.Commit()
+	if err != nil {
+		return fmt.Errorf("failed to commit transaction: %w", err)
+	}
+
+	return nil
+}
+
+// writeRouteRows upserts the row of each change in tx, or deletes it for a
+// removed pair.
+func writeRouteRows(
+	ctx context.Context, tx *loggingTx, changes []routetable.Change,
+) error {
+	upsert, err := tx.PrepareContext(ctx, upsertRouteSQL)
+	if err != nil {
+		return fmt.Errorf("failed to prepare route upsert: %w", err)
+	}
+	defer func() { _ = upsert.Close() }()
+
+	del, err := tx.PrepareContext(ctx, deleteRouteSQL)
+	if err != nil {
+		return fmt.Errorf("failed to prepare route delete: %w", err)
+	}
+	defer func() { _ = del.Close() }()
+
+	peerNumber, err := tx.PrepareContext(ctx, routePeerNumberSQL)
+	if err != nil {
+		return fmt.Errorf("failed to prepare peer number lookup: %w", err)
+	}
+	defer func() { _ = peerNumber.Close() }()
+
+	for _, change := range changes {
+		prefix := change.Prefix.String()
+
+		if change.Removed {
+			_, err := del.ExecContext(ctx, prefix, change.OriginASN)
+			if err != nil {
+				return fmt.Errorf("failed to delete route %s: %w", prefix, err)
+			}
+
+			continue
+		}
+
+		row, err := routeRow(ctx, peerNumber, change)
+		if err != nil {
+			return fmt.Errorf("failed to write route %s: %w", prefix, err)
+		}
+
+		_, err = upsert.ExecContext(ctx, row...)
+		if err != nil {
+			return fmt.Errorf("failed to write route %s: %w", prefix, err)
+		}
+	}
+
+	return nil
+}
+
+// routeRow returns the values of the routes row of change, in the order of
+// upsertRouteSQL. peerNumber looks up a peer's number by its address. The
+// table keeps at most keptRoutesPerRow routes of a pair.
+func routeRow(
+	ctx context.Context, peerNumber *sql.Stmt, change routetable.Change,
+) ([]any, error) {
+	peers, err := change.Peers.MarshalBinary()
+	if err != nil {
+		return nil, err
+	}
+
+	row := []any{change.Prefix.String(), change.OriginASN, peers}
+
+	for i := range keptRoutesPerRow {
+		if i >= len(change.Routes) {
+			row = append(row, nil, nil, nil)
+
+			continue
+		}
+
+		r := change.Routes[i]
+
+		var number uint
+
+		err := peerNumber.QueryRowContext(ctx, r.Peer).Scan(&number)
+		if errors.Is(err, sql.ErrNoRows) {
+			return nil, fmt.Errorf("%w: %s", errUnknownRoutePeer, r.Peer)
+		}
+
+		if err != nil {
+			return nil, err
+		}
+
+		path, err := json.Marshal(r.ASPath)
+		if err != nil {
+			return nil, err
+		}
+
+		row = append(row, number, string(path), r.Time.Unix())
+	}
+
+	return row, nil
+}
+
+// ReadRouteTable returns the routing table that the routes and route_peers
+// tables hold, as startup reads it. It reads the peers first, so that the
+// table keeps their numbers, then loads each row into the table as it reads
+// it. The two reads are separate, so nothing may write the tables meanwhile.
+// It fails on a row that names a peer number route_peers lacks.
+func (d *Database) ReadRouteTable(ctx context.Context) (*routetable.Table, error) {
+	peerNumbers, err := d.readRoutePeers(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	addresses := make(map[uint]string, len(peerNumbers))
+	numbers := new(bitset.BitSet)
+
+	for address, number := range peerNumbers {
+		addresses[number] = address
+		numbers.Set(number)
+	}
+
+	table := routetable.New(routetable.Params{PeerNumbers: peerNumbers})
+
+	rows, err := d.db.QueryContext(ctx, routesSQL)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read routes: %w", err)
+	}
+	defer func() { _ = rows.Close() }()
+
+	for rows.Next() {
+		change, err := scanRouteRow(rows, addresses, numbers)
+		if err != nil {
+			return nil, err
+		}
+
+		table.Load(change)
+	}
+
+	err = rows.Err()
+	if err != nil {
+		return nil, fmt.Errorf("failed to read routes: %w", err)
+	}
+
+	return table, nil
+}
+
+// readRoutePeers returns the number of each peer in route_peers, by address.
+func (d *Database) readRoutePeers(ctx context.Context) (map[string]uint, error) {
+	rows, err := d.db.QueryContext(ctx, routePeersSQL)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read route peers: %w", err)
+	}
+	defer func() { _ = rows.Close() }()
+
+	peerNumbers := make(map[string]uint)
+
+	for rows.Next() {
+		var (
+			number  uint
+			address string
+		)
+
+		err := rows.Scan(&number, &address)
+		if err != nil {
+			return nil, fmt.Errorf("failed to read a route peer: %w", err)
+		}
+
+		peerNumbers[address] = number
+	}
+
+	err = rows.Err()
+	if err != nil {
+		return nil, fmt.Errorf("failed to read route peers: %w", err)
+	}
+
+	return peerNumbers, nil
+}
+
+// keptRouteColumns are the three columns of one kept route in a routes row.
+type keptRouteColumns struct {
+	peer   sql.Null[uint]
+	asPath sql.NullString
+	time   sql.NullInt64
+}
+
+// scanRouteRow returns the content of the routes row rows is at, with each kept
+// route's peer named by its address in addresses. It fails when the row names a
+// peer number that addresses lacks. numbers holds the numbers of addresses as a
+// set, against which a row's whole peer set is checked at once.
+func scanRouteRow(
+	rows *sql.Rows, addresses map[uint]string, numbers *bitset.BitSet,
+) (routetable.Change, error) {
+	var (
+		change routetable.Change
+		prefix string
+		peers  []byte
+		kept   [keptRoutesPerRow]keptRouteColumns
+	)
+
+	err := rows.Scan(&prefix, &change.OriginASN, &peers,
+		&kept[0].peer, &kept[0].asPath, &kept[0].time,
+		&kept[1].peer, &kept[1].asPath, &kept[1].time)
+	if err != nil {
+		return change, fmt.Errorf("failed to read a route: %w", err)
+	}
+
+	change.Prefix, err = netip.ParsePrefix(prefix)
+	if err != nil {
+		return change, fmt.Errorf("failed to read route %s: %w", prefix, err)
+	}
+
+	change.Peers = new(bitset.BitSet)
+
+	err = change.Peers.UnmarshalBinary(peers)
+	if err != nil {
+		return change, fmt.Errorf("failed to read the peers of route %s: %w",
+			prefix, err)
+	}
+
+	if !numbers.IsSuperSet(change.Peers) {
+		unknown, _ := change.Peers.Difference(numbers).NextSet(0)
+
+		return change, fmt.Errorf("%w: number %d in the peers of route %s",
+			errUnknownRoutePeer, unknown, prefix)
+	}
+
+	for _, columns := range kept {
+		if !columns.peer.Valid {
+			continue
+		}
+
+		address, known := addresses[columns.peer.V]
+		if !known {
+			return change, fmt.Errorf("%w: number %d in route %s",
+				errUnknownRoutePeer, columns.peer.V, prefix)
+		}
+
+		r := routetable.Route{
+			Peer: address,
+			Time: time.Unix(columns.time.Int64, 0).UTC(),
+		}
+
+		err := json.Unmarshal([]byte(columns.asPath.String), &r.ASPath)
+		if err != nil {
+			return change, fmt.Errorf("failed to read an AS path of route %s: %w",
+				prefix, err)
+		}
+
+		change.Routes = append(change.Routes, r)
+	}
+
+	return change, nil
+}