diff --git a/docs/release-notes/release-notes-0.20.0.md b/docs/release-notes/release-notes-0.20.0.md index d5edcab17..496a20475 100644 --- a/docs/release-notes/release-notes-0.20.0.md +++ b/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 diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index 8dab86611..789a4c0fb 100644 --- a/graph/db/graph_test.go +++ b/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 { - t.Fatal(err) - } - compareEdgePolicies(chanExp.Policy2, chanRet.Policy2) - if err != nil { - t.Fatal(err) - } + require.NoError(t, err) + + err = compareEdgePolicies( + chanExp.Policy2, chanRet.Policy2, + ) + 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{ - BaseFee: 10, - FeeRate: 20, - } - require.Equal(t, expectedInbound, directedChan.InboundFee) + require.Equal(t, inboundFee, 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 diff --git a/graph/db/notifications.go b/graph/db/notifications.go index 6c0acba88..4af953381 100644 --- a/graph/db/notifications.go +++ b/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", diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 137c1d2e2..22512619c 100644 --- a/graph/db/sql_store.go +++ b/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 err != nil { - return nil, fmt.Errorf("unable to decode color: %w", err) + if dbNode.Color.Valid { + node.Color, err = DecodeHexColor(dbNode.Color.String) + 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, diff --git a/graph/notifications_test.go b/graph/notifications_test.go index 099c2244b..c9096efc1 100644 --- a/graph/notifications_test.go +++ b/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 { - encoded := graphdb.EncodeHexColor( - color.RGBA{tc.R, tc.G, tc.B, 0}, + for _, test := range tests { + t.Run(fmt.Sprintf("R:%d,G:%d,B:%d", test.R, test.G, test.B), + 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) - } } } diff --git a/sqldb/sqlc/graph.sql.go b/sqldb/sqlc/graph.sql.go index 884d5c340..606d40217 100644 --- a/sqldb/sqlc/graph.sql.go +++ b/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 diff --git a/sqldb/sqlc/querier.go b/sqldb/sqlc/querier.go index d4a6a9dd0..c0abde3c2 100644 --- a/sqldb/sqlc/querier.go +++ b/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 diff --git a/sqldb/sqlc/queries/graph.sql b/sqldb/sqlc/queries/graph.sql index c6dbb6e6b..510195906 100644 --- a/sqldb/sqlc/queries/graph.sql +++ b/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,