routewatch/internal/database/routes.go

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
}