• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

lightningnetwork / lnd / 15710136066

17 Jun 2025 02:30PM UTC coverage: 68.465% (-0.04%) from 68.507%
15710136066

push

github

web-flow
Merge pull request #9887 from ellemouton/graphSQL9-chan-policies-schema

graph/db+sqldb: channel policy SQL schemas, queries and upsert CRUD

0 of 213 new or added lines in 2 files covered. (0.0%)

38 existing lines in 10 files now uncovered.

134571 of 196555 relevant lines covered (68.46%)

22324.56 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0
/graph/db/sql_store.go
1
package graphdb
2

3
import (
4
        "bytes"
5
        "context"
6
        "database/sql"
7
        "encoding/hex"
8
        "errors"
9
        "fmt"
10
        "math"
11
        "net"
12
        "strconv"
13
        "sync"
14
        "time"
15

16
        "github.com/btcsuite/btcd/btcec/v2"
17
        "github.com/btcsuite/btcd/chaincfg/chainhash"
18
        "github.com/lightningnetwork/lnd/batch"
19
        "github.com/lightningnetwork/lnd/graph/db/models"
20
        "github.com/lightningnetwork/lnd/lnwire"
21
        "github.com/lightningnetwork/lnd/routing/route"
22
        "github.com/lightningnetwork/lnd/sqldb"
23
        "github.com/lightningnetwork/lnd/sqldb/sqlc"
24
        "github.com/lightningnetwork/lnd/tlv"
25
        "github.com/lightningnetwork/lnd/tor"
26
)
27

28
// ProtocolVersion is an enum that defines the gossip protocol version of a
29
// message.
30
type ProtocolVersion uint8
31

32
const (
33
        // ProtocolV1 is the gossip protocol version defined in BOLT #7.
34
        ProtocolV1 ProtocolVersion = 1
35
)
36

37
// String returns a string representation of the protocol version.
38
func (v ProtocolVersion) String() string {
×
39
        return fmt.Sprintf("V%d", v)
×
40
}
×
41

42
// SQLQueries is a subset of the sqlc.Querier interface that can be used to
43
// execute queries against the SQL graph tables.
44
//
45
//nolint:ll,interfacebloat
46
type SQLQueries interface {
47
        /*
48
                Node queries.
49
        */
50
        UpsertNode(ctx context.Context, arg sqlc.UpsertNodeParams) (int64, error)
51
        GetNodeByPubKey(ctx context.Context, arg sqlc.GetNodeByPubKeyParams) (sqlc.Node, error)
52
        GetNodesByLastUpdateRange(ctx context.Context, arg sqlc.GetNodesByLastUpdateRangeParams) ([]sqlc.Node, error)
53
        DeleteNodeByPubKey(ctx context.Context, arg sqlc.DeleteNodeByPubKeyParams) (sql.Result, error)
54

55
        GetExtraNodeTypes(ctx context.Context, nodeID int64) ([]sqlc.NodeExtraType, error)
56
        UpsertNodeExtraType(ctx context.Context, arg sqlc.UpsertNodeExtraTypeParams) error
57
        DeleteExtraNodeType(ctx context.Context, arg sqlc.DeleteExtraNodeTypeParams) error
58

59
        InsertNodeAddress(ctx context.Context, arg sqlc.InsertNodeAddressParams) error
60
        GetNodeAddressesByPubKey(ctx context.Context, arg sqlc.GetNodeAddressesByPubKeyParams) ([]sqlc.GetNodeAddressesByPubKeyRow, error)
61
        DeleteNodeAddresses(ctx context.Context, nodeID int64) error
62

63
        InsertNodeFeature(ctx context.Context, arg sqlc.InsertNodeFeatureParams) error
64
        GetNodeFeatures(ctx context.Context, nodeID int64) ([]sqlc.NodeFeature, error)
65
        GetNodeFeaturesByPubKey(ctx context.Context, arg sqlc.GetNodeFeaturesByPubKeyParams) ([]int32, error)
66
        DeleteNodeFeature(ctx context.Context, arg sqlc.DeleteNodeFeatureParams) error
67

68
        /*
69
                Source node queries.
70
        */
71
        AddSourceNode(ctx context.Context, nodeID int64) error
72
        GetSourceNodesByVersion(ctx context.Context, version int16) ([]sqlc.GetSourceNodesByVersionRow, error)
73

74
        /*
75
                Channel queries.
76
        */
77
        CreateChannel(ctx context.Context, arg sqlc.CreateChannelParams) (int64, error)
78
        GetChannelBySCID(ctx context.Context, arg sqlc.GetChannelBySCIDParams) (sqlc.Channel, error)
79
        GetChannelAndNodesBySCID(ctx context.Context, arg sqlc.GetChannelAndNodesBySCIDParams) (sqlc.GetChannelAndNodesBySCIDRow, error)
80
        HighestSCID(ctx context.Context, version int16) ([]byte, error)
81

82
        CreateChannelExtraType(ctx context.Context, arg sqlc.CreateChannelExtraTypeParams) error
83
        InsertChannelFeature(ctx context.Context, arg sqlc.InsertChannelFeatureParams) error
84

85
        /*
86
                Channel Policy table queries.
87
        */
88
        UpsertEdgePolicy(ctx context.Context, arg sqlc.UpsertEdgePolicyParams) (int64, error)
89

90
        InsertChanPolicyExtraType(ctx context.Context, arg sqlc.InsertChanPolicyExtraTypeParams) error
91
        DeleteChannelPolicyExtraTypes(ctx context.Context, channelPolicyID int64) error
92
}
93

94
// BatchedSQLQueries is a version of SQLQueries that's capable of batched
95
// database operations.
96
type BatchedSQLQueries interface {
97
        SQLQueries
98
        sqldb.BatchedTx[SQLQueries]
99
}
100

101
// SQLStore is an implementation of the V1Store interface that uses a SQL
102
// database as the backend.
103
//
104
// NOTE: currently, this temporarily embeds the KVStore struct so that we can
105
// implement the V1Store interface incrementally. For any method not
106
// implemented,  things will fall back to the KVStore. This is ONLY the case
107
// for the time being while this struct is purely used in unit tests only.
108
type SQLStore struct {
109
        cfg *SQLStoreConfig
110
        db  BatchedSQLQueries
111

112
        // cacheMu guards all caches (rejectCache and chanCache). If
113
        // this mutex will be acquired at the same time as the DB mutex then
114
        // the cacheMu MUST be acquired first to prevent deadlock.
115
        cacheMu     sync.RWMutex
116
        rejectCache *rejectCache
117
        chanCache   *channelCache
118

119
        chanScheduler batch.Scheduler[SQLQueries]
120
        nodeScheduler batch.Scheduler[SQLQueries]
121

122
        // Temporary fall-back to the KVStore so that we can implement the
123
        // interface incrementally.
124
        *KVStore
125
}
126

127
// A compile-time assertion to ensure that SQLStore implements the V1Store
128
// interface.
129
var _ V1Store = (*SQLStore)(nil)
130

131
// SQLStoreConfig holds the configuration for the SQLStore.
132
type SQLStoreConfig struct {
133
        // ChainHash is the genesis hash for the chain that all the gossip
134
        // messages in this store are aimed at.
135
        ChainHash chainhash.Hash
136
}
137

138
// NewSQLStore creates a new SQLStore instance given an open BatchedSQLQueries
139
// storage backend.
140
func NewSQLStore(cfg *SQLStoreConfig, db BatchedSQLQueries, kvStore *KVStore,
141
        options ...StoreOptionModifier) (*SQLStore, error) {
×
142

×
143
        opts := DefaultOptions()
×
144
        for _, o := range options {
×
145
                o(opts)
×
146
        }
×
147

148
        if opts.NoMigration {
×
149
                return nil, fmt.Errorf("the NoMigration option is not yet " +
×
150
                        "supported for SQL stores")
×
151
        }
×
152

153
        s := &SQLStore{
×
NEW
154
                cfg:         cfg,
×
155
                db:          db,
×
156
                KVStore:     kvStore,
×
157
                rejectCache: newRejectCache(opts.RejectCacheSize),
×
158
                chanCache:   newChannelCache(opts.ChannelCacheSize),
×
159
        }
×
160

×
161
        s.chanScheduler = batch.NewTimeScheduler(
×
162
                db, &s.cacheMu, opts.BatchCommitInterval,
×
163
        )
×
164
        s.nodeScheduler = batch.NewTimeScheduler(
×
165
                db, nil, opts.BatchCommitInterval,
×
166
        )
×
167

×
168
        return s, nil
×
169
}
170

