func TestFundingManager(t *testing.T) {
	h := newManagerHarness(t)
	defer h.stop()
	_, pubKeyAsk := test.CreateKey(0)
	_, pubKeyBid := test.CreateKey(1)
	ask := &order.Ask{
		Kit: newKitFromTemplate(order.Nonce{0x01}, &order.Kit{
			MultiSigKeyLocator: keychain.KeyLocator{
				Family: keychain.KeyFamilyTowerSession,
				Index:  0,
			},
			Units:            4,
			UnitsUnfulfilled: 4,
			FixedRate:        10000,
			LeaseDuration:    2500,
		}),
	}
	err := h.db.SubmitOrder(ask)
	require.NoError(t, err)
	bid := &order.Bid{
		Kit: newKitFromTemplate(order.Nonce{0x02}, &order.Kit{
			MultiSigKeyLocator: keychain.KeyLocator{
				Family: keychain.KeyFamilyTowerSession,
				Index:  1,
			},
			Units:            4,
			UnitsUnfulfilled: 4,
			FixedRate:        10000,
			LeaseDuration:    2500,
		}),
	}
	err = h.db.SubmitOrder(bid)
	require.NoError(t, err)
	matchedAsk := &order.MatchedOrder{
		Order:       ask,
		UnitsFilled: 4,
		MultiSigKey: [33]byte{2, 3, 4},
		NodeKey:     node1Key,
		NodeAddrs:   []net.Addr{addr1},
	}
	matchedBid := &order.MatchedOrder{
		Order:       bid,
		UnitsFilled: 4,
		MultiSigKey: [33]byte{3, 4, 5},
		NodeKey:     node2Key,
		NodeAddrs:   []net.Addr{addr2},
	}
	_, fundingOutput1, _ := input.GenFundingPkScript(
		pubKeyAsk.SerializeCompressed(),
		matchedBid.MultiSigKey[:], int64(4*order.BaseSupplyUnit),
	)
	_, fundingOutput2, _ := input.GenFundingPkScript(
		pubKeyBid.SerializeCompressed(),
		matchedAsk.MultiSigKey[:], int64(4*order.BaseSupplyUnit),
	)
	batchTx := &wire.MsgTx{
		TxOut: []*wire.TxOut{fundingOutput1, fundingOutput2},
	}
	batch := &order.Batch{
		ID: order.BatchID{9, 8, 7},
		MatchedOrders: map[order.Nonce][]*order.MatchedOrder{
			ask.Nonce(): {matchedBid},
			bid.Nonce(): {matchedAsk},
		},
		BatchTX: batchTx,
	}
	pendingChanID := order.PendingChanKey(ask.Nonce(), bid.Nonce())
	h.baseClientMock.peerList = map[route.Vertex]string{
		node1Key: "1.1.1.1",
	}
	err = h.mgr.PrepChannelFunding(batch, h.db.GetOrder)
	require.NoError(t, err)
	require.Eventually(t, func() bool {
		return len(h.lnMock.Connections()) == 1
	}, testTimeout, testTimeout/10)
	conns := h.lnMock.Connections()
	require.Equal(t, 1, len(conns))
	require.Equal(t, addr1.String(), conns[node1Key])
	require.Equal(t, 1, len(h.baseClientMock.fundingShims))
	shim := h.baseClientMock.fundingShims[pendingChanID]
	require.NotNil(t, shim)
	assertFundingShim(
		t, shim, uint32(2500), int64(order.SupplyUnit(4).ToSatoshis()),
		batchTx, 1, &keychain.KeyDescriptor{
			KeyLocator: keychain.KeyLocator{
				Family: bid.Kit.MultiSigKeyLocator.Family,
				Index:  bid.Kit.MultiSigKeyLocator.Index,
			},
			PubKey: pubKeyBid,
		}, matchedAsk.MultiSigKey[:])
	h.mgr.cfg.NewNodesOnly = true
	h.lnMock.Channels = append(h.lnMock.Channels, lndclient.ChannelInfo{
		PubKeyBytes: node1Key,
	})
	err = h.mgr.PrepChannelFunding(batch, h.db.GetOrder)
	require.Error(t, err)
	expectedErr := &MatchRejectErr{
		RejectedOrders: map[order.Nonce]*auctioneerrpc.OrderReject{
			ask.Nonce(): {
				ReasonCode: auctioneerrpc.OrderReject_DUPLICATE_PEER,
				Reason: "already have open/pending channel " +
					"with peer",
			},
		},
	}
	require.Equal(t, expectedErr, err)
	h.mgr.cfg.NewNodesOnly = false
	h.baseClientMock.peerList = make(map[route.Vertex]string)
	err = h.mgr.PrepChannelFunding(batch, h.db.GetOrder)
	require.Error(t, err)
	expectedErr = &MatchRejectErr{
		RejectedOrders: map[order.Nonce]*auctioneerrpc.OrderReject{
			ask.Nonce(): {
				ReasonCode: auctioneerrpc.OrderReject_CHANNEL_FUNDING_FAILED,
				Reason: "connection not established before " +
					"timeout",
			},
		},
	}
	require.Equal(t, expectedErr, err)
	h.lnMock.ScbKeyRing.EncryptionKey.PubKey = pubKeyAsk
	callBatchChannelSetup(t, h, batch, false)
	callBatchChannelSetup(t, h, batch, true)
	_, err = h.mgr.BatchChannelSetup(batch)
	require.Error(t, err)
	code := &auctioneerrpc.OrderReject{
		ReasonCode: auctioneerrpc.OrderReject_CHANNEL_FUNDING_FAILED,
		Reason: "timed out waiting for pending open " +
			"channel notification",
	}
	expectedErr = &MatchRejectErr{
		RejectedOrders: map[order.Nonce]*auctioneerrpc.OrderReject{
			ask.Nonce(): code,
			bid.Nonce(): code,
		},
	}
	require.Equal(t, expectedErr, err)
	_, err = h.mgr.SidecarBatchChannelSetup(
		batch, h.mgr.pendingOpenChanClient, h.db.GetOrder,
	)
	require.Error(t, err)
	require.Equal(t, expectedErr, err)
}
TestFundingManager tests that the two main steps of the funding manager (the
channel funding preparation and the batch channel setup) are executed
correctly. This involves checking the derived keys, funding shims and the
established connections.
package funding
import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"fmt"
	"io/ioutil"
	"math/big"
	"net"
	"os"
	"testing"
	"time"
	"github.com/btcsuite/btcd/btcec"
	"github.com/btcsuite/btcd/wire"
	"github.com/btcsuite/btcutil"
	"github.com/lightninglabs/lndclient"
	"github.com/lightninglabs/pool/auctioneerrpc"
	"github.com/lightninglabs/pool/chaninfo"
	"github.com/lightninglabs/pool/clientdb"
	"github.com/lightninglabs/pool/internal/test"
	"github.com/lightninglabs/pool/order"
	"github.com/lightninglabs/pool/sidecar"
	"github.com/lightningnetwork/lnd/input"
	"github.com/lightningnetwork/lnd/keychain"
	"github.com/lightningnetwork/lnd/lnrpc"
	"github.com/lightningnetwork/lnd/routing/route"
	"github.com/stretchr/testify/require"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)
