lightning-terminal/accounts/rpcserver.go

package accounts

import (
	"context"
	"encoding/hex"
	"errors"
	"fmt"
	"sync"
	"time"

	"github.com/btcsuite/btcd/btcutil"
	"github.com/lightninglabs/lightning-terminal/litrpc"
	litmac "github.com/lightninglabs/lightning-terminal/macaroons"
	"github.com/lightningnetwork/lnd/lnrpc"
	"github.com/lightningnetwork/lnd/lnrpc/routerrpc"
	"github.com/lightningnetwork/lnd/lntypes"
	"github.com/lightningnetwork/lnd/lnwire"
	"github.com/lightningnetwork/lnd/macaroons"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"gopkg.in/macaroon-bakery.v2/bakery/checkers"
	"gopkg.in/macaroon.v2"
)

const (
	// DefaultMaxPayments is the default page size for listing payments.
	DefaultMaxPayments = 20

	// MaxPaymentsLimit is the maximum page size allowed for listing
	// payments.
	MaxPaymentsLimit = 50

	// MaxIndexOffset is the maximum index offset allowed.
	MaxIndexOffset = 0x7fffffff

	// DefaultTrackPaymentTimeout is the timeout for track payment RPC
	// calls.
	DefaultTrackPaymentTimeout = 8 * time.Second

	// MaxConcurrentTrackPayments is the maximum number of concurrent track
	// payment requests.
	MaxConcurrentTrackPayments = 10
)

var (
	// ErrServerNotActive indicates that the server has started but hasn't
	// fully finished the startup process.
	ErrServerNotActive = errors.New("accounts server is still in the " +
		"process of starting")
)

// RPCServer is the main server that implements the Accounts gRPC service.
type RPCServer struct {
	litrpc.UnimplementedAccountsServer

	service *InterceptorService

	superMacBaker litmac.Baker
}

// NewRPCServer returns a new RPC server for the given service.
func NewRPCServer() *RPCServer {
	return &RPCServer{}
}

// Start adds the necessary dependencies for the RPCServer to be able to process
// requests, and starts the RPCServer.
func (s *RPCServer) Start(service *InterceptorService,
	superMacBaker litmac.Baker) {

	s.service = service
	s.superMacBaker = superMacBaker
}

// CreateAccount adds an entry to the account database. This entry represents
// an amount of satoshis (account balance) that can be spent using off-chain
// transactions (e.g. paying invoices).
//
// Macaroons can be created to be locked to an account. This makes sure that
// the bearer of the macaroon can only spend at most that amount of satoshis
// through the daemon that has issued the macaroon.
//
// Accounts only assert a maximum amount spendable. Having a certain account
// balance does not guarantee that the node has the channel liquidity to
// actually spend that amount.
func (s *RPCServer) CreateAccount(ctx context.Context,
	req *litrpc.CreateAccountRequest) (*litrpc.CreateAccountResponse,
	error) {

	log.Infof("[createaccount] label=%v, balance=%d, expiration=%d",
		req.Label, req.AccountBalance, req.ExpirationDate)

	var (
		balanceMsat    lnwire.MilliSatoshi
		expirationDate time.Time
	)

	// If the expiration date was set, parse it as a unix time stamp.
	// Otherwise, we leave it nil to indicate the account has no expiration
	// date.
	if req.ExpirationDate > 0 {
		expirationDate = time.Unix(req.ExpirationDate, 0)
	}

	// Convert from satoshis to millisatoshis for storage.
	balance := btcutil.Amount(req.AccountBalance)
	balanceMsat = lnwire.NewMSatFromSatoshis(balance)

	// Create the actual account in the macaroon account store.
	account, err := s.service.NewAccount(
		ctx, balanceMsat, expirationDate, req.Label,
	)
	if err != nil {
		return nil, fmt.Errorf("unable to create account: %w", err)
	}

	var rootKeyIdSuffix [4]byte
	copy(rootKeyIdSuffix[:], account.ID[0:4])
	macRootKey := litmac.NewSuperMacaroonRootKeyID(rootKeyIdSuffix)

	accountCaveat := checkers.Condition(
		macaroons.CondLndCustom,
		fmt.Sprintf("%s %x", CondAccount, account.ID[:]),
	)

	macHex, err := s.superMacBaker(
		ctx, macRootKey, MacaroonPermissions,
		[]macaroon.Caveat{{Id: []byte(accountCaveat)}},
	)
	if err != nil {
		return nil, fmt.Errorf("error baking account macaroon: %w", err)
	}

	macBytes, err := hex.DecodeString(macHex)
	if err != nil {
		return nil, fmt.Errorf("error decoding account macaroon: %w",
			err)
	}

	return &litrpc.CreateAccountResponse{
		Account:  marshalAccount(account),
		Macaroon: macBytes,
	}, nil
}