171
// AddLightningNode adds a vertex/node to the graph database. If the node is not
172
// in the database from before, this will add a new, unconnected one to the
173
// graph. If it is present from before, this will update that node's
174
// information.
175
//
176
// NOTE: part of the V1Store interface.
177
func (s *SQLStore) AddLightningNode(node *models.LightningNode,
178
        opts ...batch.SchedulerOption) error {
×
179

×
180
        ctx := context.TODO()
×
181

×
182
        r := &batch.Request[SQLQueries]{
×
183
                Opts: batch.NewSchedulerOptions(opts...),
×
184
                Do: func(queries SQLQueries) error {
×
185
                        _, err := upsertNode(ctx, queries, node)
×
186
                        return err
×
187
                },
×
188
        }
189

190
        return s.nodeScheduler.Execute(ctx, r)
×
191
}
192

193
// FetchLightningNode attempts to look up a target node by its identity public
194
// key. If the node isn't found in the database, then ErrGraphNodeNotFound is
195
// returned.
196
//
197
// NOTE: part of the V1Store interface.
198
func (s *SQLStore) FetchLightningNode(pubKey route.Vertex) (
199
        *models.LightningNode, error) {
×
200

×
201
        ctx := context.TODO()
×
202

×
203
        var node *models.LightningNode
×
204
        err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
×
205
                var err error
×
206
                _, node, err = getNodeByPubKey(ctx, db, pubKey)
×
207

×
208
                return err
×
209
        }, sqldb.NoOpReset)
×
210
        if err != nil {
×
211
                return nil, fmt.Errorf("unable to fetch node: %w", err)
×
212
        }
×
213

214
        return node, nil
×
215
}
216

217
// HasLightningNode determines if the graph has a vertex identified by the
218
// target node identity public key. If the node exists in the database, a
219
// timestamp of when the data for the node was lasted updated is returned along
220
// with a true boolean. Otherwise, an empty time.Time is returned with a false
221
// boolean.
222
//
223
// NOTE: part of the V1Store interface.
224
func (s *SQLStore) HasLightningNode(pubKey [33]byte) (time.Time, bool,
225
        error) {
×
226

×
227
        ctx := context.TODO()
×
228

×
229
        var (
×
230
                exists     bool
×
231
                lastUpdate time.Time
×
232
        )
×
233
        err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
×
234
                dbNode, err := db.GetNodeByPubKey(
×
235
                        ctx, sqlc.GetNodeByPubKeyParams{
×
236
                                Version: int16(ProtocolV1),
×
237
                                PubKey:  pubKey[:],
×
238
                        },
×
239
                )
×
240
                if errors.Is(err, sql.ErrNoRows) {
×
241
                        return nil
×
242
                } else if err != nil {
×
243
                        return fmt.Errorf("unable to fetch node: %w", err)
×
244
                }
×
245

246
                exists = true
×
247

×
248
                if dbNode.LastUpdate.Valid {
×
249
                        lastUpdate = time.Unix(dbNode.LastUpdate.Int64, 0)
×
250
                }
×
251

252
                return nil
×
253
        }, sqldb.NoOpReset)
254
        if err != nil {
×
255
                return time.Time{}, false,
×
256
                        fmt.Errorf("unable to fetch node: %w", err)
×
257
        }
×
258

259
        return lastUpdate, exists, nil
×
260
}
261

262
// AddrsForNode returns all known addresses for the target node public key
263
// that the graph DB is aware of. The returned boolean indicates if the
264
// given node is unknown to the graph DB or not.
265
//
266
// NOTE: part of the V1Store interface.
267
func (s *SQLStore) AddrsForNode(nodePub *btcec.PublicKey) (bool, []net.Addr,
268
        error) {
×
269

×
270
        ctx := context.TODO()
×
271

×
272
        var (
×
273
                addresses []net.Addr
×
274
                known     bool
×
275
        )
×
276
        err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
×
277
                var err error
×
278
                known, addresses, err = getNodeAddresses(
×
279
                        ctx, db, nodePub.SerializeCompressed(),
×
280
                )
×
281
                if err != nil {
×
282
                        return fmt.Errorf("unable to fetch node addresses: %w",
×
283
                                err)
×
284
                }
×
285

286
                return nil
×
287
        }, sqldb.NoOpReset)
288
        if err != nil {
×
289
                return false, nil, fmt.Errorf("unable to get addresses for "+
×
290
                        "node(%x): %w", nodePub.SerializeCompressed(), err)
×
291
        }
×
292

293
        return known, addresses, nil
×
294
}
295

296
// DeleteLightningNode starts a new database transaction to remove a vertex/node
297
// from the database according to the node's public key.
298
//
299
// NOTE: part of the V1Store interface.
300
func (s *SQLStore) DeleteLightningNode(pubKey route.Vertex) error {
×
301
        ctx := context.TODO()
×
302

×
303
        err := s.db.ExecTx(ctx, sqldb.WriteTxOpt(), func(db SQLQueries) error {
×
304
                res, err := db.DeleteNodeByPubKey(
×
305
                        ctx, sqlc.DeleteNodeByPubKeyParams{
×
306
                                Version: int16(ProtocolV1),
×
307
                                PubKey:  pubKey[:],
×
308
                        },
×
309
                )
×
310
                if err != nil {
×
311
                        return err
×
312
                }
×
313

314
                rows, err := res.RowsAffected()
×
315
                if err != nil {
×
316
                        return err
×
317
                }
×
318

319
                if rows == 0 {
×
320
                        return ErrGraphNodeNotFound
×
321
                } else if rows > 1 {
×
322
                        return fmt.Errorf("deleted %d rows, expected 1", rows)
×
323
                }
×
324

325
                return err
×
326
        }, sqldb.NoOpReset)
327
        if err != nil {
×
328
                return fmt.Errorf("unable to delete node: %w", err)
×
329
        }
×
330

331
        return nil
×
332
}
333

334
// FetchNodeFeatures returns the features of the given node. If no features are
335
// known for the node, an empty feature vector is returned.
336
//
337
// NOTE: this is part of the graphdb.NodeTraverser interface.
338
func (s *SQLStore) FetchNodeFeatures(nodePub route.Vertex) (
339
        *lnwire.FeatureVector, error) {
×
340

×
341
        ctx := context.TODO()
×
342

×
343
        return fetchNodeFeatures(ctx, s.db, nodePub)
×
344
}
×
345

346
// LookupAlias attempts to return the alias as advertised by the target node.
347
//
348
// NOTE: part of the V1Store interface.
349
func (s *SQLStore) LookupAlias(pub *btcec.PublicKey) (string, error) {
×
350
        var (
×
351
                ctx   = context.TODO()
×
352
                alias string
×
353
        )
×
354
        err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
×
355
                dbNode, err := db.GetNodeByPubKey(
×
356
                        ctx, sqlc.GetNodeByPubKeyParams{
×
357
                                Version: int16(ProtocolV1),
×
358
                                PubKey:  pub.SerializeCompressed(),
×
359
                        },
×
360
                )
×
361
                if errors.Is(err, sql.ErrNoRows) {
×
362
                        return ErrNodeAliasNotFound
×
363
                } else if err != nil {
×
364
                        return fmt.Errorf("unable to fetch node: %w", err)
×
365
                }
×
366

367
                if !dbNode.Alias.Valid {
×
368
                        return ErrNodeAliasNotFound
×
369
                }
×
370

371
                alias = dbNode.Alias.String
×
372

×
373
                return nil
×
374
        }, sqldb.NoOpReset)
375
        if err != nil {
×
376
                return "", fmt.Errorf("unable to look up alias: %w", err)
×
377
        }
×
378

379
        return alias, nil
×
380
}
381

