From e8fe45fe65756392feb4604b48c0c20b6efcbba1 Mon Sep 17 00:00:00 2001 From: ziggie Date: Sun, 12 Oct 2025 19:15:49 +0200 Subject: [PATCH] paymentsdb: implement QueryPayments for sql backend --- payments/db/sql_converters.go | 272 ++++++++++++++++ payments/db/sql_store.go | 569 ++++++++++++++++++++++++++++++++++ sqldb/sqlc/db_custom.go | 76 +++++ 3 files changed, 917 insertions(+) create mode 100644 payments/db/sql_converters.go diff --git a/payments/db/sql_converters.go b/payments/db/sql_converters.go new file mode 100644 index 000000000..fd0cad2dc --- /dev/null +++ b/payments/db/sql_converters.go @@ -0,0 +1,272 @@ +package paymentsdb + +import ( + "bytes" + "fmt" + "strconv" + "time" + + "github.com/btcsuite/btcd/btcec/v2" + "github.com/lightningnetwork/lnd/lntypes" + "github.com/lightningnetwork/lnd/lnwire" + "github.com/lightningnetwork/lnd/record" + "github.com/lightningnetwork/lnd/routing/route" + "github.com/lightningnetwork/lnd/sqldb/sqlc" + "github.com/lightningnetwork/lnd/tlv" +) + +// dbPaymentToCreationInfo converts database payment data to the +// PaymentCreationInfo struct. +func dbPaymentToCreationInfo(paymentIdentifier []byte, amountMsat int64, + createdAt time.Time, intentPayload []byte, + firstHopCustomRecords lnwire.CustomRecords) *PaymentCreationInfo { + + // This is the payment hash for non-AMP payments and the SetID for AMP + // payments. + var identifier lntypes.Hash + copy(identifier[:], paymentIdentifier) + + return &PaymentCreationInfo{ + PaymentIdentifier: identifier, + Value: lnwire.MilliSatoshi(amountMsat), + CreationTime: createdAt.Local(), + PaymentRequest: intentPayload, + FirstHopCustomRecords: firstHopCustomRecords, + } +} + +// dbAttemptToHTLCAttempt converts a database HTLC attempt to an HTLCAttempt. +func dbAttemptToHTLCAttempt(dbAttempt sqlc.FetchHtlcAttemptsForPaymentsRow, + hops []sqlc.FetchHopsForAttemptsRow, + hopCustomRecords map[int64][]sqlc.PaymentHopCustomRecord, + routeCustomRecords []sqlc.PaymentAttemptFirstHopCustomRecord) ( + *HTLCAttempt, error) { + + // Convert route-level first hop custom records to CustomRecords map. + var firstHopWireCustomRecords lnwire.CustomRecords + if len(routeCustomRecords) > 0 { + firstHopWireCustomRecords = make(lnwire.CustomRecords) + for _, record := range routeCustomRecords { + firstHopWireCustomRecords[uint64(record.Key)] = + record.Value + } + } + + // Build the route from the database data. + route, err := dbDataToRoute( + hops, hopCustomRecords, dbAttempt.FirstHopAmountMsat, + dbAttempt.RouteTotalTimeLock, dbAttempt.RouteTotalAmount, + dbAttempt.RouteSourceKey, firstHopWireCustomRecords, + ) + if err != nil { + return nil, fmt.Errorf("failed to convert to route: %w", + err) + } + + hash, err := lntypes.MakeHash(dbAttempt.PaymentHash) + if err != nil { + return nil, fmt.Errorf("failed to parse payment "+ + "hash: %w", err) + } + + // Create the attempt info. + var sessionKey [32]byte + copy(sessionKey[:], dbAttempt.SessionKey) + + info := HTLCAttemptInfo{ + AttemptID: uint64(dbAttempt.AttemptIndex), + sessionKey: sessionKey, + Route: *route, + AttemptTime: dbAttempt.AttemptTime, + Hash: &hash, + } + + attempt := &HTLCAttempt{ + HTLCAttemptInfo: info, + } + + // If there's no resolution type, the attempt is still in-flight. + // Return early without processing settlement or failure info. + if !dbAttempt.ResolutionType.Valid { + return attempt, nil + } + + // Add settlement info if present. + if HTLCAttemptResolutionType(dbAttempt.ResolutionType.Int32) == + HTLCAttemptResolutionSettled { + + var preimage lntypes.Preimage + copy(preimage[:], dbAttempt.SettlePreimage) + + attempt.Settle = &HTLCSettleInfo{ + Preimage: preimage, + SettleTime: dbAttempt.ResolutionTime.Time, + } + } + + // Add failure info if present. + if HTLCAttemptResolutionType(dbAttempt.ResolutionType.Int32) == + HTLCAttemptResolutionFailed { + + failure := &HTLCFailInfo{ + FailTime: dbAttempt.ResolutionTime.Time, + } + + if dbAttempt.HtlcFailReason.Valid { + failure.Reason = HTLCFailReason( + dbAttempt.HtlcFailReason.Int32, + ) + } + + if dbAttempt.FailureSourceIndex.Valid { + failure.FailureSourceIndex = uint32( + dbAttempt.FailureSourceIndex.Int32, + ) + } + + // Decode the failure message if present. + if len(dbAttempt.FailureMsg) > 0 { + msg, err := lnwire.DecodeFailureMessage( + bytes.NewReader(dbAttempt.FailureMsg), 0, + ) + if err != nil { + return nil, fmt.Errorf("failed to decode "+ + "failure message: %w", err) + } + failure.Message = msg + } + + attempt.Failure = failure + } + + return attempt, nil +} + +// dbDataToRoute converts database route data to a route.Route. +func dbDataToRoute(hops []sqlc.FetchHopsForAttemptsRow, + hopCustomRecords map[int64][]sqlc.PaymentHopCustomRecord, + firstHopAmountMsat int64, totalTimeLock int32, totalAmount int64, + sourceKey []byte, firstHopWireCustomRecords lnwire.CustomRecords) ( + *route.Route, error) { + + if len(hops) == 0 { + return nil, fmt.Errorf("no hops provided") + } + + // Hops are already sorted by hop_index from the SQL query. + routeHops := make([]*route.Hop, len(hops)) + + for i, hop := range hops { + pubKey, err := route.NewVertexFromBytes(hop.PubKey) + if err != nil { + return nil, fmt.Errorf("failed to parse pub key: %w", + err) + } + + var channelID uint64 + if hop.Scid != "" { + // The SCID is stored as a string representation + // of the uint64. + var err error + channelID, err = strconv.ParseUint(hop.Scid, 10, 64) + if err != nil { + return nil, fmt.Errorf("failed to parse "+ + "scid: %w", err) + } + } + + routeHop := &route.Hop{ + PubKeyBytes: pubKey, + ChannelID: channelID, + OutgoingTimeLock: uint32(hop.OutgoingTimeLock), + AmtToForward: lnwire.MilliSatoshi(hop.AmtToForward), + } + + // Add MPP record if present. + if len(hop.MppPaymentAddr) > 0 { + var paymentAddr [32]byte + copy(paymentAddr[:], hop.MppPaymentAddr) + routeHop.MPP = record.NewMPP( + lnwire.MilliSatoshi(hop.MppTotalMsat.Int64), + paymentAddr, + ) + } + + // Add AMP record if present. + if len(hop.AmpRootShare) > 0 { + var rootShare [32]byte + copy(rootShare[:], hop.AmpRootShare) + var setID [32]byte + copy(setID[:], hop.AmpSetID) + + routeHop.AMP = record.NewAMP( + rootShare, setID, + uint32(hop.AmpChildIndex.Int32), + ) + } + + // Add blinding point if present (only for introduction node). + if len(hop.BlindingPoint) > 0 { + pubKey, err := btcec.ParsePubKey(hop.BlindingPoint) + if err != nil { + return nil, fmt.Errorf("failed to parse "+ + "blinding point: %w", err) + } + routeHop.BlindingPoint = pubKey + } + + // Add encrypted data if present (for all blinded hops). + if len(hop.EncryptedData) > 0 { + routeHop.EncryptedData = hop.EncryptedData + } + + // Add total amount if present (only for final hop in blinded + // route). + if hop.BlindedPathTotalAmt.Valid { + routeHop.TotalAmtMsat = lnwire.MilliSatoshi( + hop.BlindedPathTotalAmt.Int64, + ) + } + + // Add hop-level custom records. + if records, ok := hopCustomRecords[hop.ID]; ok { + routeHop.CustomRecords = make( + record.CustomSet, + ) + for _, rec := range records { + routeHop.CustomRecords[uint64(rec.Key)] = + rec.Value + } + } + + // Add metadata if present. + if len(hop.MetaData) > 0 { + routeHop.Metadata = hop.MetaData + } + + routeHops[i] = routeHop + } + + // Parse the source node public key. + var sourceNode route.Vertex + copy(sourceNode[:], sourceKey) + + route := &route.Route{ + TotalTimeLock: uint32(totalTimeLock), + TotalAmount: lnwire.MilliSatoshi(totalAmount), + SourcePubKey: sourceNode, + Hops: routeHops, + FirstHopWireCustomRecords: firstHopWireCustomRecords, + } + + // Set the first hop amount if it is set. + if firstHopAmountMsat != 0 { + route.FirstHopAmount = tlv.NewRecordT[tlv.TlvType0]( + tlv.NewBigSizeT(lnwire.MilliSatoshi( + firstHopAmountMsat, + )), + ) + } + + return route, nil +} diff --git a/payments/db/sql_store.go b/payments/db/sql_store.go index ced061afa..b0ce408f9 100644 --- a/payments/db/sql_store.go +++ b/payments/db/sql_store.go @@ -2,14 +2,40 @@ package paymentsdb import ( "context" + "errors" "fmt" + "math" + "time" + "github.com/lightningnetwork/lnd/lnwire" "github.com/lightningnetwork/lnd/sqldb" "github.com/lightningnetwork/lnd/sqldb/sqlc" ) +// PaymentIntentType represents the type of payment intent. +type PaymentIntentType int16 + +const ( + // PaymentIntentTypeBolt11 indicates a BOLT11 invoice payment. + PaymentIntentTypeBolt11 PaymentIntentType = 0 +) + +// HTLCAttemptResolutionType represents the type of HTLC attempt resolution. +type HTLCAttemptResolutionType int32 + +const ( + // HTLCAttemptResolutionSettled indicates the HTLC attempt was settled + // successfully with a preimage. + HTLCAttemptResolutionSettled HTLCAttemptResolutionType = 1 + + // HTLCAttemptResolutionFailed indicates the HTLC attempt failed. + HTLCAttemptResolutionFailed HTLCAttemptResolutionType = 2 +) + // SQLQueries is a subset of the sqlc.Querier interface that can be used to // execute queries against the SQL payments tables. +// +//nolint:ll type SQLQueries interface { /* Payment DB read operations. @@ -83,3 +109,546 @@ func NewSQLStore(cfg *SQLStoreConfig, db BatchedSQLQueries, // A compile-time constraint to ensure SQLStore implements DB. var _ DB = (*SQLStore)(nil) + +// fetchPaymentWithCompleteData fetches a payment with all its related data +// including attempts, hops, and custom records from the database. +// This is a convenience wrapper around the batch loading functions for single +// payment operations. +func (s *SQLStore) fetchPaymentWithCompleteData(ctx context.Context, + db SQLQueries, dbPayment sqlc.PaymentAndIntent) (*MPPayment, error) { + + payment := dbPayment.GetPayment() + + // Load batch data for this single payment. + batchData, err := s.loadPaymentsBatchData(ctx, db, []int64{payment.ID}) + if err != nil { + return nil, fmt.Errorf("failed to load batch data: %w", err) + } + + // Build the payment from the batch data. + return s.buildPaymentFromBatchData(dbPayment, batchData) +} + +// paymentsBatchData holds all the batch-loaded data for multiple payments. +type paymentsBatchData struct { + // paymentCustomRecords maps payment ID to its custom records. + paymentCustomRecords map[int64][]sqlc.PaymentFirstHopCustomRecord + + // attempts maps payment ID to its HTLC attempts. + attempts map[int64][]sqlc.FetchHtlcAttemptsForPaymentsRow + + // hopsByAttempt maps attempt index to its hops. + hopsByAttempt map[int64][]sqlc.FetchHopsForAttemptsRow + + // hopCustomRecords maps hop ID to its custom records. + hopCustomRecords map[int64][]sqlc.PaymentHopCustomRecord + + // routeCustomRecords maps attempt index to its route-level custom + // records. + routeCustomRecords map[int64][]sqlc.PaymentAttemptFirstHopCustomRecord +} + +// loadPaymentCustomRecords loads payment-level custom records for a given +// set of payment IDs. +func (s *SQLStore) loadPaymentCustomRecords(ctx context.Context, + db SQLQueries, paymentIDs []int64, + batchData *paymentsBatchData) error { + + return sqldb.ExecuteBatchQuery( + ctx, s.cfg.QueryCfg, paymentIDs, + func(id int64) int64 { return id }, + func(ctx context.Context, ids []int64) ( + []sqlc.PaymentFirstHopCustomRecord, error) { + + //nolint:ll + records, err := db.FetchPaymentLevelFirstHopCustomRecords( + ctx, ids, + ) + + return records, err + }, + func(ctx context.Context, + record sqlc.PaymentFirstHopCustomRecord) error { + + paymentRecords := + batchData.paymentCustomRecords[record.PaymentID] + + batchData.paymentCustomRecords[record.PaymentID] = + append(paymentRecords, record) + + return nil + }, + ) +} + +// loadHtlcAttempts loads HTLC attempts for all payments and returns all +// attempt indices. +func (s *SQLStore) loadHtlcAttempts(ctx context.Context, db SQLQueries, + paymentIDs []int64, batchData *paymentsBatchData) ([]int64, error) { + + var allAttemptIndices []int64 + + err := sqldb.ExecuteBatchQuery( + ctx, s.cfg.QueryCfg, paymentIDs, + func(id int64) int64 { return id }, + func(ctx context.Context, ids []int64) ( + []sqlc.FetchHtlcAttemptsForPaymentsRow, error) { + + return db.FetchHtlcAttemptsForPayments(ctx, ids) + }, + func(ctx context.Context, + attempt sqlc.FetchHtlcAttemptsForPaymentsRow) error { + + batchData.attempts[attempt.PaymentID] = append( + batchData.attempts[attempt.PaymentID], attempt, + ) + allAttemptIndices = append( + allAttemptIndices, attempt.AttemptIndex, + ) + + return nil + }, + ) + + return allAttemptIndices, err +} + +// loadHopsForAttempts loads hops for all attempts and returns all hop IDs. +func (s *SQLStore) loadHopsForAttempts(ctx context.Context, db SQLQueries, + attemptIndices []int64, batchData *paymentsBatchData) ([]int64, error) { + + var hopIDs []int64 + + err := sqldb.ExecuteBatchQuery( + ctx, s.cfg.QueryCfg, attemptIndices, + func(idx int64) int64 { return idx }, + func(ctx context.Context, indices []int64) ( + []sqlc.FetchHopsForAttemptsRow, error) { + + return db.FetchHopsForAttempts(ctx, indices) + }, + func(ctx context.Context, + hop sqlc.FetchHopsForAttemptsRow) error { + + attemptHops := + batchData.hopsByAttempt[hop.HtlcAttemptIndex] + + batchData.hopsByAttempt[hop.HtlcAttemptIndex] = + append(attemptHops, hop) + + hopIDs = append(hopIDs, hop.ID) + + return nil + }, + ) + + return hopIDs, err +} + +// loadHopCustomRecords loads hop-level custom records for all hops. +func (s *SQLStore) loadHopCustomRecords(ctx context.Context, db SQLQueries, + hopIDs []int64, batchData *paymentsBatchData) error { + + return sqldb.ExecuteBatchQuery( + ctx, s.cfg.QueryCfg, hopIDs, + func(id int64) int64 { return id }, + func(ctx context.Context, ids []int64) ( + []sqlc.PaymentHopCustomRecord, error) { + + return db.FetchHopLevelCustomRecords(ctx, ids) + }, + func(ctx context.Context, + record sqlc.PaymentHopCustomRecord) error { + + // TODO(ziggie): Can we get rid of this? + // This has to be in place otherwise the + // comparison will not match. + if record.Value == nil { + record.Value = []byte{} + } + + batchData.hopCustomRecords[record.HopID] = append( + batchData.hopCustomRecords[record.HopID], + record, + ) + + return nil + }, + ) +} + +// loadRouteCustomRecords loads route-level first hop custom records for all +// attempts. +func (s *SQLStore) loadRouteCustomRecords(ctx context.Context, db SQLQueries, + attemptIndices []int64, batchData *paymentsBatchData) error { + + return sqldb.ExecuteBatchQuery( + ctx, s.cfg.QueryCfg, attemptIndices, + func(idx int64) int64 { return idx }, + func(ctx context.Context, indices []int64) ( + []sqlc.PaymentAttemptFirstHopCustomRecord, error) { + + return db.FetchRouteLevelFirstHopCustomRecords( + ctx, indices, + ) + }, + func(ctx context.Context, + record sqlc.PaymentAttemptFirstHopCustomRecord) error { + + idx := record.HtlcAttemptIndex + attemptRecords := batchData.routeCustomRecords[idx] + + batchData.routeCustomRecords[idx] = + append(attemptRecords, record) + + return nil + }, + ) +} + +// loadPaymentsBatchData loads all related data for multiple payments in batch. +func (s *SQLStore) loadPaymentsBatchData(ctx context.Context, db SQLQueries, + paymentIDs []int64) (*paymentsBatchData, error) { + + batchData := &paymentsBatchData{ + paymentCustomRecords: make( + map[int64][]sqlc.PaymentFirstHopCustomRecord, + ), + attempts: make( + map[int64][]sqlc.FetchHtlcAttemptsForPaymentsRow, + ), + hopsByAttempt: make( + map[int64][]sqlc.FetchHopsForAttemptsRow, + ), + hopCustomRecords: make( + map[int64][]sqlc.PaymentHopCustomRecord, + ), + routeCustomRecords: make( + map[int64][]sqlc.PaymentAttemptFirstHopCustomRecord, + ), + } + + if len(paymentIDs) == 0 { + return batchData, nil + } + + // Load payment-level custom records. + err := s.loadPaymentCustomRecords(ctx, db, paymentIDs, batchData) + if err != nil { + return nil, fmt.Errorf("failed to fetch payment custom "+ + "records: %w", err) + } + + // Load HTLC attempts and collect attempt indices. + allAttemptIndices, err := s.loadHtlcAttempts( + ctx, db, paymentIDs, batchData, + ) + if err != nil { + return nil, fmt.Errorf("failed to fetch HTLC attempts: %w", + err) + } + + if len(allAttemptIndices) == 0 { + // No attempts, return early. + return batchData, nil + } + + // Load hops for all attempts and collect hop IDs. + hopIDs, err := s.loadHopsForAttempts( + ctx, db, allAttemptIndices, batchData, + ) + if err != nil { + return nil, fmt.Errorf("failed to fetch hops for attempts: %w", + err) + } + + // Load hop-level custom records if there are any hops. + if len(hopIDs) > 0 { + err = s.loadHopCustomRecords(ctx, db, hopIDs, batchData) + if err != nil { + return nil, fmt.Errorf("failed to fetch hop custom "+ + "records: %w", err) + } + } + + // Load route-level first hop custom records. + err = s.loadRouteCustomRecords(ctx, db, allAttemptIndices, batchData) + if err != nil { + return nil, fmt.Errorf("failed to fetch route custom "+ + "records: %w", err) + } + + return batchData, nil +} + +// buildPaymentFromBatchData builds a complete MPPayment from a database payment +// and pre-loaded batch data. +func (s *SQLStore) buildPaymentFromBatchData(dbPayment sqlc.PaymentAndIntent, + batchData *paymentsBatchData) (*MPPayment, error) { + + // The query will only return BOLT 11 payment intents or intents with + // no intent type set. + paymentIntent := dbPayment.GetPaymentIntent() + paymentRequest := paymentIntent.IntentPayload + + payment := dbPayment.GetPayment() + + // Get payment-level custom records from batch data. + customRecords := batchData.paymentCustomRecords[payment.ID] + + // Convert to the FirstHopCustomRecords map. + var firstHopCustomRecords lnwire.CustomRecords + if len(customRecords) > 0 { + firstHopCustomRecords = make(lnwire.CustomRecords) + for _, record := range customRecords { + firstHopCustomRecords[uint64(record.Key)] = record.Value + } + } + + // Convert database payment data to the PaymentCreationInfo struct. + info := dbPaymentToCreationInfo( + payment.PaymentIdentifier, payment.AmountMsat, + payment.CreatedAt, paymentRequest, firstHopCustomRecords, + ) + + // Get all HTLC attempts from batch data for a given payment. + dbAttempts := batchData.attempts[payment.ID] + + // Convert all attempts to HTLCAttempt structs using the pre-loaded + // batch data. + attempts := make([]HTLCAttempt, 0, len(dbAttempts)) + for _, dbAttempt := range dbAttempts { + attemptIndex := dbAttempt.AttemptIndex + // Convert the batch row type to the single row type. + attempt, err := dbAttemptToHTLCAttempt( + dbAttempt, batchData.hopsByAttempt[attemptIndex], + batchData.hopCustomRecords, + batchData.routeCustomRecords[attemptIndex], + ) + if err != nil { + return nil, fmt.Errorf("failed to convert attempt "+ + "%d: %w", attemptIndex, err) + } + attempts = append(attempts, *attempt) + } + + // Set the failure reason if present. + // + // TODO(ziggie): Rename it to Payment Memo in the database? + var failureReason *FailureReason + if payment.FailReason.Valid { + reason := FailureReason(payment.FailReason.Int32) + failureReason = &reason + } + + mpPayment := &MPPayment{ + SequenceNum: uint64(payment.ID), + Info: info, + HTLCs: attempts, + FailureReason: failureReason, + } + + // The status and state will be determined by calling + // SetState after construction. + if err := mpPayment.SetState(); err != nil { + return nil, fmt.Errorf("failed to set payment state: %w", err) + } + + return mpPayment, nil +} + +// QueryPayments queries and retrieves payments from the database with support +// for filtering, pagination, and efficient batch loading of related data. +// +// The function accepts a Query parameter that controls: +// - Pagination: IndexOffset specifies where to start (exclusive), and +// MaxPayments limits the number of results returned +// - Ordering: Reversed flag determines if results are returned in reverse +// chronological order +// - Filtering: CreationDateStart/End filter by creation time, and +// IncludeIncomplete controls whether non-succeeded payments are included +// - Metadata: CountTotal flag determines if the total payment count should +// be calculated +// +// The function optimizes performance by loading all related data (HTLCs, +// sequences, failure reasons, etc.) for multiple payments in a single batch +// query, rather than fetching each payment's data individually. +// +// Returns a Response containing: +// - Payments: the list of matching payments with complete data +// - FirstIndexOffset/LastIndexOffset: pagination cursors for the first and +// last payment in the result set +// - TotalCount: total number of payments in the database (if CountTotal was +// requested, otherwise 0) +// +// This is part of the DB interface. +func (s *SQLStore) QueryPayments(ctx context.Context, query Query) (Response, + error) { + + if query.MaxPayments == 0 { + return Response{}, fmt.Errorf("max payments must be non-zero") + } + + var ( + allPayments []*MPPayment + totalCount int64 + initialCursor int64 + ) + + extractCursor := func( + row sqlc.FilterPaymentsRow) int64 { + + return row.Payment.ID + } + + err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { + // We first count all payments to determine the total count + // if requested. + if query.CountTotal { + totalPayments, err := db.CountPayments(ctx) + if err != nil { + return fmt.Errorf("failed to count "+ + "payments: %w", err) + } + totalCount = totalPayments + } + + // collectFunc extracts the payment ID from each payment row. + collectFunc := func(row sqlc.FilterPaymentsRow) (int64, + error) { + + return row.Payment.ID, nil + } + + // batchDataFunc loads all related data for a batch of payments. + batchDataFunc := func(ctx context.Context, paymentIDs []int64) ( + *paymentsBatchData, error) { + + return s.loadPaymentsBatchData(ctx, db, paymentIDs) + } + + // processPayment processes each payment with the batch-loaded + // data. + processPayment := func(ctx context.Context, + dbPayment sqlc.FilterPaymentsRow, + batchData *paymentsBatchData) error { + + // Build the payment from the pre-loaded batch data. + mpPayment, err := s.buildPaymentFromBatchData( + dbPayment, batchData, + ) + if err != nil { + return fmt.Errorf("failed to fetch payment "+ + "with complete data: %w", err) + } + + // To keep compatibility with the old API, we only + // return non-succeeded payments if requested. + if mpPayment.Status != StatusSucceeded && + !query.IncludeIncomplete { + + return nil + } + + if uint64(len(allPayments)) >= query.MaxPayments { + return errMaxPaymentsReached + } + + allPayments = append(allPayments, mpPayment) + + return nil + } + + queryFunc := func(ctx context.Context, lastID int64, + limit int32) ([]sqlc.FilterPaymentsRow, error) { + + filterParams := sqlc.FilterPaymentsParams{ + NumLimit: limit, + Reverse: query.Reversed, + // For now there only BOLT 11 payment intents + // exist. + IntentType: sqldb.SQLInt16( + PaymentIntentTypeBolt11, + ), + } + + if query.Reversed { + filterParams.IndexOffsetLet = sqldb.SQLInt64( + lastID, + ) + } else { + filterParams.IndexOffsetGet = sqldb.SQLInt64( + lastID, + ) + } + + // Add potential date filters if specified. + if query.CreationDateStart != 0 { + filterParams.CreatedAfter = sqldb.SQLTime( + time.Unix(query.CreationDateStart, 0). + UTC(), + ) + } + if query.CreationDateEnd != 0 { + filterParams.CreatedBefore = sqldb.SQLTime( + time.Unix(query.CreationDateEnd, 0). + UTC(), + ) + } + + return db.FilterPayments(ctx, filterParams) + } + + if query.Reversed { + if query.IndexOffset == 0 { + initialCursor = int64(math.MaxInt64) + } else { + initialCursor = int64(query.IndexOffset) + } + } else { + initialCursor = int64(query.IndexOffset) + } + + return sqldb.ExecuteCollectAndBatchWithSharedDataQuery( + ctx, s.cfg.QueryCfg, initialCursor, queryFunc, + extractCursor, collectFunc, batchDataFunc, + processPayment, + ) + }, func() { + allPayments = nil + }) + + // We make sure we don't return an error if we reached the maximum + // number of payments. Which is the pagination limit for the query + // itself. + if err != nil && !errors.Is(err, errMaxPaymentsReached) { + return Response{}, fmt.Errorf("failed to query payments: %w", + err) + } + + // Handle case where no payments were found + if len(allPayments) == 0 { + return Response{ + Payments: allPayments, + FirstIndexOffset: 0, + LastIndexOffset: 0, + TotalCount: uint64(totalCount), + }, nil + } + + // If the query was reversed, we need to reverse the payment list + // to match the kvstore behavior and return payments in forward order. + if query.Reversed { + for i, j := 0, len(allPayments)-1; i < j; i, j = i+1, j-1 { + allPayments[i], allPayments[j] = allPayments[j], + allPayments[i] + } + } + + return Response{ + Payments: allPayments, + FirstIndexOffset: allPayments[0].SequenceNum, + LastIndexOffset: allPayments[len(allPayments)-1].SequenceNum, + TotalCount: uint64(totalCount), + }, nil +} diff --git a/sqldb/sqlc/db_custom.go b/sqldb/sqlc/db_custom.go index d4feafe21..7888f81f4 100644 --- a/sqldb/sqlc/db_custom.go +++ b/sqldb/sqlc/db_custom.go @@ -167,3 +167,79 @@ func (r GetChannelsBySCIDRangeRow) Node1Pub() []byte { func (r GetChannelsBySCIDRangeRow) Node2Pub() []byte { return r.Node2PubKey } + +// PaymentAndIntent is an interface that provides access to a payment and its +// associated payment intent. +type PaymentAndIntent interface { + // GetPayment returns the Payment associated with this interface. + GetPayment() Payment + + // GetPaymentIntent returns the PaymentIntent associated with this payment. + GetPaymentIntent() PaymentIntent +} + +// GetPayment returns the Payment associated with this interface. +// +// NOTE: This method is part of the PaymentAndIntent interface. +func (r FilterPaymentsRow) GetPayment() Payment { + return r.Payment +} + +// GetPaymentIntent returns the PaymentIntent associated with this payment. +// If the payment has no intent (IntentType is NULL), this returns a zero-value +// PaymentIntent. +// +// NOTE: This method is part of the PaymentAndIntent interface. +func (r FilterPaymentsRow) GetPaymentIntent() PaymentIntent { + if !r.IntentType.Valid { + return PaymentIntent{} + } + return PaymentIntent{ + IntentType: r.IntentType.Int16, + IntentPayload: r.IntentPayload, + } +} + +// GetPayment returns the Payment associated with this interface. +// +// NOTE: This method is part of the PaymentAndIntent interface. +func (r FetchPaymentRow) GetPayment() Payment { + return r.Payment +} + +// GetPaymentIntent returns the PaymentIntent associated with this payment. +// If the payment has no intent (IntentType is NULL), this returns a zero-value +// PaymentIntent. +// +// NOTE: This method is part of the PaymentAndIntent interface. +func (r FetchPaymentRow) GetPaymentIntent() PaymentIntent { + if !r.IntentType.Valid { + return PaymentIntent{} + } + return PaymentIntent{ + IntentType: r.IntentType.Int16, + IntentPayload: r.IntentPayload, + } +} + +// GetPayment returns the Payment associated with this interface. +// +// NOTE: This method is part of the PaymentAndIntent interface. +func (r FetchPaymentsByIDsRow) GetPayment() Payment { + return r.Payment +} + +// GetPaymentIntent returns the PaymentIntent associated with this payment. +// If the payment has no intent (IntentType is NULL), this returns a zero-value +// PaymentIntent. +// +// NOTE: This method is part of the PaymentAndIntent interface. +func (r FetchPaymentsByIDsRow) GetPaymentIntent() PaymentIntent { + if !r.IntentType.Valid { + return PaymentIntent{} + } + return PaymentIntent{ + IntentType: r.IntentType.Int16, + IntentPayload: r.IntentPayload, + } +}