// UpdateAccount updates an existing account in the account database.
func (s *RPCServer) UpdateAccount(ctx context.Context,
	req *litrpc.UpdateAccountRequest) (*litrpc.Account, error) {

	log.Infof("[updateaccount] id=%s, label=%v, balance=%d, "+
		"expiration=%d, new_label=%v", req.Id, req.Label,
		req.AccountBalance, req.ExpirationDate, req.NewLabel)

	accountID, err := s.findAccount(ctx, req.Id, req.Label)
	if err != nil {
		return nil, err
	}

	// Ask the service to update the account.
	account, err := s.service.UpdateAccount(
		ctx, accountID, btcutil.Amount(req.AccountBalance),
		req.ExpirationDate, req.NewLabel,
	)
	if err != nil {
		return nil, err
	}

	return marshalAccount(account), nil
}

// CreditAccount increases the balance of an existing account in the account
// database, by the given amount.
func (s *RPCServer) CreditAccount(ctx context.Context,
	req *litrpc.CreditAccountRequest) (*litrpc.CreditAccountResponse,
	error) {

	if req.GetAccount() == nil {
		return nil, fmt.Errorf("account param must be specified")
	}

	var id, label string

	switch idType := req.Account.Identifier.(type) {
	case *litrpc.AccountIdentifier_Id:
		id = idType.Id
	case *litrpc.AccountIdentifier_Label:
		label = idType.Label
	}

	log.Infof("[creditaccount] id=%s, label=%v, amount=%d", id, label,
		req.Amount)

	amount := lnwire.MilliSatoshi(req.Amount * 1000)

	accountID, err := s.findAccount(ctx, id, label)
	if err != nil {
		return nil, err
	}

	account, err := s.service.CreditAccount(ctx, accountID, amount)
	if err != nil {
		return nil, err
	}

	return &litrpc.CreditAccountResponse{
		Account: marshalAccount(account),
	}, nil
}

// DebitAccount decreases the balance of an existing account in the account
// database, by the given amount.
func (s *RPCServer) DebitAccount(ctx context.Context,
	req *litrpc.DebitAccountRequest) (*litrpc.DebitAccountResponse, error) {

	if req.GetAccount() == nil {
		return nil, fmt.Errorf("account param must be specified")
	}

	var id, label string

	switch idType := req.Account.Identifier.(type) {
	case *litrpc.AccountIdentifier_Id:
		id = idType.Id
	case *litrpc.AccountIdentifier_Label:
		label = idType.Label
	}

	log.Infof("[debitaccount] id=%s, label=%v, amount=%d", id, label,
		req.Amount)

	amount := lnwire.MilliSatoshi(req.Amount * 1000)

	accountID, err := s.findAccount(ctx, id, label)
	if err != nil {
		return nil, err
	}

	account, err := s.service.DebitAccount(ctx, accountID, amount)
	if err != nil {
		return nil, err
	}

	return &litrpc.DebitAccountResponse{
		Account: marshalAccount(account),
	}, nil
}

// ListAccounts returns all accounts that are currently stored in the account
// database.
func (s *RPCServer) ListAccounts(ctx context.Context,
	_ *litrpc.ListAccountsRequest) (*litrpc.ListAccountsResponse, error) {

	log.Info("[listaccounts]")

	// Retrieve all accounts from the macaroon account store.
	accts, err := s.service.Accounts(ctx)
	if err != nil {
		return nil, fmt.Errorf("unable to list accounts: %w", err)
	}

	// Map the response into the proper response type and return it.
	rpcAccounts := make([]*litrpc.Account, len(accts))
	for i, acct := range accts {
		acct := acct

		rpcAccounts[i] = marshalAccount(acct)
	}

	return &litrpc.ListAccountsResponse{
		Accounts: rpcAccounts,
	}, nil
}

// AccountInfo returns the account with the given ID or label.
func (s *RPCServer) AccountInfo(ctx context.Context,
	req *litrpc.AccountInfoRequest) (*litrpc.Account, error) {

	log.Infof("[accountinfo] id=%v, label=%v", req.Id, req.Label)

	accountID, err := s.findAccount(ctx, req.Id, req.Label)
	if err != nil {
		return nil, err
	}

	dbAccount, err := s.service.Account(ctx, accountID)
	if err != nil {
		return nil, fmt.Errorf("error retrieving account: %w", err)
	}

	return marshalAccount(dbAccount), nil
}