382
// SourceNode returns the source node of the graph. The source node is treated
383
// as the center node within a star-graph. This method may be used to kick off
384
// a path finding algorithm in order to explore the reachability of another
385
// node based off the source node.
386
//
387
// NOTE: part of the V1Store interface.
388
func (s *SQLStore) SourceNode() (*models.LightningNode, error) {
×
389
        ctx := context.TODO()
×
390

×
391
        var node *models.LightningNode
×
392
        err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
×
393
                _, nodePub, err := getSourceNode(ctx, db, ProtocolV1)
×
394
                if err != nil {
×
395
                        return fmt.Errorf("unable to fetch V1 source node: %w",
×
396
                                err)
×
397
                }
×
398

399
                _, node, err = getNodeByPubKey(ctx, db, nodePub)
×
400

×
401
                return err
×
402
        }, sqldb.NoOpReset)
403
        if err != nil {
×
404
                return nil, fmt.Errorf("unable to fetch source node: %w", err)
×
405
        }
×
406

407
        return node, nil
×
408
}
409

410
// SetSourceNode sets the source node within the graph database. The source
411
// node is to be used as the center of a star-graph within path finding
412
// algorithms.
413
//
414
// NOTE: part of the V1Store interface.
415
func (s *SQLStore) SetSourceNode(node *models.LightningNode) error {
×
416
        ctx := context.TODO()
×
417

×
418
        return s.db.ExecTx(ctx, sqldb.WriteTxOpt(), func(db SQLQueries) error {
×
419
                id, err := upsertNode(ctx, db, node)
×
420
                if err != nil {
×
421
                        return fmt.Errorf("unable to upsert source node: %w",
×
422
                                err)
×
423
                }
×
424

425
                // Make sure that if a source node for this version is already
426
                // set, then the ID is the same as the one we are about to set.
427
                dbSourceNodeID, _, err := getSourceNode(ctx, db, ProtocolV1)
×
428
                if err != nil && !errors.Is(err, ErrSourceNodeNotSet) {
×
429
                        return fmt.Errorf("unable to fetch source node: %w",
×
430
                                err)
×
431
                } else if err == nil {
×
432
                        if dbSourceNodeID != id {
×
433
                                return fmt.Errorf("v1 source node already "+
×
434
                                        "set to a different node: %d vs %d",
×
435
                                        dbSourceNodeID, id)
×
436
                        }
×
437

438
                        return nil
×
439
                }
440

441
                return db.AddSourceNode(ctx, id)
×
442
        }, sqldb.NoOpReset)
443
}
444

445
// NodeUpdatesInHorizon returns all the known lightning node which have an
446
// update timestamp within the passed range. This method can be used by two
447
// nodes to quickly determine if they have the same set of up to date node
448
// announcements.
449
//
450
// NOTE: This is part of the V1Store interface.
451
func (s *SQLStore) NodeUpdatesInHorizon(startTime,
452
        endTime time.Time) ([]models.LightningNode, error) {
×
453

×
454
        ctx := context.TODO()
×
455

×
456
        var nodes []models.LightningNode
×
457
        err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
×
458
                dbNodes, err := db.GetNodesByLastUpdateRange(
×
459
                        ctx, sqlc.GetNodesByLastUpdateRangeParams{
×
460
                                StartTime: sqldb.SQLInt64(startTime.Unix()),
×
461
                                EndTime:   sqldb.SQLInt64(endTime.Unix()),
×
462
                        },
×
463
                )
×
464
                if err != nil {
×
465
                        return fmt.Errorf("unable to fetch nodes: %w", err)
×
466
                }
×
467

468
                for _, dbNode := range dbNodes {
×
469
                        node, err := buildNode(ctx, db, &dbNode)
×
470
                        if err != nil {
×
471
                                return fmt.Errorf("unable to build node: %w",
×
472
                                        err)
×
473
                        }
×
474

475
                        nodes = append(nodes, *node)
×
476
                }
477

478
                return nil
×
479
        }, sqldb.NoOpReset)
480
        if err != nil {
×
481
                return nil, fmt.Errorf("unable to fetch nodes: %w", err)
×
482
        }
×
483

484
        return nodes, nil
×
485
}
486

487
// AddChannelEdge adds a new (undirected, blank) edge to the graph database. An
488
// undirected edge from the two target nodes are created. The information stored
489
// denotes the static attributes of the channel, such as the channelID, the keys
490
// involved in creation of the channel, and the set of features that the channel
491
// supports. The chanPoint and chanID are used to uniquely identify the edge
492
// globally within the database.
493
//
494
// NOTE: part of the V1Store interface.
495
func (s *SQLStore) AddChannelEdge(edge *models.ChannelEdgeInfo,
496
        opts ...batch.SchedulerOption) error {
×
497

×
498
        ctx := context.TODO()
×
499

×
500
        var alreadyExists bool
×
501
        r := &batch.Request[SQLQueries]{
×
502
                Opts: batch.NewSchedulerOptions(opts...),
×
503
                Reset: func() {
×
504
                        alreadyExists = false
×
505
                },
×
506
                Do: func(tx SQLQueries) error {
×
507
                        err := insertChannel(ctx, tx, edge)
×
508

×
509
                        // Silence ErrEdgeAlreadyExist so that the batch can
×
510
                        // succeed, but propagate the error via local state.
×
511
                        if errors.Is(err, ErrEdgeAlreadyExist) {
×
512
                                alreadyExists = true
×
513
                                return nil
×
514
                        }
×
515

516
                        return err
×
517
                },
518
                OnCommit: func(err error) error {
×
519
                        switch {
×
520
                        case err != nil:
×
521
                                return err
×
522
                        case alreadyExists:
×
523
                                return ErrEdgeAlreadyExist
×
524
                        default:
×
525
                                s.rejectCache.remove(edge.ChannelID)
×
526
                                s.chanCache.remove(edge.ChannelID)
×
527
                                return nil
×
528
                        }
529
                },
530
        }
531

532
        return s.chanScheduler.Execute(ctx, r)
×
533
}
534

535
// HighestChanID returns the "highest" known channel ID in the channel graph.
536
// This represents the "newest" channel from the PoV of the chain. This method
537
// can be used by peers to quickly determine if their graphs are in sync.
538
//
539
// NOTE: This is part of the V1Store interface.
540
func (s *SQLStore) HighestChanID() (uint64, error) {
×
541
        ctx := context.TODO()
×
542

×
543
        var highestChanID uint64
×
544
        err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
×
545
                chanID, err := db.HighestSCID(ctx, int16(ProtocolV1))
×
546
                if errors.Is(err, sql.ErrNoRows) {
×
547
                        return nil
×
548
                } else if err != nil {
×
549
                        return fmt.Errorf("unable to fetch highest chan ID: %w",
×
550
                                err)
×
551
                }
×
552

553
                highestChanID = byteOrder.Uint64(chanID)
×
554

×
555
                return nil
×
556
        }, sqldb.NoOpReset)
557
        if err != nil {
×
558
                return 0, fmt.Errorf("unable to fetch highest chan ID: %w", err)
×
559
        }
×
560

561
        return highestChanID, nil
×
562
}
563

