Merge pull request #9935 from ellemouton/graphSQL11-forEachMethods

[11] graph/db: Implement various "ForEach" methods on the graph SQLStore

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

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

graph/db/graph_test.go

@@ -23,6 +23,7 @@ import (
 	"github.com/btcsuite/btcd/chaincfg"
 	"github.com/btcsuite/btcd/chaincfg/chainhash"
 	"github.com/btcsuite/btcd/wire"
+	"github.com/lightningnetwork/lnd/fn/v2"
 	"github.com/lightningnetwork/lnd/graph/db/models"
 	"github.com/lightningnetwork/lnd/kvdb"
 	"github.com/lightningnetwork/lnd/lnwire"
@@ -1366,7 +1367,7 @@ func TestGraphTraversal(t *testing.T) {
 func TestGraphTraversalCacheable(t *testing.T) {
 	t.Parallel()
 
-	graph := MakeTestGraph(t)
+	graph := MakeTestGraphNew(t)
 
 	// We'd like to test some of the graph traversal capabilities within
 	// the DB, so we'll create a series of fake nodes to insert into the
@@ -1436,7 +1437,7 @@ func TestGraphTraversalCacheable(t *testing.T) {
 func TestGraphCacheTraversal(t *testing.T) {
 	t.Parallel()
 
-	graph := MakeTestGraph(t)
+	graph := MakeTestGraphNew(t)
 
 	// We'd like to test some of the graph traversal capabilities within
 	// the DB, so we'll create a series of fake nodes to insert into the
@@ -2069,16 +2070,15 @@ func TestChanUpdatesInHorizon(t *testing.T) {
 
 			assertEdgeInfoEqual(t, chanExp.Info, chanRet.Info)
 
-			err := compareEdgePolicies(
+			err = compareEdgePolicies(
 				chanExp.Policy1, chanRet.Policy1,
 			)
-			if err != nil {
+			require.NoError(t, err)
-				t.Fatal(err)
+
-			}
+			err = compareEdgePolicies(
-			compareEdgePolicies(chanExp.Policy2, chanRet.Policy2)
+				chanExp.Policy2, chanRet.Policy2,
-			if err != nil {
+			)
-				t.Fatal(err)
+			require.NoError(t, err)
-			}
 		}
 	}
 }
@@ -3041,7 +3041,7 @@ func TestIncompleteChannelPolicies(t *testing.T) {
 	t.Parallel()
 	ctx := context.Background()
 
-	graph := MakeTestGraph(t)
+	graph := MakeTestGraphNew(t)
 
 	// Create two nodes.
 	node1 := createTestVertex(t)
@@ -4110,7 +4110,7 @@ func TestBatchedUpdateEdgePolicy(t *testing.T) {
 // BenchmarkForEachChannel is a benchmark test that measures the number of
 // allocations and the total memory consumed by the full graph traversal.
 func BenchmarkForEachChannel(b *testing.B) {
-	graph := MakeTestGraph(b)
+	graph := MakeTestGraphNew(b)
 
 	const numNodes = 100
 	const numChannels = 4
@@ -4163,7 +4163,7 @@ func TestGraphCacheForEachNodeChannel(t *testing.T) {
 	t.Parallel()
 	ctx := context.Background()
 
-	graph := MakeTestGraph(t)
+	graph := MakeTestGraphNew(t)
 
 	// Unset the channel graph cache to simulate the user running with the
 	// option turned off.
@@ -4213,21 +4213,25 @@ func TestGraphCacheForEachNodeChannel(t *testing.T) {
 	edge1.ExtraOpaqueData = []byte{
 		253, 217, 3, 8, 0, 0, 0, 10, 0, 0, 0, 20,
 	}
+	inboundFee := lnwire.Fee{
+		BaseFee: 10,
+		FeeRate: 20,
+	}
+	edge1.InboundFee = fn.Some(inboundFee)
 	require.NoError(t, graph.UpdateEdgePolicy(edge1))
 	edge1 = copyEdgePolicy(edge1) // Avoid read/write race conditions.
 
 	directedChan := getSingleChannel()
 	require.NotNil(t, directedChan)
-	expectedInbound := lnwire.Fee{
+	require.Equal(t, inboundFee, directedChan.InboundFee)
-		BaseFee: 10,
-		FeeRate: 20,
-	}
-	require.Equal(t, expectedInbound, directedChan.InboundFee)
 
 	// Set an invalid inbound fee and check that persistence fails.
 	edge1.ExtraOpaqueData = []byte{
 		253, 217, 3, 8, 0,
 	}
+	// We need to update the timestamp so that we don't hit the DB conflict
+	// error when we try to update the edge policy.
+	edge1.LastUpdate = edge1.LastUpdate.Add(time.Second)
 	require.ErrorIs(
 		t, graph.UpdateEdgePolicy(edge1), ErrParsingExtraTLVBytes,
 	)
@@ -4236,7 +4240,7 @@ func TestGraphCacheForEachNodeChannel(t *testing.T) {
 	// the previous result when we query the channel again.
 	directedChan = getSingleChannel()
 	require.NotNil(t, directedChan)
-	require.Equal(t, expectedInbound, directedChan.InboundFee)
+	require.Equal(t, inboundFee, directedChan.InboundFee)
 }
 
 // TestGraphLoading asserts that the cache is properly reconstructed after a

graph/db/notifications.go

@@ -473,6 +473,11 @@ func EncodeHexColor(color color.RGBA) string {
 // DecodeHexColor takes a hex color string like "#rrggbb" and returns a
 // color.RGBA.
 func DecodeHexColor(hex string) (color.RGBA, error) {
+	if len(hex) != 7 || hex[0] != '#' {
+		return color.RGBA{}, fmt.Errorf("invalid hex color string: %s",
+			hex)
+	}
+
 	r, err := strconv.ParseUint(hex[1:3], 16, 8)
 	if err != nil {
 		return color.RGBA{}, fmt.Errorf("invalid red component: %w",

graph/db/sql_store.go

@@ -28,6 +28,10 @@ import (
 	"github.com/lightningnetwork/lnd/tor"
 )
 
+// pageSize is the limit for the number of records that can be returned
+// in a paginated query. This can be tuned after some benchmarks.
+const pageSize = 2000
+
 // ProtocolVersion is an enum that defines the gossip protocol version of a
 // message.
 type ProtocolVersion uint8
@@ -52,7 +56,10 @@ type SQLQueries interface {
 	*/
 	UpsertNode(ctx context.Context, arg sqlc.UpsertNodeParams) (int64, error)
 	GetNodeByPubKey(ctx context.Context, arg sqlc.GetNodeByPubKeyParams) (sqlc.Node, error)
+	GetNodeIDByPubKey(ctx context.Context, arg sqlc.GetNodeIDByPubKeyParams) (int64, error)
 	GetNodesByLastUpdateRange(ctx context.Context, arg sqlc.GetNodesByLastUpdateRangeParams) ([]sqlc.Node, error)
+	ListNodesPaginated(ctx context.Context, arg sqlc.ListNodesPaginatedParams) ([]sqlc.Node, error)
+	ListNodeIDsAndPubKeys(ctx context.Context, arg sqlc.ListNodeIDsAndPubKeysParams) ([]sqlc.ListNodeIDsAndPubKeysRow, error)
 	DeleteNodeByPubKey(ctx context.Context, arg sqlc.DeleteNodeByPubKeyParams) (sql.Result, error)
 
 	GetExtraNodeTypes(ctx context.Context, nodeID int64) ([]sqlc.NodeExtraType, error)
@@ -715,6 +722,383 @@ func (s *SQLStore) ForEachSourceNodeChannel(cb func(chanPoint wire.OutPoint,
 	}, sqldb.NoOpReset)
 }
 
+// ForEachNode iterates through all the stored vertices/nodes in the graph,
+// executing the passed callback with each node encountered. If the callback
+// returns an error, then the transaction is aborted and the iteration stops
+// early. Any operations performed on the NodeTx passed to the call-back are
+// executed under the same read transaction and so, methods on the NodeTx object
+// _MUST_ only be called from within the call-back.
+//
+// NOTE: part of the V1Store interface.
+func (s *SQLStore) ForEachNode(cb func(tx NodeRTx) error) error {
+	var (
+		ctx          = context.TODO()
+		lastID int64 = 0
+	)
+
+	handleNode := func(db SQLQueries, dbNode sqlc.Node) error {
+		node, err := buildNode(ctx, db, &dbNode)
+		if err != nil {
+			return fmt.Errorf("unable to build node(id=%d): %w",
+				dbNode.ID, err)
+		}
+
+		err = cb(
+			newSQLGraphNodeTx(db, s.cfg.ChainHash, dbNode.ID, node),
+		)
+		if err != nil {
+			return fmt.Errorf("callback failed for node(id=%d): %w",
+				dbNode.ID, err)
+		}
+
+		return nil
+	}
+
+	return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
+		for {
+			nodes, err := db.ListNodesPaginated(
+				ctx, sqlc.ListNodesPaginatedParams{
+					Version: int16(ProtocolV1),
+					ID:      lastID,
+					Limit:   pageSize,
+				},
+			)
+			if err != nil {
+				return fmt.Errorf("unable to fetch nodes: %w",
+					err)
+			}
+
+			if len(nodes) == 0 {
+				break
+			}
+
+			for _, dbNode := range nodes {
+				err = handleNode(db, dbNode)
+				if err != nil {
+					return err
+				}
+
+				lastID = dbNode.ID
+			}
+		}
+
+		return nil
+	}, sqldb.NoOpReset)
+}
+
+// sqlGraphNodeTx is an implementation of the NodeRTx interface backed by the
+// SQLStore and a SQL transaction.
+type sqlGraphNodeTx struct {
+	db    SQLQueries
+	id    int64
+	node  *models.LightningNode
+	chain chainhash.Hash
+}
+
+// A compile-time constraint to ensure sqlGraphNodeTx implements the NodeRTx
+// interface.
+var _ NodeRTx = (*sqlGraphNodeTx)(nil)
+
+func newSQLGraphNodeTx(db SQLQueries, chain chainhash.Hash,
+	id int64, node *models.LightningNode) *sqlGraphNodeTx {
+
+	return &sqlGraphNodeTx{
+		db:    db,
+		chain: chain,
+		id:    id,
+		node:  node,
+	}
+}
+
+// Node returns the raw information of the node.
+//
+// NOTE: This is a part of the NodeRTx interface.
+func (s *sqlGraphNodeTx) Node() *models.LightningNode {
+	return s.node
+}
+
+// ForEachChannel can be used to iterate over the node's channels under the same
+// transaction used to fetch the node.
+//
+// NOTE: This is a part of the NodeRTx interface.
+func (s *sqlGraphNodeTx) ForEachChannel(cb func(*models.ChannelEdgeInfo,
+	*models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error) error {
+
+	ctx := context.TODO()
+
+	return forEachNodeChannel(ctx, s.db, s.chain, s.id, cb)
+}
+
+// FetchNode fetches the node with the given pub key under the same transaction
+// used to fetch the current node. The returned node is also a NodeRTx and any
+// operations on that NodeRTx will also be done under the same transaction.
+//
+// NOTE: This is a part of the NodeRTx interface.
+func (s *sqlGraphNodeTx) FetchNode(nodePub route.Vertex) (NodeRTx, error) {
+	ctx := context.TODO()
+
+	id, node, err := getNodeByPubKey(ctx, s.db, nodePub)
+	if err != nil {
+		return nil, fmt.Errorf("unable to fetch V1 node(%x): %w",
+			nodePub, err)
+	}
+
+	return newSQLGraphNodeTx(s.db, s.chain, id, node), 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 callback returns an error, then the iteration
+// is halted with the error propagated back up to the caller.
+//
+// Unknown policies are passed into the callback as nil values.
+//
+// NOTE: this is part of the graphdb.NodeTraverser interface.
+func (s *SQLStore) ForEachNodeDirectedChannel(nodePub route.Vertex,
+	cb func(channel *DirectedChannel) error) error {
+
+	var ctx = context.TODO()
+
+	return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
+		return forEachNodeDirectedChannel(ctx, db, nodePub, cb)
+	}, sqldb.NoOpReset)
+}
+
+// ForEachNodeCacheable iterates through all the stored vertices/nodes in the
+// graph, executing the passed callback with each node encountered. If the
+// callback returns an error, then the transaction is aborted and the iteration
+// stops early.
+//
+// NOTE: This is a part of the V1Store interface.
+func (s *SQLStore) ForEachNodeCacheable(cb func(route.Vertex,
+	*lnwire.FeatureVector) error) error {
+
+	ctx := context.TODO()
+
+	err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
+		return forEachNodeCacheable(ctx, db, func(nodeID int64,
+			nodePub route.Vertex) error {
+
+			features, err := getNodeFeatures(ctx, db, nodeID)
+			if err != nil {
+				return fmt.Errorf("unable to fetch node "+
+					"features: %w", err)
+			}
+
+			return cb(nodePub, features)
+		})
+	}, sqldb.NoOpReset)
+	if err != nil {
+		return fmt.Errorf("unable to fetch nodes: %w", err)
+	}
+
+	return nil
+}
+
+// ForEachNodeChannel iterates through all channels of the given node,
+// executing the passed callback with an edge info structure and the policies
+// of each end of the channel. The first edge policy is the outgoing edge *to*
+// the connecting node, while the second is the incoming edge *from* the
+// connecting node. If the callback returns an error, then the iteration is
+// halted with the error propagated back up to the caller.
+//
+// Unknown policies are passed into the callback as nil values.
+//
+// NOTE: part of the V1Store interface.
+func (s *SQLStore) ForEachNodeChannel(nodePub route.Vertex,
+	cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
+		*models.ChannelEdgePolicy) error) error {
+
+	var ctx = context.TODO()
+
+	return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
+		dbNode, err := db.GetNodeByPubKey(
+			ctx, sqlc.GetNodeByPubKeyParams{
+				Version: int16(ProtocolV1),
+				PubKey:  nodePub[:],
+			},
+		)
+		if errors.Is(err, sql.ErrNoRows) {
+			return nil
+		} else if err != nil {
+			return fmt.Errorf("unable to fetch node: %w", err)
+		}
+
+		return forEachNodeChannel(
+			ctx, db, s.cfg.ChainHash, dbNode.ID, cb,
+		)
+	}, sqldb.NoOpReset)
+}
+
+// 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
+// returned.
+func forEachNodeDirectedChannel(ctx context.Context, db SQLQueries,
+	nodePub route.Vertex, cb func(channel *DirectedChannel) error) error {
+
+	toNodeCallback := func() route.Vertex {
+		return nodePub
+	}
+
+	dbID, err := db.GetNodeIDByPubKey(
+		ctx, sqlc.GetNodeIDByPubKeyParams{
+			Version: int16(ProtocolV1),
+			PubKey:  nodePub[:],
+		},
+	)
+	if errors.Is(err, sql.ErrNoRows) {
+		return nil
+	} else if err != nil {
+		return fmt.Errorf("unable to fetch node: %w", err)
+	}
+
+	rows, err := db.ListChannelsByNodeID(
+		ctx, sqlc.ListChannelsByNodeIDParams{
+			Version: int16(ProtocolV1),
+			NodeID1: dbID,
+		},
+	)
+	if err != nil {
+		return fmt.Errorf("unable to fetch channels: %w", err)
+	}
+
+	// Exit early if there are no channels for this node so we don't
+	// do the unnecessary feature fetching.
+	if len(rows) == 0 {
+		return nil
+	}
+
+	features, err := getNodeFeatures(ctx, db, dbID)
+	if err != nil {
+		return fmt.Errorf("unable to fetch node features: %w", err)
+	}
+
+	for _, row := range rows {
+		node1, node2, err := buildNodeVertices(
+			row.Node1Pubkey, row.Node2Pubkey,
+		)
+		if err != nil {
+			return fmt.Errorf("unable to build node vertices: %w",
+				err)
+		}
+
+		edge, err := buildCacheableChannelInfo(
+			row.Channel, node1, node2,
+		)
+		if err != nil {
+			return err
+		}
+
+		dbPol1, dbPol2, err := extractChannelPolicies(row)
+		if err != nil {
+			return err
+		}
+
+		var p1, p2 *models.CachedEdgePolicy
+		if dbPol1 != nil {
+			policy1, err := buildChanPolicy(
+				*dbPol1, edge.ChannelID, nil, node2, true,
+			)
+			if err != nil {
+				return err
+			}
+
+			p1 = models.NewCachedPolicy(policy1)
+		}
+		if dbPol2 != nil {
+			policy2, err := buildChanPolicy(
+				*dbPol2, edge.ChannelID, nil, node1, false,
+			)
+			if err != nil {
+				return err
+			}
+
+			p2 = models.NewCachedPolicy(policy2)
+		}
+
+		// Determine the outgoing and incoming policy for this
+		// channel and node combo.
+		outPolicy, inPolicy := p1, p2
+		if p1 != nil && node2 == nodePub {
+			outPolicy, inPolicy = p2, p1
+		} else if p2 != nil && node1 != nodePub {
+			outPolicy, inPolicy = p2, p1
+		}
+
+		var cachedInPolicy *models.CachedEdgePolicy
+		if inPolicy != nil {
+			cachedInPolicy = inPolicy
+			cachedInPolicy.ToNodePubKey = toNodeCallback
+			cachedInPolicy.ToNodeFeatures = features
+		}
+
+		directedChannel := &DirectedChannel{
+			ChannelID:    edge.ChannelID,
+			IsNode1:      nodePub == edge.NodeKey1Bytes,
+			OtherNode:    edge.NodeKey2Bytes,
+			Capacity:     edge.Capacity,
+			OutPolicySet: outPolicy != nil,
+			InPolicy:     cachedInPolicy,
+		}
+		if outPolicy != nil {
+			outPolicy.InboundFee.WhenSome(func(fee lnwire.Fee) {
+				directedChannel.InboundFee = fee
+			})
+		}
+
+		if nodePub == edge.NodeKey2Bytes {
+			directedChannel.OtherNode = edge.NodeKey1Bytes
+		}
+
+		if err := cb(directedChannel); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+// forEachNodeCacheable fetches all V1 node IDs and pub keys from the database,
+// and executes the provided callback for each node.
+func forEachNodeCacheable(ctx context.Context, db SQLQueries,
+	cb func(nodeID int64, nodePub route.Vertex) error) error {
+
+	var lastID int64
+
+	for {
+		nodes, err := db.ListNodeIDsAndPubKeys(
+			ctx, sqlc.ListNodeIDsAndPubKeysParams{
+				Version: int16(ProtocolV1),
+				ID:      lastID,
+				Limit:   pageSize,
+			},
+		)
+		if err != nil {
+			return fmt.Errorf("unable to fetch nodes: %w", err)
+		}
+
+		if len(nodes) == 0 {
+			break
+		}
+
+		for _, node := range nodes {
+			var pub route.Vertex
+			copy(pub[:], node.PubKey)
+
+			if err := cb(node.ID, pub); err != nil {
+				return fmt.Errorf("forEachNodeCacheable "+
+					"callback failed for node(id=%d): %w",
+					node.ID, err)
+			}
+
+			lastID = node.ID
+		}
+	}
+
+	return nil
+}
+
 // forEachNodeChannel iterates through all channels of a node, executing
 // the passed callback on each. The call-back is provided with the channel's
 // edge information, the outgoing policy and the incoming policy for the
@@ -904,6 +1288,20 @@ func getNodeByPubKey(ctx context.Context, db SQLQueries,
 	return dbNode.ID, node, nil
 }
 
+// buildCacheableChannelInfo builds a models.CachedEdgeInfo instance from the
+// provided database channel row and the public keys of the two nodes
+// involved in the channel.
+func buildCacheableChannelInfo(dbChan sqlc.Channel, node1Pub,
+	node2Pub route.Vertex) (*models.CachedEdgeInfo, error) {
+
+	return &models.CachedEdgeInfo{
+		ChannelID:     byteOrder.Uint64(dbChan.Scid),
+		NodeKey1Bytes: node1Pub,
+		NodeKey2Bytes: node2Pub,
+		Capacity:      btcutil.Amount(dbChan.Capacity.Int64),
+	}, nil
+}
+
 // buildNode constructs a LightningNode instance from the given database node
 // record. The node's features, addresses and extra signed fields are also
 // fetched from the database and set on the node.
@@ -934,9 +1332,12 @@ func buildNode(ctx context.Context, db SQLQueries, dbNode *sqlc.Node) (
 	node.LastUpdate = time.Unix(dbNode.LastUpdate.Int64, 0)
 
 	var err error
-	node.Color, err = DecodeHexColor(dbNode.Color.String)
+	if dbNode.Color.Valid {
-	if err != nil {
+		node.Color, err = DecodeHexColor(dbNode.Color.String)
-		return nil, fmt.Errorf("unable to decode color: %w", err)
+		if err != nil {
+			return nil, fmt.Errorf("unable to decode color: %w",
+				err)
+		}
 	}
 
 	// Fetch the node's features.
@@ -1457,7 +1858,7 @@ func marshalExtraOpaqueData(data []byte) (map[uint64][]byte, error) {
 	// pass it into the P2P decoding variant.
 	parsedTypes, err := tlvStream.DecodeWithParsedTypesP2P(r)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("%w: %w", ErrParsingExtraTLVBytes, err)
 	}
 	if len(parsedTypes) == 0 {
 		return nil, nil
@@ -1702,7 +2103,10 @@ func getAndBuildEdgeInfo(ctx context.Context, db SQLQueries,
 		ExtraOpaqueData:  recs,
 	}
 
-	if dbChan.Bitcoin1Signature != nil {
+	// We always set all the signatures at the same time, so we can
+	// safely check if one signature is present to determine if we have the
+	// rest of the signatures for the auth proof.
+	if len(dbChan.Bitcoin1Signature) > 0 {
 		channel.AuthProof = &models.ChannelAuthProof{
 			NodeSig1Bytes:    dbChan.Node1Signature,
 			NodeSig2Bytes:    dbChan.Node2Signature,

graph/notifications_test.go

@@ -991,7 +991,9 @@ func TestChannelCloseNotification(t *testing.T) {
 // TestEncodeHexColor tests that the string used to represent a node color is
 // correctly encoded.
 func TestEncodeHexColor(t *testing.T) {
-	var colorTestCases = []struct {
+	t.Parallel()
+
+	var tests = []struct {
 		R       uint8
 		G       uint8
 		B       uint8
@@ -1006,14 +1008,32 @@ func TestEncodeHexColor(t *testing.T) {
 		{1, 2, 3, "#", false},
 	}
 
-	for _, tc := range colorTestCases {
+	for _, test := range tests {
-		encoded := graphdb.EncodeHexColor(
+		t.Run(fmt.Sprintf("R:%d,G:%d,B:%d", test.R, test.G, test.B),
-			color.RGBA{tc.R, tc.G, tc.B, 0},
+			func(t *testing.T) {
+				expColor := color.RGBA{
+					R: test.R,
+					G: test.G,
+					B: test.B,
+				}
+
+				encoded := graphdb.EncodeHexColor(expColor)
+
+				if test.isValid {
+					require.Equal(t, test.encoded, encoded)
+				}
+
+				decoded, err := graphdb.DecodeHexColor(
+					test.encoded,
+				)
+				if test.isValid {
+					require.NoError(t, err)
+					require.Equal(t, expColor, decoded)
+				} else {
+					require.Error(t, err)
+				}
+			},
 		)
-		if (encoded == tc.encoded) != tc.isValid {
-			t.Fatalf("incorrect color encoding, "+
-				"want: %v, got: %v", tc.encoded, encoded)
-		}
 	}
 }
 

sqldb/sqlc/graph.sql.go

@@ -533,6 +533,25 @@ func (q *Queries) GetNodeFeaturesByPubKey(ctx context.Context, arg GetNodeFeatur
 	return items, nil
 }
 
+const getNodeIDByPubKey = `-- name: GetNodeIDByPubKey :one
+SELECT id
+FROM nodes
+WHERE pub_key = $1
+  AND version = $2
+`
+
+type GetNodeIDByPubKeyParams struct {
+	PubKey  []byte
+	Version int16
+}
+
+func (q *Queries) GetNodeIDByPubKey(ctx context.Context, arg GetNodeIDByPubKeyParams) (int64, error) {
+	row := q.db.QueryRowContext(ctx, getNodeIDByPubKey, arg.PubKey, arg.Version)
+	var id int64
+	err := row.Scan(&id)
+	return id, err
+}
+
 const getNodesByLastUpdateRange = `-- name: GetNodesByLastUpdateRange :many
 SELECT id, version, pub_key, alias, last_update, color, signature
 FROM nodes
@@ -729,7 +748,6 @@ func (q *Queries) InsertNodeFeature(ctx context.Context, arg InsertNodeFeaturePa
 }
 
 const listChannelsByNodeID = `-- name: ListChannelsByNodeID :many
-
 SELECT c.id, c.version, c.scid, c.node_id_1, c.node_id_2, c.outpoint, c.capacity, c.bitcoin_key_1, c.bitcoin_key_2, c.node_1_signature, c.node_2_signature, c.bitcoin_1_signature, c.bitcoin_2_signature,
     n1.pub_key AS node1_pubkey,
     n2.pub_key AS node2_pubkey,
@@ -880,6 +898,93 @@ func (q *Queries) ListChannelsByNodeID(ctx context.Context, arg ListChannelsByNo
 	return items, nil
 }
 
+const listNodeIDsAndPubKeys = `-- name: ListNodeIDsAndPubKeys :many
+SELECT id, pub_key
+FROM nodes
+WHERE version = $1  AND id > $2
+ORDER BY id
+LIMIT $3
+`
+
+type ListNodeIDsAndPubKeysParams struct {
+	Version int16
+	ID      int64
+	Limit   int32
+}
+
+type ListNodeIDsAndPubKeysRow struct {
+	ID     int64
+	PubKey []byte
+}
+
+func (q *Queries) ListNodeIDsAndPubKeys(ctx context.Context, arg ListNodeIDsAndPubKeysParams) ([]ListNodeIDsAndPubKeysRow, error) {
+	rows, err := q.db.QueryContext(ctx, listNodeIDsAndPubKeys, arg.Version, arg.ID, arg.Limit)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var items []ListNodeIDsAndPubKeysRow
+	for rows.Next() {
+		var i ListNodeIDsAndPubKeysRow
+		if err := rows.Scan(&i.ID, &i.PubKey); err != nil {
+			return nil, err
+		}
+		items = append(items, i)
+	}
+	if err := rows.Close(); err != nil {
+		return nil, err
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return items, nil
+}
+
+const listNodesPaginated = `-- name: ListNodesPaginated :many
+SELECT id, version, pub_key, alias, last_update, color, signature
+FROM nodes
+WHERE version = $1 AND id > $2
+ORDER BY id
+LIMIT $3
+`
+
+type ListNodesPaginatedParams struct {
+	Version int16
+	ID      int64
+	Limit   int32
+}
+
+func (q *Queries) ListNodesPaginated(ctx context.Context, arg ListNodesPaginatedParams) ([]Node, error) {
+	rows, err := q.db.QueryContext(ctx, listNodesPaginated, arg.Version, arg.ID, arg.Limit)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var items []Node
+	for rows.Next() {
+		var i Node
+		if err := rows.Scan(
+			&i.ID,
+			&i.Version,
+			&i.PubKey,
+			&i.Alias,
+			&i.LastUpdate,
+			&i.Color,
+			&i.Signature,
+		); err != nil {
+			return nil, err
+		}
+		items = append(items, i)
+	}
+	if err := rows.Close(); err != nil {
+		return nil, err
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return items, nil
+}
+
 const upsertEdgePolicy = `-- name: UpsertEdgePolicy :one
 /* ─────────────────────────────────────────────
    channel_policies table queries

sqldb/sqlc/querier.go

@@ -48,6 +48,7 @@ type Querier interface {
 	GetNodeByPubKey(ctx context.Context, arg GetNodeByPubKeyParams) (Node, error)
 	GetNodeFeatures(ctx context.Context, nodeID int64) ([]NodeFeature, error)
 	GetNodeFeaturesByPubKey(ctx context.Context, arg GetNodeFeaturesByPubKeyParams) ([]int32, error)
+	GetNodeIDByPubKey(ctx context.Context, arg GetNodeIDByPubKeyParams) (int64, error)
 	GetNodesByLastUpdateRange(ctx context.Context, arg GetNodesByLastUpdateRangeParams) ([]Node, error)
 	GetSourceNodesByVersion(ctx context.Context, version int16) ([]GetSourceNodesByVersionRow, error)
 	HighestSCID(ctx context.Context, version int16) ([]byte, error)
@@ -64,6 +65,8 @@ type Querier interface {
 	InsertNodeAddress(ctx context.Context, arg InsertNodeAddressParams) error
 	InsertNodeFeature(ctx context.Context, arg InsertNodeFeatureParams) error
 	ListChannelsByNodeID(ctx context.Context, arg ListChannelsByNodeIDParams) ([]ListChannelsByNodeIDRow, error)
+	ListNodeIDsAndPubKeys(ctx context.Context, arg ListNodeIDsAndPubKeysParams) ([]ListNodeIDsAndPubKeysRow, error)
+	ListNodesPaginated(ctx context.Context, arg ListNodesPaginatedParams) ([]Node, error)
 	NextInvoiceSettleIndex(ctx context.Context) (int64, error)
 	OnAMPSubInvoiceCanceled(ctx context.Context, arg OnAMPSubInvoiceCanceledParams) error
 	OnAMPSubInvoiceCreated(ctx context.Context, arg OnAMPSubInvoiceCreatedParams) error

sqldb/sqlc/queries/graph.sql

@@ -27,6 +27,26 @@ FROM nodes
 WHERE pub_key = $1
   AND version = $2;
 
+-- name: GetNodeIDByPubKey :one
+SELECT id
+FROM nodes
+WHERE pub_key = $1
+  AND version = $2;
+
+-- name: ListNodesPaginated :many
+SELECT *
+FROM nodes
+WHERE version = $1 AND id > $2
+ORDER BY id
+LIMIT $3;
+
+-- name: ListNodeIDsAndPubKeys :many
+SELECT id, pub_key
+FROM nodes
+WHERE version = $1  AND id > $2
+ORDER BY id
+LIMIT $3;
+
 -- name: DeleteNodeByPubKey :execresult
 DELETE FROM nodes
 WHERE pub_key = $1
@@ -194,7 +214,6 @@ ORDER BY scid DESC
 LIMIT 1;
 
 -- name: ListChannelsByNodeID :many
-
 SELECT sqlc.embed(c),
     n1.pub_key AS node1_pubkey,
     n2.pub_key AS node2_pubkey,