// RemoveAccount removes the given account from the account database.
func (s *RPCServer) RemoveAccount(ctx context.Context,
	req *litrpc.RemoveAccountRequest) (*litrpc.RemoveAccountResponse,
	error) {

	log.Infof("[removeaccount] id=%v, label=%v", req.Id, req.Label)

	accountID, err := s.findAccount(ctx, req.Id, req.Label)
	if err != nil {
		return nil, err
	}

	// Now remove the account.
	err = s.service.RemoveAccount(ctx, accountID)
	if err != nil {
		return nil, fmt.Errorf("error removing account: %w", err)
	}

	return &litrpc.RemoveAccountResponse{}, nil
}

// findAccount finds an account by its ID or label.
func (s *RPCServer) findAccount(ctx context.Context, id string, label string) (
	AccountID, error) {

	switch {
	case id != "" && label != "":
		return AccountID{}, fmt.Errorf("either account ID or label " +
			"must be specified, not both")

	case id != "":
		// Account ID is always a hex string, convert it to our account
		// ID type.
		var accountID AccountID
		decoded, err := hex.DecodeString(id)
		if err != nil {
			return AccountID{}, fmt.Errorf("error decoding "+
				"account ID: %w", err)
		}
		copy(accountID[:], decoded)

		return accountID, nil

	case label != "":
		// We need to find the account by its label.
		accounts, err := s.service.Accounts(ctx)
		if err != nil {
			return AccountID{}, fmt.Errorf("unable to list "+
				"accounts: %w", err)
		}

		for _, acct := range accounts {
			if acct.Label == label {
				return acct.ID, nil
			}
		}

		return AccountID{}, fmt.Errorf("unable to find account "+
			"with label '%s'", label)

	default:
		return AccountID{}, fmt.Errorf("either account ID or label " +
			"must be specified")
	}
}

// marshalAccount converts an account into its RPC counterpart.
func marshalAccount(acct *OffChainBalanceAccount) *litrpc.Account {
	rpcAccount := &litrpc.Account{
		Id:             hex.EncodeToString(acct.ID[:]),
		InitialBalance: uint64(acct.InitialBalance.ToSatoshis()),
		CurrentBalance: acct.CurrentBalanceSats(),
		LastUpdate:     acct.LastUpdate.Unix(),
		ExpirationDate: int64(0),
		Invoices: make(
			[]*litrpc.AccountInvoice, 0, len(acct.Invoices),
		),
		Payments: make(
			[]*litrpc.AccountPayment, 0, len(acct.Payments),
		),
		Label: acct.Label,
	}

	for hash := range acct.Invoices {
		i := &litrpc.AccountInvoice{
			Hash: make([]byte, lntypes.HashSize),
		}
		copy(i.Hash, hash[:])
		rpcAccount.Invoices = append(rpcAccount.Invoices, i)
	}
	for hash, paymentEntry := range acct.Payments {
		p := &litrpc.AccountPayment{
			Hash:       make([]byte, lntypes.HashSize),
			State:      paymentEntry.Status.String(),
			FullAmount: int64(paymentEntry.FullAmount.ToSatoshis()),
		}
		copy(p.Hash, hash[:])
		rpcAccount.Payments = append(rpcAccount.Payments, p)
	}

	if !acct.ExpirationDate.IsZero() {
		rpcAccount.ExpirationDate = acct.ExpirationDate.Unix()
	}

	return rpcAccount
}