564
// UpdateEdgePolicy updates the edge routing policy for a single directed edge
565
// within the database for the referenced channel. The `flags` attribute within
566
// the ChannelEdgePolicy determines which of the directed edges are being
567
// updated. If the flag is 1, then the first node's information is being
568
// updated, otherwise it's the second node's information. The node ordering is
569
// determined by the lexicographical ordering of the identity public keys of the
570
// nodes on either side of the channel.
571
//
572
// NOTE: part of the V1Store interface.
573
func (s *SQLStore) UpdateEdgePolicy(edge *models.ChannelEdgePolicy,
NEW
574
        opts ...batch.SchedulerOption) (route.Vertex, route.Vertex, error) {
×
NEW
575

×
NEW
576
        ctx := context.TODO()
×
NEW
577

×
NEW
578
        var (
×
NEW
579
                isUpdate1    bool
×
NEW
580
                edgeNotFound bool
×
NEW
581
                from, to     route.Vertex
×
NEW
582
        )
×
NEW
583

×
NEW
584
        r := &batch.Request[SQLQueries]{
×
NEW
585
                Opts: batch.NewSchedulerOptions(opts...),
×
NEW
586
                Reset: func() {
×
NEW
587
                        isUpdate1 = false
×
NEW
588
                        edgeNotFound = false
×
NEW
589
                },
×
NEW
590
                Do: func(tx SQLQueries) error {
×
NEW
591
                        var err error
×
NEW
592
                        from, to, isUpdate1, err = updateChanEdgePolicy(
×
NEW
593
                                ctx, tx, edge,
×
NEW
594
                        )
×
NEW
595
                        if err != nil {
×
NEW
596
                                log.Errorf("UpdateEdgePolicy faild: %v", err)
×
NEW
597
                        }
×
598

599
                        // Silence ErrEdgeNotFound so that the batch can
600
                        // succeed, but propagate the error via local state.
NEW
601
                        if errors.Is(err, ErrEdgeNotFound) {
×
NEW
602
                                edgeNotFound = true
×
NEW
603
                                return nil
×
NEW
604
                        }
×
605

NEW
606
                        return err
×
607
                },
NEW
608
                OnCommit: func(err error) error {
×
NEW
609
                        switch {
×
NEW
610
                        case err != nil:
×
NEW
611
                                return err
×
NEW
612
                        case edgeNotFound:
×
NEW
613
                                return ErrEdgeNotFound
×
NEW
614
                        default:
×
NEW
615
                                s.updateEdgeCache(edge, isUpdate1)
×
NEW
616
                                return nil
×
617
                        }
618
                },
619
        }
620

NEW
621
        err := s.chanScheduler.Execute(ctx, r)
×
NEW
622

×
NEW
623
        return from, to, err
×
624
}
625

626
// updateEdgeCache updates our reject and channel caches with the new
627
// edge policy information.
628
func (s *SQLStore) updateEdgeCache(e *models.ChannelEdgePolicy,
NEW
629
        isUpdate1 bool) {
×
NEW
630

×
NEW
631
        // If an entry for this channel is found in reject cache, we'll modify
×
NEW
632
        // the entry with the updated timestamp for the direction that was just
×
NEW
633
        // written. If the edge doesn't exist, we'll load the cache entry lazily
×
NEW
634
        // during the next query for this edge.
×
NEW
635
        if entry, ok := s.rejectCache.get(e.ChannelID); ok {
×
NEW
636
                if isUpdate1 {
×
NEW
637
                        entry.upd1Time = e.LastUpdate.Unix()
×
NEW
638
                } else {
×
NEW
639
                        entry.upd2Time = e.LastUpdate.Unix()
×
NEW
640
                }
×
NEW
641
                s.rejectCache.insert(e.ChannelID, entry)
×
642
        }
643

644
        // If an entry for this channel is found in channel cache, we'll modify
645
        // the entry with the updated policy for the direction that was just
646
        // written. If the edge doesn't exist, we'll defer loading the info and
647
        // policies and lazily read from disk during the next query.
NEW
648
        if channel, ok := s.chanCache.get(e.ChannelID); ok {
×
NEW
649
                if isUpdate1 {
×
NEW
650
                        channel.Policy1 = e
×
NEW
651
                } else {
×
NEW
652
                        channel.Policy2 = e
×
NEW
653
                }
×
NEW
654
                s.chanCache.insert(e.ChannelID, channel)
×
655
        }
656
}
657

658
// updateChanEdgePolicy upserts the channel policy info we have stored for
659
// a channel we already know of.
660
func updateChanEdgePolicy(ctx context.Context, tx SQLQueries,
661
        edge *models.ChannelEdgePolicy) (route.Vertex, route.Vertex, bool,
NEW
662
        error) {
×
NEW
663

×
NEW
664
        var (
×
NEW
665
                node1Pub, node2Pub route.Vertex
×
NEW
666
                isNode1            bool
×
NEW
667
                chanIDB            [8]byte
×
NEW
668
        )
×
NEW
669
        byteOrder.PutUint64(chanIDB[:], edge.ChannelID)
×
NEW
670

×
NEW
671
        // Check that this edge policy refers to a channel that we already
×
NEW
672
        // know of. We do this explicitly so that we can return the appropriate
×
NEW
673
        // ErrEdgeNotFound error if the channel doesn't exist, rather than
×
NEW
674
        // abort the transaction which would abort the entire batch.
×
NEW
675
        dbChan, err := tx.GetChannelAndNodesBySCID(
×
NEW
676
                ctx, sqlc.GetChannelAndNodesBySCIDParams{
×
NEW
677
                        Scid:    chanIDB[:],
×
NEW
678
                        Version: int16(ProtocolV1),
×
NEW
679
                },
×
NEW
680
        )
×
NEW
681
        if errors.Is(err, sql.ErrNoRows) {
×
NEW
682
                return node1Pub, node2Pub, false, ErrEdgeNotFound
×
NEW
683
        } else if err != nil {
×
NEW
684
                return node1Pub, node2Pub, false, fmt.Errorf("unable to "+
×
NEW
685
                        "fetch channel(%v): %w", edge.ChannelID, err)
×
NEW
686
        }
×
687

NEW
688
        copy(node1Pub[:], dbChan.Node1PubKey)
×
NEW
689
        copy(node2Pub[:], dbChan.Node2PubKey)
×
NEW
690

×
NEW
691
        // Figure out which node this edge is from.
×
NEW
692
        isNode1 = edge.ChannelFlags&lnwire.ChanUpdateDirection == 0
×
NEW
693
        nodeID := dbChan.NodeID1
×
NEW
694
        if !isNode1 {
×
NEW
695
                nodeID = dbChan.NodeID2
×
NEW
696
        }
×
697

NEW
698
        var (
×
NEW
699
                inboundBase sql.NullInt64
×
NEW
700
                inboundRate sql.NullInt64
×
NEW
701
        )
×
NEW
702
        edge.InboundFee.WhenSome(func(fee lnwire.Fee) {
×
NEW
703
                inboundRate = sqldb.SQLInt64(fee.FeeRate)
×
NEW
704
                inboundBase = sqldb.SQLInt64(fee.BaseFee)
×
NEW
705
        })
×
706

NEW
707
        id, err := tx.UpsertEdgePolicy(ctx, sqlc.UpsertEdgePolicyParams{
×
NEW
708
                Version:     int16(ProtocolV1),
×
NEW
709
                ChannelID:   dbChan.ID,
×
NEW
710
                NodeID:      nodeID,
×
NEW
711
                Timelock:    int32(edge.TimeLockDelta),
×
NEW
712
                FeePpm:      int64(edge.FeeProportionalMillionths),
×
NEW
713
                BaseFeeMsat: int64(edge.FeeBaseMSat),
×
NEW
714
                MinHtlcMsat: int64(edge.MinHTLC),
×
NEW
715
                LastUpdate:  sqldb.SQLInt64(edge.LastUpdate.Unix()),
×
NEW
716
                Disabled: sql.NullBool{
×
NEW
717
                        Valid: true,
×
NEW
718
                        Bool:  edge.IsDisabled(),
×
NEW
719
                },
×
NEW
720
                MaxHtlcMsat: sql.NullInt64{
×
NEW
721
                        Valid: edge.MessageFlags.HasMaxHtlc(),
×
NEW
722
                        Int64: int64(edge.MaxHTLC),
×
NEW
723
                },
×
NEW
724
                InboundBaseFeeMsat:      inboundBase,
×
NEW
725
                InboundFeeRateMilliMsat: inboundRate,
×
NEW
726
                Signature:               edge.SigBytes,
×
NEW
727
        })
×
NEW
728
        if err != nil {
×
NEW
729
                return node1Pub, node2Pub, isNode1,
×
NEW
730
                        fmt.Errorf("unable to upsert edge policy: %w", err)
×
NEW
731
        }
×
732

733
        // Convert the flat extra opaque data into a map of TLV types to
734
        // values.
NEW
735
        extra, err := marshalExtraOpaqueData(edge.ExtraOpaqueData)
×
NEW
736
        if err != nil {
×
NEW
737
                return node1Pub, node2Pub, false, fmt.Errorf("unable to "+
×
NEW
738
                        "marshal extra opaque data: %w", err)
×
NEW
739
        }
×
740

741
        // Update the channel policy's extra signed fields.
NEW
742
        err = upsertChanPolicyExtraSignedFields(ctx, tx, id, extra)
×
NEW
743
        if err != nil {
×
NEW
744
                return node1Pub, node2Pub, false, fmt.Errorf("inserting chan "+
×
NEW
745
                        "policy extra TLVs: %w", err)
×
NEW
746
        }
×
747

NEW
748
        return node1Pub, node2Pub, isNode1, nil
×
749
}
750

