paymentsdb: implement QueryPayments for sql backend

diff --git a/payments/db/sql_converters.go b/payments/db/sql_converters.go
new file mode 100644
index 0000000..fd0cad2
--- /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 ced061a..b0ce408 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 d4feafe..7888f81 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,
+	}
+}