const (
	testTimeout = 100 * time.Millisecond
)
var (
	node1Key = [33]byte{2, 3, 4}
	node2Key = [33]byte{3, 4, 5}
	addr1, _ = net.ResolveTCPAddr("tcp4", "10.0.1.1:9735")
	addr2, _ = net.ResolveTCPAddr("tcp4", "192.168.1.1:9735")
)
type openChannelStream struct {
	lnrpc.Lightning_OpenChannelClient
	updateChan chan *lnrpc.OpenStatusUpdate
	quit       chan struct{}
}
func (i *openChannelStream) Recv() (*lnrpc.OpenStatusUpdate, error) {
	select {
	case msg := <-i.updateChan:
		return msg, nil
	case <-i.quit:
		return nil, context.Canceled
	}
}
type peerEventStream struct {
	lnrpc.Lightning_SubscribePeerEventsClient
	updateChan         chan *lnrpc.PeerEvent
	ctx                context.Context
	cancelSubscription chan struct{}
	quit               chan struct{}
}
func (i *peerEventStream) Recv() (*lnrpc.PeerEvent, error) {
	select {
	case msg := <-i.updateChan:
		return msg, nil
	case <-i.ctx.Done():
		return nil, status.Error(
			codes.DeadlineExceeded, i.ctx.Err().Error(),
		)
	case <-i.cancelSubscription:
		return nil, status.Error(
			codes.Canceled, "subscription canceled",
		)
	case <-i.quit:
		return nil, context.Canceled
	}
}
type channelEventStream struct {
	lnrpc.Lightning_SubscribeChannelEventsClient
	updateChan         chan *lnrpc.ChannelEventUpdate
	ctx                context.Context
	cancelSubscription chan struct{}
	quit               chan struct{}
}
func (c *channelEventStream) Recv() (*lnrpc.ChannelEventUpdate, error) {
	select {
	case msg := <-c.updateChan:
		return msg, nil
	case <-c.ctx.Done():
		return nil, status.Error(
			codes.DeadlineExceeded, c.ctx.Err().Error(),
		)
	case <-c.cancelSubscription:
		return nil, status.Error(
			codes.Canceled, "subscription canceled",
		)
	case <-c.quit:
		return nil, context.Canceled
	}
}
type fundingBaseClientMock struct {
	lightningClient *test.MockLightning
	fundingShims    map[[32]byte]*lnrpc.ChanPointShim
	peerList      map[route.Vertex]string
	peerEvents    chan *lnrpc.PeerEvent
	channelEvents chan *lnrpc.ChannelEventUpdate
	cancelSub     chan struct{}
	quit chan struct{}
}
func (m *fundingBaseClientMock) FundingStateStep(_ context.Context,
	req *lnrpc.FundingTransitionMsg,
	_ ...grpc.CallOption) (*lnrpc.FundingStateStepResp, error) {
	register := req.GetShimRegister()
	if register == nil || register.GetChanPointShim() == nil {
		return nil, fmt.Errorf("invalid funding shim")
	}
	var tempChanID [32]byte
	copy(tempChanID[:], register.GetChanPointShim().PendingChanId)
	m.fundingShims[tempChanID] = register.GetChanPointShim()
	return nil, nil
}
func (m *fundingBaseClientMock) OpenChannel(ctx context.Context,
	req *lnrpc.OpenChannelRequest,
	_ ...grpc.CallOption) (lnrpc.Lightning_OpenChannelClient, error) {
	if req.FundingShim == nil || req.FundingShim.GetChanPointShim() == nil {
		return nil, fmt.Errorf("invalid funding shim")
	}
	var tempChanID [32]byte
	copy(tempChanID[:], req.FundingShim.GetChanPointShim().PendingChanId)
	_, ok := m.fundingShims[tempChanID]
	if !ok {
		return nil, fmt.Errorf("invalid funding shim")
	}
	node, err := route.NewVertexFromBytes(req.NodePubkey)
	if err != nil {
		return nil, err
	}
	op, _ := m.lightningClient.OpenChannel(
		ctx, node, btcutil.Amount(req.LocalFundingAmount),
		btcutil.Amount(req.PushSat), false,
	)
	stream := &openChannelStream{
		quit:       m.quit,
		updateChan: make(chan *lnrpc.OpenStatusUpdate),
	}
	go func() {
		select {
		case stream.updateChan <- &lnrpc.OpenStatusUpdate{
			Update: &lnrpc.OpenStatusUpdate_ChanPending{
				ChanPending: &lnrpc.PendingUpdate{
					Txid:        op.Hash[:],
					OutputIndex: op.Index,
				},
			},
		}:
		case <-m.quit:
		}
	}()
	return stream, nil
}
func (m *fundingBaseClientMock) SubscribeChannelEvents(ctx context.Context,
	_ *lnrpc.ChannelEventSubscription, _ ...grpc.CallOption) (
	lnrpc.Lightning_SubscribeChannelEventsClient, error) {
	return &channelEventStream{
		quit:       m.quit,
		ctx:        ctx,
		updateChan: m.channelEvents,
	}, nil
}
func (m *fundingBaseClientMock) ListPeers(_ context.Context,
	_ *lnrpc.ListPeersRequest,
	_ ...grpc.CallOption) (*lnrpc.ListPeersResponse, error) {
	resp := &lnrpc.ListPeersResponse{}
	for nodeKey, addr := range m.peerList {
		resp.Peers = append(resp.Peers, &lnrpc.Peer{
			PubKey:  nodeKey.String(),
			Address: addr,
		})
	}
	return resp, nil
}
func (m *fundingBaseClientMock) SubscribePeerEvents(ctx context.Context,
	_ *lnrpc.PeerEventSubscription, _ ...grpc.CallOption) (
	lnrpc.Lightning_SubscribePeerEventsClient, error) {
	return &peerEventStream{
		quit:               m.quit,
		ctx:                ctx,
		cancelSubscription: m.cancelSub,
		updateChan:         m.peerEvents,
	}, nil
}
func (m *fundingBaseClientMock) AbandonChannel(_ context.Context,
	_ *lnrpc.AbandonChannelRequest,
	_ ...grpc.CallOption) (*lnrpc.AbandonChannelResponse, error) {
	return nil, nil
}
type managerHarness struct {
	t              *testing.T
	tempDir        string
	db             *clientdb.DB
	quit           chan struct{}
	lnMock         *test.MockLightning
	baseClientMock *fundingBaseClientMock
	signerMock     *test.MockSigner
	mgr            *Manager
}
func newManagerHarness(t *testing.T) *managerHarness {
	tempDir, err := ioutil.TempDir("", "client-db")
	require.NoError(t, err)
	db, err := clientdb.New(tempDir, clientdb.DBFilename)
	if err != nil {
		_ = os.RemoveAll(tempDir)
		t.Fatalf("unable to create new db: %v", err)
	}
	quit := make(chan struct{})
	lightningClient := test.NewMockLightning()
	walletKitClient := test.NewMockWalletKit()
	signerClient := test.NewMockSigner()
	baseClientMock := &fundingBaseClientMock{
		lightningClient: lightningClient,
		fundingShims:    make(map[[32]byte]*lnrpc.ChanPointShim),
		peerList:        make(map[route.Vertex]string),
		peerEvents:      make(chan *lnrpc.PeerEvent),
		channelEvents:   make(chan *lnrpc.ChannelEventUpdate),
		cancelSub:       make(chan struct{}),
		quit:            quit,
	}
	mgr := NewManager(&ManagerConfig{
		DB:              db,
		WalletKit:       walletKitClient,
		LightningClient: lightningClient,
		BaseClient:      baseClientMock,
		SignerClient:    signerClient,
		NodePubKey: &btcec.PublicKey{
			Curve: btcec.S256(),
			X:     new(big.Int),
			Y:     new(big.Int),
		},
		NewNodesOnly:     true,
		BatchStepTimeout: 400 * time.Millisecond,
	})
	err = mgr.Start()
	require.NoError(t, err)
	return &managerHarness{
		t:              t,
		tempDir:        tempDir,
		db:             db,
		quit:           quit,
		lnMock:         lightningClient,
		baseClientMock: baseClientMock,
		signerMock:     signerClient,
		mgr:            mgr,
	}
}
func (m *managerHarness) stop() {
	close(m.quit)
	require.NoError(m.t, m.db.Close())
	require.NoError(m.t, os.RemoveAll(m.tempDir))
	require.NoError(m.t, m.mgr.Stop())
}