751
// getNodeByPubKey attempts to look up a target node by its public key.
752
func getNodeByPubKey(ctx context.Context, db SQLQueries,
753
        pubKey route.Vertex) (int64, *models.LightningNode, error) {
×
754

×
755
        dbNode, err := db.GetNodeByPubKey(
×
756
                ctx, sqlc.GetNodeByPubKeyParams{
×
757
                        Version: int16(ProtocolV1),
×
758
                        PubKey:  pubKey[:],
×
759
                },
×
760
        )
×
761
        if errors.Is(err, sql.ErrNoRows) {
×
762
                return 0, nil, ErrGraphNodeNotFound
×
763
        } else if err != nil {
×
764
                return 0, nil, fmt.Errorf("unable to fetch node: %w", err)
×
765
        }
×
766

767
        node, err := buildNode(ctx, db, &dbNode)
×
768
        if err != nil {
×
769
                return 0, nil, fmt.Errorf("unable to build node: %w", err)
×
770
        }
×
771

772
        return dbNode.ID, node, nil
×
773
}
774

775
// buildNode constructs a LightningNode instance from the given database node
776
// record. The node's features, addresses and extra signed fields are also
777
// fetched from the database and set on the node.
778
func buildNode(ctx context.Context, db SQLQueries, dbNode *sqlc.Node) (
779
        *models.LightningNode, error) {
×
780

×
781
        if dbNode.Version != int16(ProtocolV1) {
×
782
                return nil, fmt.Errorf("unsupported node version: %d",
×
783
                        dbNode.Version)
×
784
        }
×
785

786
        var pub [33]byte
×
787
        copy(pub[:], dbNode.PubKey)
×
788

×
789
        node := &models.LightningNode{
×
790
                PubKeyBytes: pub,
×
791
                Features:    lnwire.EmptyFeatureVector(),
×
792
                LastUpdate:  time.Unix(0, 0),
×
793
        }
×
794

×
795
        if len(dbNode.Signature) == 0 {
×
796
                return node, nil
×
797
        }
×
798

799
        node.HaveNodeAnnouncement = true
×
800
        node.AuthSigBytes = dbNode.Signature
×
801
        node.Alias = dbNode.Alias.String
×
802
        node.LastUpdate = time.Unix(dbNode.LastUpdate.Int64, 0)
×
803

×
804
        var err error
×
805
        node.Color, err = DecodeHexColor(dbNode.Color.String)
×
806
        if err != nil {
×
807
                return nil, fmt.Errorf("unable to decode color: %w", err)
×
808
        }
×
809

810
        // Fetch the node's features.
811
        node.Features, err = getNodeFeatures(ctx, db, dbNode.ID)
×
812
        if err != nil {
×
813
                return nil, fmt.Errorf("unable to fetch node(%d) "+
×
814
                        "features: %w", dbNode.ID, err)
×
815
        }
×
816

817
        // Fetch the node's addresses.
818
        _, node.Addresses, err = getNodeAddresses(ctx, db, pub[:])
×
819
        if err != nil {
×
820
                return nil, fmt.Errorf("unable to fetch node(%d) "+
×
821
                        "addresses: %w", dbNode.ID, err)
×
822
        }
×
823

824
        // Fetch the node's extra signed fields.
825
        extraTLVMap, err := getNodeExtraSignedFields(ctx, db, dbNode.ID)
×
826
        if err != nil {
×
827
                return nil, fmt.Errorf("unable to fetch node(%d) "+
×
828
                        "extra signed fields: %w", dbNode.ID, err)
×
829
        }
×
830

831
        recs, err := lnwire.CustomRecords(extraTLVMap).Serialize()
×
832
        if err != nil {
×
833
                return nil, fmt.Errorf("unable to serialize extra signed "+
×
834
                        "fields: %w", err)
×
835
        }
×
836

837
        if len(recs) != 0 {
×
838
                node.ExtraOpaqueData = recs
×
839
        }
×
840

841
        return node, nil
×
842
}
843

844
// getNodeFeatures fetches the feature bits and constructs the feature vector
845
// for a node with the given DB ID.
846
func getNodeFeatures(ctx context.Context, db SQLQueries,
847
        nodeID int64) (*lnwire.FeatureVector, error) {
×
848

×
849
        rows, err := db.GetNodeFeatures(ctx, nodeID)
×
850
        if err != nil {
×
851
                return nil, fmt.Errorf("unable to get node(%d) features: %w",
×
852
                        nodeID, err)
×
853
        }
×
854

855
        features := lnwire.EmptyFeatureVector()
×
856
        for _, feature := range rows {
×
857
                features.Set(lnwire.FeatureBit(feature.FeatureBit))
×
858
        }
×
859

860
        return features, nil
×
861
}
862

863
// getNodeExtraSignedFields fetches the extra signed fields for a node with the
864
// given DB ID.
865
func getNodeExtraSignedFields(ctx context.Context, db SQLQueries,
866
        nodeID int64) (map[uint64][]byte, error) {
×
867

×
868
        fields, err := db.GetExtraNodeTypes(ctx, nodeID)
×
869
        if err != nil {
×
870
                return nil, fmt.Errorf("unable to get node(%d) extra "+
×
871
                        "signed fields: %w", nodeID, err)
×
872
        }
×
873

874
        extraFields := make(map[uint64][]byte)
×
875
        for _, field := range fields {
×
876
                extraFields[uint64(field.Type)] = field.Value
×
877
        }
×
878

879
        return extraFields, nil
×
880
}
881

882
// upsertNode upserts the node record into the database. If the node already
883
// exists, then the node's information is updated. If the node doesn't exist,
884
// then a new node is created. The node's features, addresses and extra TLV
885
// types are also updated. The node's DB ID is returned.
886
func upsertNode(ctx context.Context, db SQLQueries,
887
        node *models.LightningNode) (int64, error) {
×
888

×
889
        params := sqlc.UpsertNodeParams{
×
890
                Version: int16(ProtocolV1),
×
891
                PubKey:  node.PubKeyBytes[:],
×
892
        }
×
893

×
894
        if node.HaveNodeAnnouncement {
×
895
                params.LastUpdate = sqldb.SQLInt64(node.LastUpdate.Unix())
×
896
                params.Color = sqldb.SQLStr(EncodeHexColor(node.Color))
×
897
                params.Alias = sqldb.SQLStr(node.Alias)
×
898
                params.Signature = node.AuthSigBytes
×
899
        }
×
900

901
        nodeID, err := db.UpsertNode(ctx, params)
×
902
        if err != nil {
×
903
                return 0, fmt.Errorf("upserting node(%x): %w", node.PubKeyBytes,
×
904
                        err)
×
905
        }
×
906

907
        // We can exit here if we don't have the announcement yet.
908
        if !node.HaveNodeAnnouncement {
×
909
                return nodeID, nil
×
910
        }
×
911

912
        // Update the node's features.
913
        err = upsertNodeFeatures(ctx, db, nodeID, node.Features)
×
914
        if err != nil {
×
915
                return 0, fmt.Errorf("inserting node features: %w", err)
×
916
        }
×
917

918
        // Update the node's addresses.
919
        err = upsertNodeAddresses(ctx, db, nodeID, node.Addresses)
×
920
        if err != nil {
×
921
                return 0, fmt.Errorf("inserting node addresses: %w", err)
×
922
        }
×
923

924
        // Convert the flat extra opaque data into a map of TLV types to
925
        // values.
926
        extra, err := marshalExtraOpaqueData(node.ExtraOpaqueData)
×
927
        if err != nil {
×
928
                return 0, fmt.Errorf("unable to marshal extra opaque data: %w",
×
929
                        err)
×
930
        }
×
931

932
        // Update the node's extra signed fields.
933
        err = upsertNodeExtraSignedFields(ctx, db, nodeID, extra)
×
934
        if err != nil {
×
935
                return 0, fmt.Errorf("inserting node extra TLVs: %w", err)
×
936
        }
×
937

938
        return nodeID, nil
×
939
}
940