// AccountPayments returns the detailed payment history for the given account.
func (s *RPCServer) AccountPayments(ctx context.Context,
	req *litrpc.AccountPaymentsRequest) (
	*litrpc.AccountPaymentsResponse, error) {

	if req.GetAccount() == nil {
		return nil, fmt.Errorf("account param must be specified")
	}

	var id, label string
	switch idType := req.Account.Identifier.(type) {
	case *litrpc.AccountIdentifier_Id:
		id = idType.Id
	case *litrpc.AccountIdentifier_Label:
		label = idType.Label
	}

	log.Infof("[accountpayments] id=%s, label=%v, max_payments=%d, "+
		"index_offset=%d, count_total_payments=%v",
		id, label, req.MaxPayments, req.IndexOffset,
		req.CountTotalPayments)

	accountID, err := s.findAccount(ctx, id, label)
	if err != nil {
		return nil, err
	}

	// Determine limits.
	limit := req.MaxPayments
	if limit == 0 {
		limit = DefaultMaxPayments
	} else if limit > MaxPaymentsLimit {
		return nil, fmt.Errorf(
			"max_payments cannot exceed %d", MaxPaymentsLimit,
		)
	}

	if req.IndexOffset > MaxIndexOffset {
		return nil, fmt.Errorf("index_offset out of range")
	}
	offset := req.IndexOffset

	// Fetch the paginated payment entries from the store.
	paymentsFromStore, err := s.service.store.ListAccountPayments(
		ctx, accountID, int32(offset), int32(limit),
	)
	if err != nil {
		return nil, fmt.Errorf(
			"unable to list account payments: %w", err,
		)
	}

	// Fetch total payment count if requested.
	var totalNumPayments uint64
	if req.CountTotalPayments {
		total, err := s.service.store.CountAccountPayments(
			ctx, accountID,
		)
		if err != nil {
			return nil, fmt.Errorf(
				"unable to count account payments: %w", err,
			)
		}

		totalNumPayments = total
	}

	// Fetch the detailed payments concurrently from LND.
	var (
		wg       sync.WaitGroup
		mu       sync.Mutex
		payments = make(map[lntypes.Hash]*lnrpc.Payment)
		errs     []error
		sem      = make(chan struct{}, MaxConcurrentTrackPayments)
	)

	rawCtx, _, client := s.service.routerClient.RawClientWithMacAuth(ctx)

	for _, entry := range paymentsFromStore {
		entry := entry

		wg.Add(1)
		go func() {
			defer wg.Done()

			// Bounded concurrency using semaphore.
			select {
			case sem <- struct{}{}:
			case <-rawCtx.Done():
				return
			}
			defer func() { <-sem }()

			// Add a timeout to prevent hanging streaming calls.
			trackCtx, trackCancel := context.WithTimeout(
				rawCtx, DefaultTrackPaymentTimeout,
			)
			defer trackCancel()

			stream, err := client.TrackPaymentV2(
				trackCtx, &routerrpc.TrackPaymentRequest{
					PaymentHash:       entry.Hash[:],
					NoInflightUpdates: false,
				},
			)
			if err != nil {
				// Skip error logging and recording if the
				// parent context was cancelled or timed out.
				if rawCtx.Err() != nil {
					return
				}

				sErr, ok := status.FromError(err)
				if ok && sErr.Code() == codes.NotFound {
					log.Warnf("Payment %x not found in "+
						"lnd, creating placeholder: %v",
						entry.Hash[:], err)

					mu.Lock()
					payments[entry.Hash] =
						notFoundPayment(entry.Hash)
					mu.Unlock()

					return
				}

				log.Errorf("Failed to track payment %x: %v",
					entry.Hash[:], err)

				mu.Lock()
				errs = append(errs, err)
				mu.Unlock()

				return
			}

			payment, err := stream.Recv()
			if err != nil {
				// Skip error logging and recording if the
				// parent context was cancelled or timed out.
				if rawCtx.Err() != nil {
					return
				}

				sErr, ok := status.FromError(err)
				if ok && sErr.Code() == codes.NotFound {
					log.Warnf("Payment %x not found in "+
						"lnd, creating placeholder: %v",
						entry.Hash[:], err)

					mu.Lock()
					payments[entry.Hash] =
						notFoundPayment(entry.Hash)
					mu.Unlock()

					return
				}

				log.Errorf("Failed to receive payment "+
					"update for %x: %v", entry.Hash[:], err)

				mu.Lock()
				errs = append(errs, err)
				mu.Unlock()

				return
			}

			mu.Lock()
			payments[entry.Hash] = payment
			mu.Unlock()
		}()
	}

	wg.Wait()

	// Return immediately if the parent context was cancelled or timed out.
	if err := rawCtx.Err(); err != nil {
		return nil, err
	}

	// If there were any errors tracking the payments, return the
	// error.
	if len(errs) > 0 {
		return nil, fmt.Errorf(
			"failed to fetch payment details: %w", errs[0],
		)
	}

	var finalPayments []*lnrpc.Payment
	for _, entry := range paymentsFromStore {
		if p, ok := payments[entry.Hash]; ok {
			finalPayments = append(finalPayments, p)
		}
	}

	var firstIndexOffset, lastIndexOffset uint64
	if len(finalPayments) > 0 {
		firstIndexOffset = offset
		lastIndexOffset = offset + uint64(len(finalPayments))
	}

	return &litrpc.AccountPaymentsResponse{
		Payments:         finalPayments,
		FirstIndexOffset: firstIndexOffset,
		LastIndexOffset:  lastIndexOffset,
		TotalNumPayments: totalNumPayments,
	}, nil
}

// notFoundPayment creates a placeholder payment for a desynced entry.
func notFoundPayment(hash lntypes.Hash) *lnrpc.Payment {
	return &lnrpc.Payment{
		PaymentHash: hex.EncodeToString(hash[:]),
		Status:      lnrpc.Payment_UNKNOWN,
		FailureReason: lnrpc.
			PaymentFailureReason_FAILURE_REASON_NONE,
	}
}