sqldb+graph/db: implement FilterChannelRange

This lets us run `TestFilterChannelRange` against the SQL backends.

Author: ellemouton

docs/release-notes/release-notes-0.20.0.md

@@ -85,6 +85,7 @@ circuit. The indices are only available for forwarding events saved after v0.20.
     * [3](https://github.com/lightningnetwork/lnd/pull/9887)
     * [4](https://github.com/lightningnetwork/lnd/pull/9931)
     * [5](https://github.com/lightningnetwork/lnd/pull/9935)
+    * [6](https://github.com/lightningnetwork/lnd/pull/9936)
 
 ## RPC Updates
 

graph/db/graph_test.go

@@ -2727,7 +2727,7 @@ func TestFilterChannelRange(t *testing.T) {
 	t.Parallel()
 	ctx := context.Background()
 
-	graph := MakeTestGraph(t)
+	graph := MakeTestGraphNew(t)
 
 	// We'll first populate our graph with two nodes. All channels created
 	// below will be made between these two nodes.

graph/db/sql_store.go

@@ -9,6 +9,7 @@ import (
 	"fmt"
 	"math"
 	"net"
+	"sort"
 	"strconv"
 	"sync"
 	"time"
@@ -92,6 +93,7 @@ type SQLQueries interface {
 	ListChannelsByNodeID(ctx context.Context, arg sqlc.ListChannelsByNodeIDParams) ([]sqlc.ListChannelsByNodeIDRow, error)
 	ListChannelsPaginated(ctx context.Context, arg sqlc.ListChannelsPaginatedParams) ([]sqlc.ListChannelsPaginatedRow, error)
 	GetChannelsByPolicyLastUpdateRange(ctx context.Context, arg sqlc.GetChannelsByPolicyLastUpdateRangeParams) ([]sqlc.GetChannelsByPolicyLastUpdateRangeRow, error)
+	GetPublicV1ChannelsBySCID(ctx context.Context, arg sqlc.GetPublicV1ChannelsBySCIDParams) ([]sqlc.Channel, error)
 
 	CreateChannelExtraType(ctx context.Context, arg sqlc.CreateChannelExtraTypeParams) error
 	InsertChannelFeature(ctx context.Context, arg sqlc.InsertChannelFeatureParams) error
@@ -100,6 +102,7 @@ type SQLQueries interface {
 		Channel Policy table queries.
 	*/
 	UpsertEdgePolicy(ctx context.Context, arg sqlc.UpsertEdgePolicyParams) (int64, error)
+	GetChannelPolicyByChannelAndNode(ctx context.Context, arg sqlc.GetChannelPolicyByChannelAndNodeParams) (sqlc.ChannelPolicy, error)
 
 	InsertChanPolicyExtraType(ctx context.Context, arg sqlc.InsertChanPolicyExtraTypeParams) error
 	GetChannelPolicyExtraTypes(ctx context.Context, arg sqlc.GetChannelPolicyExtraTypesParams) ([]sqlc.GetChannelPolicyExtraTypesRow, error)
@@ -1258,6 +1261,141 @@ func (s *SQLStore) ForEachChannel(cb func(*models.ChannelEdgeInfo,
 	}, sqldb.NoOpReset)
 }
 
+// FilterChannelRange returns the channel ID's of all known channels which were
+// mined in a block height within the passed range. The channel IDs are grouped
+// by their common block height. This method can be used to quickly share with a
+// peer the set of channels we know of within a particular range to catch them
+// up after a period of time offline. If withTimestamps is true then the
+// timestamp info of the latest received channel update messages of the channel
+// will be included in the response.
+//
+// NOTE: This is part of the V1Store interface.
+func (s *SQLStore) FilterChannelRange(startHeight, endHeight uint32,
+	withTimestamps bool) ([]BlockChannelRange, error) {
+
+	var (
+		ctx       = context.TODO()
+		startSCID = &lnwire.ShortChannelID{
+			BlockHeight: startHeight,
+		}
+		endSCID = lnwire.ShortChannelID{
+			BlockHeight: endHeight,
+			TxIndex:     math.MaxUint32 & 0x00ffffff,
+			TxPosition:  math.MaxUint16,
+		}
+	)
+
+	var chanIDStart [8]byte
+	byteOrder.PutUint64(chanIDStart[:], startSCID.ToUint64())
+	var chanIDEnd [8]byte
+	byteOrder.PutUint64(chanIDEnd[:], endSCID.ToUint64())
+
+	// 1) get all channels where channelID is between start and end chan ID.
+	// 2) skip if not public (ie, no channel_proof)
+	// 3) collect that channel.
+	// 4) if timestamps are wanted, fetch both policies for node 1 and node2
+	//    and add those timestamps to the collected channel.
+	channelsPerBlock := make(map[uint32][]ChannelUpdateInfo)
+	err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
+		dbChans, err := db.GetPublicV1ChannelsBySCID(
+			ctx, sqlc.GetPublicV1ChannelsBySCIDParams{
+				StartScid: chanIDStart[:],
+				EndScid:   chanIDEnd[:],
+			},
+		)
+		if err != nil {
+			return fmt.Errorf("unable to fetch channel range: %w",
+				err)
+		}
+
+		for _, dbChan := range dbChans {
+			cid := lnwire.NewShortChanIDFromInt(
+				byteOrder.Uint64(dbChan.Scid),
+			)
+			chanInfo := NewChannelUpdateInfo(
+				cid, time.Time{}, time.Time{},
+			)
+
+			if !withTimestamps {
+				channelsPerBlock[cid.BlockHeight] = append(
+					channelsPerBlock[cid.BlockHeight],
+					chanInfo,
+				)
+
+				continue
+			}
+
+			//nolint:ll
+			node1Policy, err := db.GetChannelPolicyByChannelAndNode(
+				ctx, sqlc.GetChannelPolicyByChannelAndNodeParams{
+					Version:   int16(ProtocolV1),
+					ChannelID: dbChan.ID,
+					NodeID:    dbChan.NodeID1,
+				},
+			)
+			if err != nil && !errors.Is(err, sql.ErrNoRows) {
+				return fmt.Errorf("unable to fetch node1 "+
+					"policy: %w", err)
+			} else if err == nil {
+				chanInfo.Node1UpdateTimestamp = time.Unix(
+					node1Policy.LastUpdate.Int64, 0,
+				)
+			}
+
+			//nolint:ll
+			node2Policy, err := db.GetChannelPolicyByChannelAndNode(
+				ctx, sqlc.GetChannelPolicyByChannelAndNodeParams{
+					Version:   int16(ProtocolV1),
+					ChannelID: dbChan.ID,
+					NodeID:    dbChan.NodeID2,
+				},
+			)
+			if err != nil && !errors.Is(err, sql.ErrNoRows) {
+				return fmt.Errorf("unable to fetch node2 "+
+					"policy: %w", err)
+			} else if err == nil {
+				chanInfo.Node2UpdateTimestamp = time.Unix(
+					node2Policy.LastUpdate.Int64, 0,
+				)
+			}
+
+			channelsPerBlock[cid.BlockHeight] = append(
+				channelsPerBlock[cid.BlockHeight], chanInfo,
+			)
+		}
+
+		return nil
+	}, func() {
+		channelsPerBlock = make(map[uint32][]ChannelUpdateInfo)
+	})
+	if err != nil {
+		return nil, fmt.Errorf("unable to fetch channel range: %w", err)
+	}
+
+	if len(channelsPerBlock) == 0 {
+		return nil, nil
+	}
+
+	// Return the channel ranges in ascending block height order.
+	blocks := make([]uint32, 0, len(channelsPerBlock))
+	for block := range channelsPerBlock {
+		blocks = append(blocks, block)
+	}
+	sort.Slice(blocks, func(i, j int) bool {
+		return blocks[i] < blocks[j]
+	})
+
+	channelRanges := make([]BlockChannelRange, 0, len(channelsPerBlock))
+	for _, block := range blocks {
+		channelRanges = append(channelRanges, BlockChannelRange{
+			Height:   block,
+			Channels: channelsPerBlock[block],
+		})
+	}
+
+	return channelRanges, nil
+}
+
 // forEachNodeDirectedChannel iterates through all channels of a given
 // node, executing the passed callback on the directed edge representing the
 // channel and its incoming policy. If the node is not found, no error is

sqldb/sqlc/graph.sql.go

@@ -314,6 +314,13 @@ FROM channels c
 WHERE c.version = $1
   AND (c.node_id_1 = $2 OR c.node_id_2 = $2);
 
+-- name: GetPublicV1ChannelsBySCID :many
+SELECT *
+FROM channels
+WHERE node_1_signature IS NOT NULL
+  AND scid >= sqlc.arg(start_scid)
+  AND scid < sqlc.arg(end_scid);
+
 -- name: ListChannelsPaginated :many
 SELECT
     sqlc.embed(c),
@@ -417,6 +424,13 @@ ON CONFLICT (channel_id, node_id, version)
 WHERE EXCLUDED.last_update > channel_policies.last_update
 RETURNING id;
 
+-- name: GetChannelPolicyByChannelAndNode :one
+SELECT *
+FROM channel_policies
+WHERE channel_id = $1
+  AND node_id = $2
+  AND version = $3;
+
 /* ─────────────────────────────────────────────
    channel_policy_extra_types table queries
    ─────────────────────────────────────────────