941
// upsertNodeFeatures updates the node's features node_features table. This
942
// includes deleting any feature bits no longer present and inserting any new
943
// feature bits. If the feature bit does not yet exist in the features table,
944
// then an entry is created in that table first.
945
func upsertNodeFeatures(ctx context.Context, db SQLQueries, nodeID int64,
946
        features *lnwire.FeatureVector) error {
×
947

×
948
        // Get any existing features for the node.
×
949
        existingFeatures, err := db.GetNodeFeatures(ctx, nodeID)
×
950
        if err != nil && !errors.Is(err, sql.ErrNoRows) {
×
951
                return err
×
952
        }
×
953

954
        // Copy the nodes latest set of feature bits.
955
        newFeatures := make(map[int32]struct{})
×
956
        if features != nil {
×
957
                for feature := range features.Features() {
×
958
                        newFeatures[int32(feature)] = struct{}{}
×
959
                }
×
960
        }
961

962
        // For any current feature that already exists in the DB, remove it from
963
        // the in-memory map. For any existing feature that does not exist in
964
        // the in-memory map, delete it from the database.
965
        for _, feature := range existingFeatures {
×
966
                // The feature is still present, so there are no updates to be
×
967
                // made.
×
968
                if _, ok := newFeatures[feature.FeatureBit]; ok {
×
969
                        delete(newFeatures, feature.FeatureBit)
×
970
                        continue
×
971
                }
972

973
                // The feature is no longer present, so we remove it from the
974
                // database.
975
                err := db.DeleteNodeFeature(ctx, sqlc.DeleteNodeFeatureParams{
×
976
                        NodeID:     nodeID,
×
977
                        FeatureBit: feature.FeatureBit,
×
978
                })
×
979
                if err != nil {
×
980
                        return fmt.Errorf("unable to delete node(%d) "+
×
981
                                "feature(%v): %w", nodeID, feature.FeatureBit,
×
982
                                err)
×
983
                }
×
984
        }
985

986
        // Any remaining entries in newFeatures are new features that need to be
987
        // added to the database for the first time.
988
        for feature := range newFeatures {
×
989
                err = db.InsertNodeFeature(ctx, sqlc.InsertNodeFeatureParams{
×
990
                        NodeID:     nodeID,
×
991
                        FeatureBit: feature,
×
992
                })
×
993
                if err != nil {
×
994
                        return fmt.Errorf("unable to insert node(%d) "+
×
995
                                "feature(%v): %w", nodeID, feature, err)
×
996
                }
×
997
        }
998

999
        return nil
×
1000
}
1001

1002
// fetchNodeFeatures fetches the features for a node with the given public key.
1003
func fetchNodeFeatures(ctx context.Context, queries SQLQueries,
1004
        nodePub route.Vertex) (*lnwire.FeatureVector, error) {
×
1005

×
1006
        rows, err := queries.GetNodeFeaturesByPubKey(
×
1007
                ctx, sqlc.GetNodeFeaturesByPubKeyParams{
×
1008
                        PubKey:  nodePub[:],
×
1009
                        Version: int16(ProtocolV1),
×
1010
                },
×
1011
        )
×
1012
        if err != nil {
×
1013
                return nil, fmt.Errorf("unable to get node(%s) features: %w",
×
1014
                        nodePub, err)
×
1015
        }
×
1016

1017
        features := lnwire.EmptyFeatureVector()
×
1018
        for _, bit := range rows {
×
1019
                features.Set(lnwire.FeatureBit(bit))
×
1020
        }
×
1021

1022
        return features, nil
×
1023
}
1024

1025
// dbAddressType is an enum type that represents the different address types
1026
// that we store in the node_addresses table. The address type determines how
1027
// the address is to be serialised/deserialize.
1028
type dbAddressType uint8
1029

1030
const (
1031
        addressTypeIPv4   dbAddressType = 1
1032
        addressTypeIPv6   dbAddressType = 2
1033
        addressTypeTorV2  dbAddressType = 3
1034
        addressTypeTorV3  dbAddressType = 4
1035
        addressTypeOpaque dbAddressType = math.MaxInt8
1036
)
1037

1038
// upsertNodeAddresses updates the node's addresses in the database. This
1039
// includes deleting any existing addresses and inserting the new set of
1040
// addresses. The deletion is necessary since the ordering of the addresses may
1041
// change, and we need to ensure that the database reflects the latest set of
1042
// addresses so that at the time of reconstructing the node announcement, the
1043
// order is preserved and the signature over the message remains valid.
1044
func upsertNodeAddresses(ctx context.Context, db SQLQueries, nodeID int64,
1045
        addresses []net.Addr) error {
×
1046

×
1047
        // Delete any existing addresses for the node. This is required since
×
1048
        // even if the new set of addresses is the same, the ordering may have
×
1049
        // changed for a given address type.
×
1050
        err := db.DeleteNodeAddresses(ctx, nodeID)
×
1051
        if err != nil {
×
1052
                return fmt.Errorf("unable to delete node(%d) addresses: %w",
×
1053
                        nodeID, err)
×
1054
        }
×
1055

1056
        // Copy the nodes latest set of addresses.
1057
        newAddresses := map[dbAddressType][]string{
×
1058
                addressTypeIPv4:   {},
×
1059
                addressTypeIPv6:   {},
×
1060
                addressTypeTorV2:  {},
×
1061
                addressTypeTorV3:  {},
×
1062
                addressTypeOpaque: {},
×
1063
        }
×
1064
        addAddr := func(t dbAddressType, addr net.Addr) {
×
1065
                newAddresses[t] = append(newAddresses[t], addr.String())
×
1066
        }
×
1067

1068
        for _, address := range addresses {
×
1069
                switch addr := address.(type) {
×
1070
                case *net.TCPAddr:
×
1071
                        if ip4 := addr.IP.To4(); ip4 != nil {
×
1072
                                addAddr(addressTypeIPv4, addr)
×
1073
                        } else if ip6 := addr.IP.To16(); ip6 != nil {
×
1074
                                addAddr(addressTypeIPv6, addr)
×
1075
                        } else {
×
1076
                                return fmt.Errorf("unhandled IP address: %v",
×
1077
                                        addr)
×
1078
                        }
×
1079

1080
                case *tor.OnionAddr:
×
1081
                        switch len(addr.OnionService) {
×
1082
                        case tor.V2Len:
×
1083
                                addAddr(addressTypeTorV2, addr)
×
1084
                        case tor.V3Len:
×
1085
                                addAddr(addressTypeTorV3, addr)
×
1086
                        default:
×
1087
                                return fmt.Errorf("invalid length for a tor " +
×
1088
                                        "address")
×
1089
                        }
1090

1091
                case *lnwire.OpaqueAddrs:
×
1092
                        addAddr(addressTypeOpaque, addr)
×
1093

1094
                default:
×
1095
                        return fmt.Errorf("unhandled address type: %T", addr)
×
1096
                }
1097
        }
1098

1099
        // Any remaining entries in newAddresses are new addresses that need to
1100
        // be added to the database for the first time.
1101
        for addrType, addrList := range newAddresses {
×
1102
                for position, addr := range addrList {
×
1103
                        err := db.InsertNodeAddress(
×
1104
                                ctx, sqlc.InsertNodeAddressParams{
×
1105
                                        NodeID:   nodeID,
×
1106
                                        Type:     int16(addrType),
×
1107
                                        Address:  addr,
×
1108
                                        Position: int32(position),
×
1109
                                },
×
1110
                        )
×
1111
                        if err != nil {
×
1112
                                return fmt.Errorf("unable to insert "+
×
1113
                                        "node(%d) address(%v): %w", nodeID,
×
1114
                                        addr, err)
×
1115
                        }
×
1116
                }
1117
        }
1118

1119
        return nil
×
1120
}
1121

