func MarshallAccount(a *account.Account) (*poolrpc.Account, error) {
	var rpcState poolrpc.AccountState
	switch a.State {
	case account.StateInitiated, account.StatePendingOpen:
		rpcState = poolrpc.AccountState_PENDING_OPEN
	case account.StatePendingUpdate:
		rpcState = poolrpc.AccountState_PENDING_UPDATE
	case account.StateOpen:
		rpcState = poolrpc.AccountState_OPEN
	case account.StateExpired:
		rpcState = poolrpc.AccountState_EXPIRED
	case account.StatePendingClosed:
		rpcState = poolrpc.AccountState_PENDING_CLOSED
	case account.StateClosed:
		rpcState = poolrpc.AccountState_CLOSED
	case account.StateCanceledAfterRecovery:
		rpcState = poolrpc.AccountState_RECOVERY_FAILED
	case account.StatePendingBatch:
		rpcState = poolrpc.AccountState_PENDING_BATCH
	default:
		return nil, fmt.Errorf("unknown state %v", a.State)
	}
	var latestTxHash chainhash.Hash
	if a.LatestTx != nil {
		latestTxHash = a.LatestTx.TxHash()
	}
	return &poolrpc.Account{
		TraderKey: a.TraderKey.PubKey.SerializeCompressed(),
		Outpoint: &auctioneerrpc.OutPoint{
			Txid:        a.OutPoint.Hash[:],
			OutputIndex: a.OutPoint.Index,
		},
		Value:            uint64(a.Value),
		ExpirationHeight: a.Expiry,
		State:            rpcState,
		LatestTxid:       latestTxHash[:],
	}, nil
}
MarshallAccount returns the RPC representation of an account.
package pool
import (
	"bytes"
	"context"
	"encoding/hex"
	"errors"
	"fmt"
	"sync"
	"sync/atomic"
	"time"
	"github.com/btcsuite/btcd/btcec"
	"github.com/btcsuite/btcd/chaincfg/chainhash"
	"github.com/btcsuite/btcd/txscript"
	"github.com/btcsuite/btcd/wire"
	"github.com/btcsuite/btcutil"
	"github.com/davecgh/go-spew/spew"
	"github.com/lightninglabs/lndclient"
	"github.com/lightninglabs/pool/account"
	"github.com/lightninglabs/pool/auctioneer"
	"github.com/lightninglabs/pool/auctioneerrpc"
	"github.com/lightninglabs/pool/chaninfo"
	"github.com/lightninglabs/pool/clientdb"
	"github.com/lightninglabs/pool/event"
	"github.com/lightninglabs/pool/funding"
	"github.com/lightninglabs/pool/order"
	"github.com/lightninglabs/pool/poolrpc"
	"github.com/lightninglabs/pool/poolscript"
	"github.com/lightninglabs/pool/sidecar"
	"github.com/lightninglabs/pool/terms"
	"github.com/lightningnetwork/lnd/chanbackup"
	lndFunding "github.com/lightningnetwork/lnd/funding"
	"github.com/lightningnetwork/lnd/input"
	"github.com/lightningnetwork/lnd/lnrpc"
	"github.com/lightningnetwork/lnd/lnwallet/chainfee"
	"github.com/lightningnetwork/lnd/lnwire"
)
const (
	getInfoTimeout = 5 * time.Second
)
type rpcServer struct {
	started uint32  
	stopped uint32  
	bestHeight uint32
	poolrpc.UnimplementedTraderServer
	server         *Server
	lndServices    *lndclient.LndServices
	lndClient      lnrpc.LightningClient
	auctioneer     *auctioneer.Client
	accountManager *account.Manager
	orderManager   *order.Manager
	quit            chan struct{}
	wg              sync.WaitGroup
	blockNtfnCancel func()
	recoveryMutex   sync.Mutex
	recoveryPending bool
	wumboSupported bool
}
type accountStore struct {
	*clientdb.DB
}
var _ account.Store = (*accountStore)(nil)
func (s *accountStore) PendingBatch() error {
	_, err := s.DB.PendingBatchSnapshot()
	return err
}
func newRPCServer(server *Server) *rpcServer {
	accountStore := &accountStore{server.db}
	lndServices := &server.lndServices.LndServices
	return &rpcServer{
		server:      server,
		lndServices: lndServices,
		lndClient:   server.lndClient,
		auctioneer:  server.AuctioneerClient,
		accountManager: account.NewManager(&account.ManagerConfig{
			Store:          accountStore,
			Auctioneer:     server.AuctioneerClient,
			Wallet:         lndServices.WalletKit,
			Signer:         lndServices.Signer,
			ChainNotifier:  lndServices.ChainNotifier,
			TxSource:       lndServices.Client,
			TxFeeEstimator: lndServices.Client,
			TxLabelPrefix:  server.cfg.TxLabelPrefix,
		}),
		orderManager: order.NewManager(&order.ManagerConfig{
			Store:     server.db,
			AcctStore: accountStore,
			Lightning: lndServices.Client,
			Wallet:    lndServices.WalletKit,
			Signer:    lndServices.Signer,
		}),
		quit: make(chan struct{}),
	}
}
func (s *rpcServer) Start() error {
	if !atomic.CompareAndSwapUint32(&s.started, 0, 1) {
		return nil
	}
	rpcLog.Infof("Starting trader server")
	ctx := context.Background()
	lndCtx, lndCancel := context.WithTimeout(ctx, getInfoTimeout)
	defer lndCancel()
	info, err := s.lndClient.GetInfo(lndCtx, &lnrpc.GetInfoRequest{})
	if err != nil {
		return fmt.Errorf("error in GetInfo: %v", err)
	}
	_, s.wumboSupported = info.Features[uint32(lnwire.WumboChannelsRequired)]
	if !s.wumboSupported {
		_, s.wumboSupported = info.Features[uint32(lnwire.WumboChannelsOptional)]
	}
	rpcLog.Infof("Connected to lnd node %v with pubkey %v", info.Alias,
		info.IdentityPubkey)
	var blockCtx context.Context
	blockCtx, s.blockNtfnCancel = context.WithCancel(ctx)
	chainNotifier := s.lndServices.ChainNotifier
	blockChan, blockErrChan, err := chainNotifier.RegisterBlockEpochNtfn(
		blockCtx,
	)
	if err != nil {
		return err
	}
	var height int32
	select {
	case height = <-blockChan:
	case err := <-blockErrChan:
		return fmt.Errorf("unable to receive first block "+
			"notification: %v", err)
	case <-ctx.Done():
		return nil
	}
	s.updateHeight(height)
	if err := s.auctioneer.Start(); err != nil {
		return fmt.Errorf("unable to start auctioneer client: %v", err)
	}
	if err := s.accountManager.Start(); err != nil {
		return fmt.Errorf("unable to start account manager: %v", err)
	}
	if err := s.orderManager.Start(); err != nil {
		return fmt.Errorf("unable to start order manager: %v", err)
	}
	if err := s.server.fundingManager.Start(); err != nil {
		return fmt.Errorf("unable to start funding manager: %v", err)
	}
	if err := s.server.sidecarAcceptor.Start(blockErrChan); err != nil {
		return fmt.Errorf("unable to start sidecar acceptor: %v", err)
	}
	s.wg.Add(1)
	go s.serverHandler(blockChan, blockErrChan)
	rpcLog.Infof("Trader server is now active")
	return nil
}
func (s *rpcServer) Stop() error {
	if !atomic.CompareAndSwapUint32(&s.stopped, 0, 1) {
		return nil
	}
	var returnErr error
	rpcLog.Info("Trader server stopping")
	if err := s.server.sidecarAcceptor.Stop(); err != nil {
		rpcLog.Errorf("Error stopping sidecar acceptor: %v", err)
		returnErr = err
	}
	if err := s.server.fundingManager.Stop(); err != nil {
		rpcLog.Errorf("Error stopping funding manager: %v", err)
	}
	s.accountManager.Stop()
	s.orderManager.Stop()
	if err := s.auctioneer.Stop(); err != nil {
		rpcLog.Errorf("Error closing server stream: %v", err)
	}
	close(s.quit)
	s.wg.Wait()
	s.blockNtfnCancel()
	rpcLog.Info("Stopped trader server")
	return returnErr
}
func (s *rpcServer) serverHandler(blockChan chan int32, blockErrChan chan error) {
	defer s.wg.Done()
	for {
		select {
		case msg := <-s.auctioneer.FromServerChan:
			if msg == nil {
				continue
			}
			rpcLog.Debugf("Received message from the server: %v", msg)
			err := s.handleServerMessage(msg)
			if err != nil && !errors.Is(err, order.ErrMismatchErr) {
				rpcLog.Errorf("Error handling server message: %v",
					err)
				interceptor.RequestShutdown()
			}
		case err := <-s.auctioneer.StreamErrChan:
			if err != nil && err != auctioneer.ErrServerShutdown {
				rpcLog.Errorf("Error in server stream: %v", err)
				err := s.auctioneer.HandleServerShutdown(err)
				if err != nil {
					rpcLog.Errorf("Error closing stream: %v",
						err)
				}
			}
			rpcLog.Error("Unknown server error: %v", err)
		case height := <-blockChan:
			rpcLog.Infof("Received new block notification: height=%v",
				height)
			s.updateHeight(height)
		case err := <-blockErrChan:
			if err != nil {
				rpcLog.Errorf("Unable to receive block "+
					"notification: %v", err)
				interceptor.RequestShutdown()
			}
		case <-s.quit:
			return
		}
	}
}
func (s *rpcServer) updateHeight(height int32) {
	atomic.StoreUint32(&s.bestHeight, uint32(height))
}
func (s *rpcServer) handleServerMessage(
	rpcMsg *auctioneerrpc.ServerAuctionMessage) error {
	switch msg := rpcMsg.Msg.(type) {
	case *auctioneerrpc.ServerAuctionMessage_Prepare:
		rpcLog.Tracef("Received prepare msg from server, batch_id=%x: %v",
			msg.Prepare.BatchId, spew.Sdump(msg))
		batch, err := order.ParseRPCBatch(msg.Prepare)
		if err != nil {
			log.Error("unable to parse batch: %v", err)
			return s.sendRejectBatch(batch, err)
		}
		rpcLog.Infof("Received PrepareMsg for batch=%x, num_orders=%v",
			batch.ID[:], len(batch.MatchedOrders))
		if err := s.server.db.StoreBatchEvents(
			batch, order.MatchStatePrepare,
			poolrpc.MatchRejectReason_NONE,
		); err != nil {
			rpcLog.Errorf("Unable to store order events: %v", err)
		}
		if s.orderManager.HasPendingBatch() {
			pendingBatch := s.orderManager.PendingBatch()
			err = s.server.fundingManager.RemovePendingBatchArtifacts(
				pendingBatch.MatchedOrders, pendingBatch.BatchTX,
			)
			if err != nil {
				rpcLog.Errorf("Error clearing previous batch "+
					"artifacts: %v", err)
				return s.sendRejectBatch(batch, err)
			}
			err = s.server.fundingManager.DeletePendingBatch()
			if err != nil {
				rpcLog.Errorf("Error clearing previous "+
					"pending batch: %v", err)
				return s.sendRejectBatch(batch, err)
			}
		}
		err = s.orderManager.OrderMatchValidate(batch)
		if err != nil {
			rpcLog.Errorf("Error validating batch: %v", err)
			return s.sendRejectBatch(batch, err)
		}
		err = s.server.fundingManager.PrepChannelFunding(
			batch, s.server.db.GetOrder,
		)
		if err != nil {
			rpcLog.Warnf("Error preparing channel funding: %v",
				err)
			return s.sendRejectBatch(batch, err)
		}
		err = s.sendAcceptBatch(batch)
		if err != nil {
			rpcLog.Errorf("Error sending accept msg: %v", err)
			return s.sendRejectBatch(batch, err)
		}
	case *auctioneerrpc.ServerAuctionMessage_Sign:
		batch := s.orderManager.PendingBatch()
		channelKeys, err := s.server.fundingManager.BatchChannelSetup(
			batch,
		)
		if err != nil {
			rpcLog.Errorf("Error setting up channels: %v", err)
			return s.sendRejectBatch(batch, err)
		}
		rpcLog.Infof("Received OrderMatchSignBegin for batch=%x, "+
			"num_orders=%v", batch.ID[:], len(batch.MatchedOrders))
		bestHeight := atomic.LoadUint32(&s.bestHeight)
		sigs, err := s.orderManager.BatchSign(bestHeight)
		if err != nil {
			rpcLog.Errorf("Error signing batch: %v", err)
			return s.sendRejectBatch(batch, err)
		}
		err = s.sendSignBatch(batch, sigs, channelKeys)
		if err != nil {
			rpcLog.Errorf("Error sending sign msg: %v", err)
			return s.sendRejectBatch(batch, err)
		}
	case *auctioneerrpc.ServerAuctionMessage_Finalize:
		rpcLog.Tracef("Received finalize msg from server, batch_id=%x: %v",
			msg.Finalize.BatchId, spew.Sdump(msg))
		rpcLog.Infof("Received FinalizeMsg for batch=%x",
			msg.Finalize.BatchId)
		batch := s.orderManager.PendingBatch()
		var batchID order.BatchID
		copy(batchID[:], msg.Finalize.BatchId)
		err := s.orderManager.BatchFinalize(batchID)
		if err != nil {
			return fmt.Errorf("error finalizing batch: %v", err)
		}
		if err := s.server.db.StoreBatchEvents(
			batch, order.MatchStateFinalized,
			poolrpc.MatchRejectReason_NONE,
		); err != nil {
			rpcLog.Errorf("Unable to store order events: %v", err)
		}
		for ourOrderNonce := range batch.MatchedOrders {
			if err := s.setTicketStateForOrder(
				sidecar.StateCompleted, ourOrderNonce,
			); err != nil {
				rpcLog.Errorf("Unable to update our sidecar "+
					"ticket after completing batch: %v",
					err)
			}
		}
		matchedAccounts := make(
			[]*btcec.PublicKey, len(batch.AccountDiffs),
		)
		for idx, acct := range batch.AccountDiffs {
			matchedAccounts[idx] = acct.AccountKey
		}
		return s.accountManager.WatchMatchedAccounts(
			context.Background(), matchedAccounts,
		)
	default:
		return fmt.Errorf("unknown server message: %v", msg)
	}
	return nil
}
func (s *rpcServer) QuoteAccount(ctx context.Context,
	req *poolrpc.QuoteAccountRequest) (*poolrpc.QuoteAccountResponse, error) {
	confTarget := req.GetConfTarget()
	if confTarget < 1 {
		return nil, fmt.Errorf("confirmation target must be " +
			"greater than 0")
	}
	feeRate, totalFee, err := s.accountManager.QuoteAccount(
		ctx, btcutil.Amount(req.AccountValue), confTarget,
	)
	if err != nil {
		return nil, err
	}
	return &poolrpc.QuoteAccountResponse{
		MinerFeeRateSatPerKw: uint64(feeRate),
		MinerFeeTotal:        uint64(totalFee),
	}, nil
}
func (s *rpcServer) InitAccount(ctx context.Context,
	req *poolrpc.InitAccountRequest) (*poolrpc.Account, error) {
	bestHeight := atomic.LoadUint32(&s.bestHeight)
	var expiryHeight uint32
	switch {
	case req.GetAbsoluteHeight() != 0 && req.GetRelativeHeight() != 0:
		return nil, fmt.Errorf("you must set only one of the relative " +
			"and absolute height parameters")
	case req.GetAbsoluteHeight() != 0:
		expiryHeight = req.GetAbsoluteHeight()
	case req.GetRelativeHeight() != 0:
		expiryHeight = req.GetRelativeHeight() + bestHeight
	default:
		return nil, fmt.Errorf("either relative or absolute height " +
			"must be specified")
	}
	var feeRate chainfee.SatPerKWeight
	switch {
	case req.GetFeeRateSatPerKw() > 0 && req.GetConfTarget() > 0:
		return nil, fmt.Errorf("you must set only one of the sats/kw " +
			"and confirmation target parameters")
	case req.GetFeeRateSatPerKw() > 0:
		feeRate = chainfee.SatPerKWeight(req.GetFeeRateSatPerKw())
	case req.GetConfTarget() > 0:
		value := btcutil.Amount(req.AccountValue)
		confTarget := req.GetConfTarget()
		var err error
		feeRate, _, err = s.accountManager.QuoteAccount(
			ctx, value, confTarget,
		)
		if err != nil {
			return nil, fmt.Errorf("unable to estimate on-chain fees: "+
				"%v", err)
		}
		if feeRate < chainfee.FeePerKwFloor {
			feeRate = chainfee.FeePerKwFloor
		}
		log.Infof("Estimated total chain fee of %v for new account with "+
			"value=%v, conf_target=%v", feeRate, value, confTarget)
	default:
		return nil, fmt.Errorf("either sats/kw or confirmation target " +
			"must be specified")
	}
	if feeRate < chainfee.FeePerKwFloor {
		return nil, fmt.Errorf("fee rate of %d sat/kw is too low, "+
			"minimum is %d sat/kw", feeRate, chainfee.FeePerKwFloor)
	}
	acct, err := s.accountManager.InitAccount(
		ContextWithInitiator(ctx, req.Initiator),
		btcutil.Amount(req.AccountValue), feeRate,
		expiryHeight, bestHeight,
	)
	if err != nil {
		return nil, err
	}
	return MarshallAccount(acct)
}
func (s *rpcServer) ListAccounts(ctx context.Context,
	req *poolrpc.ListAccountsRequest) (*poolrpc.ListAccountsResponse, error) {
	accounts, err := s.server.db.Accounts()
	if err != nil {
		return nil, err
	}
	validAccounts := make([]*account.Account, 0, len(accounts))
	for _, acct := range accounts {
		if req.ActiveOnly && !acct.State.IsActive() {
			continue
		}
		validAccounts = append(validAccounts, acct)
	}
	rpcAccounts, err := s.MarshallAccountsWithAvailableBalance(
		ctx, validAccounts,
	)
	if err != nil {
		return nil, fmt.Errorf("unable to list marshalled accounts: "+
			"%v", err)
	}
	return &poolrpc.ListAccountsResponse{
		Accounts: rpcAccounts,
	}, nil
}
func (s *rpcServer) MarshallAccountsWithAvailableBalance(ctx context.Context,
	accounts []*account.Account) ([]*poolrpc.Account, error) {
	rpcAccounts := make([]*poolrpc.Account, 0, len(accounts))
	for _, acct := range accounts {
		rpcAccount, err := MarshallAccount(acct)
		if err != nil {
			return nil, err
		}
		rpcAccounts = append(rpcAccounts, rpcAccount)
	}
	orders, err := s.server.db.GetOrders()
	if err != nil {
		return nil, err
	}
	auctionTerms, err := s.auctioneer.Terms(ctx)
	if err != nil {
		return nil, fmt.Errorf("unable to query auctioneer terms: %v",
			err)
	}
	accountDebits := make(map[[33]byte]btcutil.Amount)
	auctionFeeSchedule := auctionTerms.FeeSchedule()
	for _, acct := range accounts {
		var (
			debitAmt btcutil.Amount
			acctKey  [33]byte
		)
		copy(
			acctKey[:],
			acct.TraderKey.PubKey.SerializeCompressed(),
		)
		for _, o := range orders {
			if o.Details().AcctKey != acctKey {
				continue
			}
			debitAmt += o.ReservedValue(auctionFeeSchedule)
		}
		accountDebits[acctKey] = debitAmt
	}
	for _, rpcAccount := range rpcAccounts {
		var acctKey [33]byte
		copy(acctKey[:], rpcAccount.TraderKey)
		accountDebit := accountDebits[acctKey]
		availableBalance := rpcAccount.Value - uint64(accountDebit)
		rpcAccount.AvailableBalance = availableBalance
	}
	return rpcAccounts, nil
}