1122
// getNodeAddresses fetches the addresses for a node with the given public key.
1123
func getNodeAddresses(ctx context.Context, db SQLQueries, nodePub []byte) (bool,
1124
        []net.Addr, error) {
×
1125

×
1126
        // GetNodeAddressesByPubKey ensures that the addresses for a given type
×
1127
        // are returned in the same order as they were inserted.
×
1128
        rows, err := db.GetNodeAddressesByPubKey(
×
1129
                ctx, sqlc.GetNodeAddressesByPubKeyParams{
×
1130
                        Version: int16(ProtocolV1),
×
1131
                        PubKey:  nodePub,
×
1132
                },
×
1133
        )
×
1134
        if err != nil {
×
1135
                return false, nil, err
×
1136
        }
×
1137

1138
        // GetNodeAddressesByPubKey uses a left join so there should always be
1139
        // at least one row returned if the node exists even if it has no
1140
        // addresses.
1141
        if len(rows) == 0 {
×
1142
                return false, nil, nil
×
1143
        }
×
1144

1145
        addresses := make([]net.Addr, 0, len(rows))
×
1146
        for _, addr := range rows {
×
1147
                if !(addr.Type.Valid && addr.Address.Valid) {
×
1148
                        continue
×
1149
                }
1150

1151
                address := addr.Address.String
×
1152

×
1153
                switch dbAddressType(addr.Type.Int16) {
×
1154
                case addressTypeIPv4:
×
1155
                        tcp, err := net.ResolveTCPAddr("tcp4", address)
×
1156
                        if err != nil {
×
1157
                                return false, nil, nil
×
1158
                        }
×
1159
                        tcp.IP = tcp.IP.To4()
×
1160

×
1161
                        addresses = append(addresses, tcp)
×
1162

1163
                case addressTypeIPv6:
×
1164
                        tcp, err := net.ResolveTCPAddr("tcp6", address)
×
1165
                        if err != nil {
×
1166
                                return false, nil, nil
×
1167
                        }
×
1168
                        addresses = append(addresses, tcp)
×
1169

1170
                case addressTypeTorV3, addressTypeTorV2:
×
1171
                        service, portStr, err := net.SplitHostPort(address)
×
1172
                        if err != nil {
×
1173
                                return false, nil, fmt.Errorf("unable to "+
×
1174
                                        "split tor v3 address: %v",
×
1175
                                        addr.Address)
×
1176
                        }
×
1177

1178
                        port, err := strconv.Atoi(portStr)
×
1179
                        if err != nil {
×
1180
                                return false, nil, err
×
1181
                        }
×
1182

1183
                        addresses = append(addresses, &tor.OnionAddr{
×
1184
                                OnionService: service,
×
1185
                                Port:         port,
×
1186
                        })
×
1187

1188
                case addressTypeOpaque:
×
1189
                        opaque, err := hex.DecodeString(address)
×
1190
                        if err != nil {
×
1191
                                return false, nil, fmt.Errorf("unable to "+
×
1192
                                        "decode opaque address: %v", addr)
×
1193
                        }
×
1194

1195
                        addresses = append(addresses, &lnwire.OpaqueAddrs{
×
1196
                                Payload: opaque,
×
1197
                        })
×
1198

1199
                default:
×
1200
                        return false, nil, fmt.Errorf("unknown address "+
×
1201
                                "type: %v", addr.Type)
×
1202
                }
1203
        }
1204

1205
        return true, addresses, nil
×
1206
}
1207

1208
// upsertNodeExtraSignedFields updates the node's extra signed fields in the
1209
// database. This includes updating any existing types, inserting any new types,
1210
// and deleting any types that are no longer present.
1211
func upsertNodeExtraSignedFields(ctx context.Context, db SQLQueries,
1212
        nodeID int64, extraFields map[uint64][]byte) error {
×
1213

×
1214
        // Get any existing extra signed fields for the node.
×
1215
        existingFields, err := db.GetExtraNodeTypes(ctx, nodeID)
×
1216
        if err != nil {
×
1217
                return err
×
1218
        }
×
1219

1220
        // Make a lookup map of the existing field types so that we can use it
1221
        // to keep track of any fields we should delete.
1222
        m := make(map[uint64]bool)
×
1223
        for _, field := range existingFields {
×
1224
                m[uint64(field.Type)] = true
×
1225
        }
×
1226

1227
        // For all the new fields, we'll upsert them and remove them from the
1228
        // map of existing fields.
1229
        for tlvType, value := range extraFields {
×
1230
                err = db.UpsertNodeExtraType(
×
1231
                        ctx, sqlc.UpsertNodeExtraTypeParams{
×
1232
                                NodeID: nodeID,
×
1233
                                Type:   int64(tlvType),
×
1234
                                Value:  value,
×
1235
                        },
×
1236
                )
×
1237
                if err != nil {
×
1238
                        return fmt.Errorf("unable to upsert node(%d) extra "+
×
1239
                                "signed field(%v): %w", nodeID, tlvType, err)
×
1240
                }
×
1241

1242
                // Remove the field from the map of existing fields if it was
1243
                // present.
1244
                delete(m, tlvType)
×
1245
        }
1246

1247
        // For all the fields that are left in the map of existing fields, we'll
1248
        // delete them as they are no longer present in the new set of fields.
1249
        for tlvType := range m {
×
1250
                err = db.DeleteExtraNodeType(
×
1251
                        ctx, sqlc.DeleteExtraNodeTypeParams{
×
1252
                                NodeID: nodeID,
×
1253
                                Type:   int64(tlvType),
×
1254
                        },
×
1255
                )
×
1256
                if err != nil {
×
1257
                        return fmt.Errorf("unable to delete node(%d) extra "+
×
1258
                                "signed field(%v): %w", nodeID, tlvType, err)
×
1259
                }
×
1260
        }
1261

1262
        return nil
×
1263
}
1264

1265
// getSourceNode returns the DB node ID and pub key of the source node for the
1266
// specified protocol version.
1267
func getSourceNode(ctx context.Context, db SQLQueries,
1268
        version ProtocolVersion) (int64, route.Vertex, error) {
×
1269

×
1270
        var pubKey route.Vertex
×
1271

×
1272
        nodes, err := db.GetSourceNodesByVersion(ctx, int16(version))
×
1273
        if err != nil {
×
1274
                return 0, pubKey, fmt.Errorf("unable to fetch source node: %w",
×
1275
                        err)
×
1276
        }
×
1277

1278
        if len(nodes) == 0 {
×
1279
                return 0, pubKey, ErrSourceNodeNotSet
×
1280
        } else if len(nodes) > 1 {
×
1281
                return 0, pubKey, fmt.Errorf("multiple source nodes for "+
×
1282
                        "protocol %s found", version)
×
1283
        }
×
1284

1285
        copy(pubKey[:], nodes[0].PubKey)
×
1286

×
1287
        return nodes[0].NodeID, pubKey, nil
×
1288
}
1289

1290
// marshalExtraOpaqueData takes a flat byte slice parses it as a TLV stream.
1291
// This then produces a map from TLV type to value. If the input is not a
1292
// valid TLV stream, then an error is returned.
1293
func marshalExtraOpaqueData(data []byte) (map[uint64][]byte, error) {
×
1294
        r := bytes.NewReader(data)
×
1295

×
1296
        tlvStream, err := tlv.NewStream()
×
1297
        if err != nil {
×
1298
                return nil, err
×
1299
        }
×
1300

1301
        // Since ExtraOpaqueData is provided by a potentially malicious peer,
1302
        // pass it into the P2P decoding variant.
1303
        parsedTypes, err := tlvStream.DecodeWithParsedTypesP2P(r)
×
1304
        if err != nil {
×
1305
                return nil, err
×
1306
        }
×
1307
        if len(parsedTypes) == 0 {
×
1308
                return nil, nil
×
1309
        }
×
1310

1311
        records := make(map[uint64][]byte)
×
1312
        for k, v := range parsedTypes {
×
1313
                records[uint64(k)] = v
×
1314
        }
×
1315

1316
        return records, nil
×
1317
}
1318

1319
// insertChannel inserts a new channel record into the database.
1320
func insertChannel(ctx context.Context, db SQLQueries,
1321
        edge *models.ChannelEdgeInfo) error {
×
1322

×
1323
        var chanIDB [8]byte
×
1324
        byteOrder.PutUint64(chanIDB[:], edge.ChannelID)
×
1325

×
1326
        // Make sure that the channel doesn't already exist. We do this
×
1327
        // explicitly instead of relying on catching a unique constraint error
×
1328
        // because relying on SQL to throw that error would abort the entire
×
1329
        // batch of transactions.
×
1330
        _, err := db.GetChannelBySCID(
×
1331
                ctx, sqlc.GetChannelBySCIDParams{
×
1332
                        Scid:    chanIDB[:],
×
1333
                        Version: int16(ProtocolV1),
×
1334
                },
×
1335
        )
×
1336
        if err == nil {
×
1337
                return ErrEdgeAlreadyExist
×
1338
        } else if !errors.Is(err, sql.ErrNoRows) {
×
1339
                return fmt.Errorf("unable to fetch channel: %w", err)
×
1340
        }
×
1341

1342
        // Make sure that at least a "shell" entry for each node is present in
1343
        // the nodes table.
1344
        node1DBID, err := maybeCreateShellNode(ctx, db, edge.NodeKey1Bytes)
×
1345
        if err != nil {
×
1346
                return fmt.Errorf("unable to create shell node: %w", err)
×
1347
        }
×
1348

1349
        node2DBID, err := maybeCreateShellNode(ctx, db, edge.NodeKey2Bytes)
×
1350
        if err != nil {
×
1351
                return fmt.Errorf("unable to create shell node: %w", err)
×
1352
        }
×
1353

1354
        var capacity sql.NullInt64
×
1355
        if edge.Capacity != 0 {
×
1356
                capacity = sqldb.SQLInt64(int64(edge.Capacity))
×
1357
        }
×
1358

1359
        createParams := sqlc.CreateChannelParams{
×
1360
                Version:     int16(ProtocolV1),
×
1361
                Scid:        chanIDB[:],
×
1362
                NodeID1:     node1DBID,
×
1363
                NodeID2:     node2DBID,
×
1364
                Outpoint:    edge.ChannelPoint.String(),
×
1365
                Capacity:    capacity,
×
1366
                BitcoinKey1: edge.BitcoinKey1Bytes[:],
×
1367
                BitcoinKey2: edge.BitcoinKey2Bytes[:],
×
1368
        }
×
1369

×
1370
        if edge.AuthProof != nil {
×
1371
                proof := edge.AuthProof
×
1372

×
1373
                createParams.Node1Signature = proof.NodeSig1Bytes
×
1374
                createParams.Node2Signature = proof.NodeSig2Bytes
×
1375
                createParams.Bitcoin1Signature = proof.BitcoinSig1Bytes
×
1376
                createParams.Bitcoin2Signature = proof.BitcoinSig2Bytes
×
1377
        }
×
1378

1379
        // Insert the new channel record.
1380
        dbChanID, err := db.CreateChannel(ctx, createParams)
×
1381
        if err != nil {
×
1382
                return err
×
1383
        }
×
1384

1385
        // Insert any channel features.
1386
        if len(edge.Features) != 0 {
×
1387
                chanFeatures := lnwire.NewRawFeatureVector()
×
1388
                err := chanFeatures.Decode(bytes.NewReader(edge.Features))
×
1389
                if err != nil {
×
1390
                        return err
×
1391
                }
×
1392

1393
                fv := lnwire.NewFeatureVector(chanFeatures, lnwire.Features)
×
1394
                for feature := range fv.Features() {
×
1395
                        err = db.InsertChannelFeature(
×
1396
                                ctx, sqlc.InsertChannelFeatureParams{
×
1397
                                        ChannelID:  dbChanID,
×
1398
                                        FeatureBit: int32(feature),
×
1399
                                },
×
1400
                        )
×
1401
                        if err != nil {
×
1402
                                return fmt.Errorf("unable to insert "+
×
1403
                                        "channel(%d) feature(%v): %w", dbChanID,
×
1404
                                        feature, err)
×
1405
                        }
×
1406
                }
1407
        }
1408

1409
        // Finally, insert any extra TLV fields in the channel announcement.
1410
        extra, err := marshalExtraOpaqueData(edge.ExtraOpaqueData)
×
1411
        if err != nil {
×
1412
                return fmt.Errorf("unable to marshal extra opaque data: %w",
×
1413
                        err)
×
1414
        }
×
1415

1416
        for tlvType, value := range extra {
×
1417
                err := db.CreateChannelExtraType(
×
1418
                        ctx, sqlc.CreateChannelExtraTypeParams{
×
1419
                                ChannelID: dbChanID,
×
1420
                                Type:      int64(tlvType),
×
1421
                                Value:     value,
×
1422
                        },
×
1423
                )
×
1424
                if err != nil {
×
1425
                        return fmt.Errorf("unable to upsert channel(%d) extra "+
×
1426
                                "signed field(%v): %w", edge.ChannelID,
×
1427
                                tlvType, err)
×
1428
                }
×
1429
        }
1430

1431
        return nil
×
1432
}
1433

1434
// maybeCreateShellNode checks if a shell node entry exists for the
1435
// given public key. If it does not exist, then a new shell node entry is
1436
// created. The ID of the node is returned. A shell node only has a protocol
1437
// version and public key persisted.
1438
func maybeCreateShellNode(ctx context.Context, db SQLQueries,
1439
        pubKey route.Vertex) (int64, error) {
×
1440

×
1441
        dbNode, err := db.GetNodeByPubKey(
×
1442
                ctx, sqlc.GetNodeByPubKeyParams{
×
1443
                        PubKey:  pubKey[:],
×
1444
                        Version: int16(ProtocolV1),
×
1445
                },
×
1446
        )
×
1447
        // The node exists. Return the ID.
×
1448
        if err == nil {
×
1449
                return dbNode.ID, nil
×
1450
        } else if !errors.Is(err, sql.ErrNoRows) {
×
1451
                return 0, err
×
1452
        }
×
1453

1454
        // Otherwise, the node does not exist, so we create a shell entry for
1455
        // it.
1456
        id, err := db.UpsertNode(ctx, sqlc.UpsertNodeParams{
×
1457
                Version: int16(ProtocolV1),
×
1458
                PubKey:  pubKey[:],
×
1459
        })
×
1460
        if err != nil {
×
1461
                return 0, fmt.Errorf("unable to create shell node: %w", err)
×
1462
        }
×
1463

1464
        return id, nil
×
1465
}
1466

1467
// upsertChanPolicyExtraSignedFields updates the policy's extra signed fields in
1468
// the database. This includes deleting any existing types and then inserting
1469
// the new types.
1470
func upsertChanPolicyExtraSignedFields(ctx context.Context, db SQLQueries,
NEW
1471
        chanPolicyID int64, extraFields map[uint64][]byte) error {
×
NEW
1472

×
NEW
1473
        // Delete all existing extra signed fields for the channel policy.
×
NEW
1474
        err := db.DeleteChannelPolicyExtraTypes(ctx, chanPolicyID)
×
NEW
1475
        if err != nil {
×
NEW
1476
                return fmt.Errorf("unable to delete "+
×
NEW
1477
                        "existing policy extra signed fields for policy %d: %w",
×
NEW
1478
                        chanPolicyID, err)
×
NEW
1479
        }
×
1480

1481
        // Insert all new extra signed fields for the channel policy.
NEW
1482
        for tlvType, value := range extraFields {
×
NEW
1483
                err = db.InsertChanPolicyExtraType(
×
NEW
1484
                        ctx, sqlc.InsertChanPolicyExtraTypeParams{
×
NEW
1485
                                ChannelPolicyID: chanPolicyID,
×
NEW
1486
                                Type:            int64(tlvType),
×
NEW
1487
                                Value:           value,
×
NEW
1488
                        },
×
NEW
1489
                )
×
NEW
1490
                if err != nil {
×
NEW
1491
                        return fmt.Errorf("unable to insert "+
×
NEW
1492
                                "channel_policy(%d) extra signed field(%v): %w",
×
NEW
1493
                                chanPolicyID, tlvType, err)
×
NEW
1494
                }
×
1495
        }
1496

NEW
1497
        return nil
×
1498
}
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2025 Coveralls, Inc