server.go

package main

import (
	"bytes"
	"encoding/hex"
	"fmt"
	"net"
	"sync"
	"sync/atomic"

	"github.com/btcsuite/fastsha256"
	"github.com/lightningnetwork/lightning-onion"
	"github.com/lightningnetwork/lnd/chainntnfs"
	"github.com/lightningnetwork/lnd/channeldb"
	"github.com/lightningnetwork/lnd/lndc"
	"github.com/lightningnetwork/lnd/lnrpc"
	"github.com/lightningnetwork/lnd/lnwallet"
	"github.com/roasbeef/btcd/btcec"
	"github.com/roasbeef/btcutil"

	"github.com/BitfuryLightning/tools/routing"
	"github.com/BitfuryLightning/tools/rt/graph"
)

// server is the main server of the Lightning Network Daemon. The server
// houses global state pertianing to the wallet, database, and the rpcserver.
// Additionally, the server is also used as a central messaging bus to interact
// with any of its companion objects.
type server struct {
	started  int32 // atomic
	shutdown int32 // atomic

	// identityPriv is the private key used to authenticate any incoming
	// connections.
	identityPriv *btcec.PrivateKey

	// lightningID is the sha256 of the public key corresponding to our
	// long-term identity private key.
	lightningID [32]byte

	listeners []net.Listener
	peers     map[int32]*peer

	rpcServer *rpcServer

	chainNotifier chainntnfs.ChainNotifier

	bio      lnwallet.BlockChainIO
	lnwallet *lnwallet.LightningWallet

	// TODO(roasbeef): add to constructor
	fundingMgr *fundingManager
	chanDB     *channeldb.DB

	htlcSwitch *htlcSwitch
	invoices   *invoiceRegistry

	routingMgr *routing.RoutingManager

	utxoNursery *utxoNursery

	sphinx *sphinx.Router

	newPeers  chan *peer
	donePeers chan *peer
	queries   chan interface{}

	wg   sync.WaitGroup
	quit chan struct{}
}

// newServer creates a new instance of the server which is to listen using the
// passed listener address.
func newServer(listenAddrs []string, notifier chainntnfs.ChainNotifier,
	bio lnwallet.BlockChainIO, wallet *lnwallet.LightningWallet,
	chanDB *channeldb.DB) (*server, error) {

	privKey, err := wallet.GetIdentitykey()
	if err != nil {
		return nil, err
	}

	listeners := make([]net.Listener, len(listenAddrs))
	for i, addr := range listenAddrs {
		listeners[i], err = lndc.NewListener(privKey, addr)
		if err != nil {
			return nil, err
		}
	}

	serializedPubKey := privKey.PubKey().SerializeCompressed()
	s := &server{
		bio:           bio,
		chainNotifier: notifier,
		chanDB:        chanDB,
		fundingMgr:    newFundingManager(wallet),
		invoices:      newInvoiceRegistry(chanDB),
		lnwallet:      wallet,
		identityPriv:  privKey,
		// TODO(roasbeef): derive proper onion key based on rotation
		// schedule
		sphinx:      sphinx.NewRouter(privKey, activeNetParams.Params),
		lightningID: fastsha256.Sum256(serializedPubKey),
		listeners:   listeners,
		peers:       make(map[int32]*peer),
		newPeers:    make(chan *peer, 100),
		donePeers:   make(chan *peer, 100),
		queries:     make(chan interface{}),
		quit:        make(chan struct{}),
	}

	// If the debug HTLC flag is on, then we invoice a "master debug"
	// invoice which all outgoing payments will be sent and all incoming
	// HTLC's with the debug R-Hash immediately settled.
	if cfg.DebugHTLC {
		kiloCoin := btcutil.Amount(btcutil.SatoshiPerBitcoin * 1000)
		s.invoices.AddDebugInvoice(kiloCoin, *debugPre)
		srvrLog.Debugf("Debug HTLC invoice inserted, preimage=%x, hash=%x",
			debugPre[:], debugHash[:])
	}

	s.utxoNursery = newUtxoNursery(notifier, wallet)

	// Create a new routing manager with ourself as the sole node within
	// the graph.
	selfVertex := hex.EncodeToString(serializedPubKey)
	s.routingMgr = routing.NewRoutingManager(graph.NewID(selfVertex), nil)
	s.htlcSwitch = newHtlcSwitch(serializedPubKey, s.routingMgr)

	s.rpcServer = newRpcServer(s)

	return s, nil
}

// Start starts the main daemon server, all requested listeners, and any helper
// goroutines.
func (s *server) Start() error {
	// Already running?
	if atomic.AddInt32(&s.started, 1) != 1 {
		return nil
	}

	// Start all the listeners.
	for _, l := range s.listeners {
		s.wg.Add(1)
		go s.listener(l)
	}

	// Start the notification server. This is used so channel managment
	// goroutines can be notified when a funding transaction reaches a
	// sufficient number of confirmations, or when the input for the
	// funding transaction is spent in an attempt at an uncooperative
	// close by the counter party.
	if err := s.chainNotifier.Start(); err != nil {
		return err
	}

	if err := s.rpcServer.Start(); err != nil {
		return err
	}
	if err := s.fundingMgr.Start(); err != nil {
		return err
	}
	if err := s.htlcSwitch.Start(); err != nil {
		return err
	}
	if err := s.utxoNursery.Start(); err != nil {
		return err
	}
	s.routingMgr.Start()

	s.wg.Add(1)
	go s.queryHandler()

	return nil
}

// Stop gracefully shutsdown the main daemon server. This function will signal
// any active goroutines, or helper objects to exit, then blocks until they've
// all successfully exited. Additionally, any/all listeners are closed.
func (s *server) Stop() error {
	// Bail if we're already shutting down.
	if atomic.AddInt32(&s.shutdown, 1) != 1 {
		return nil
	}

	// Stop all the listeners.
	for _, listener := range s.listeners {
		if err := listener.Close(); err != nil {
			return err
		}
	}

	// Shutdown the wallet, funding manager, and the rpc server.
	s.chainNotifier.Stop()
	s.rpcServer.Stop()
	s.fundingMgr.Stop()
	s.routingMgr.Stop()
	s.htlcSwitch.Stop()
	s.utxoNursery.Stop()

	s.lnwallet.Shutdown()

	// Signal all the lingering goroutines to quit.
	close(s.quit)
	s.wg.Wait()

	return nil
}

// WaitForShutdown blocks all goroutines have been stopped.
func (s *server) WaitForShutdown() {
	s.wg.Wait()
}

// addPeer adds the passed peer to the server's global state of all active
// peers.
func (s *server) addPeer(p *peer) {
	if p == nil {
		return
	}

	// Ignore new peers if we're shutting down.
	if atomic.LoadInt32(&s.shutdown) != 0 {
		p.Stop()
		return
	}

	s.peers[p.id] = p
}

// removePeer removes the passed peer from the server's state of all active
// peers.
func (s *server) removePeer(p *peer) {
	srvrLog.Debugf("removing peer %v", p)

	if p == nil {
		return
	}

	// Ignore deleting peers if we're shutting down.
	if atomic.LoadInt32(&s.shutdown) != 0 {
		p.Stop()
		return
	}

	delete(s.peers, p.id)
}

// connectPeerMsg is a message requesting the server to open a connection to a
// particular peer. This message also houses an error channel which will be
// used to report success/failure.
type connectPeerMsg struct {
	addr *lndc.LNAdr
	resp chan int32
	err  chan error
}

// listPeersMsg is a message sent to the server in order to obtain a listing
// of all currently active channels.
type listPeersMsg struct {
	resp chan []*peer
}

// openChanReq is a message sent to the server in order to request the
// initiation of a channel funding workflow to the peer with either the specified
// relative peer ID, or a global lightning  ID.
type openChanReq struct {
	targetPeerID int32
	targetNodeID [32]byte

	// TODO(roasbeef): make enums in lnwire
	channelType uint8
	coinType    uint64

	localFundingAmt  btcutil.Amount
	remoteFundingAmt btcutil.Amount

	numConfs uint32

	updates chan *lnrpc.OpenStatusUpdate
	err     chan error
}

// queryHandler handles any requests to modify the server's internal state of
// all active peers, or query/mutate the server's global state. Additionally,
// any queries directed at peers will be handled by this goroutine.
//
// NOTE: This MUST be run as a goroutine.
func (s *server) queryHandler() {
out:
	for {
		select {
		// New peers.
		case p := <-s.newPeers:
			s.addPeer(p)

		// Finished peers.
		case p := <-s.donePeers:
			s.removePeer(p)

		case query := <-s.queries:
			// TODO(roasbeef): make all goroutines?
			switch msg := query.(type) {
			case *connectPeerMsg:
				s.handleConnectPeer(msg)
			case *listPeersMsg:
				s.handleListPeers(msg)
			case *openChanReq:
				s.handleOpenChanReq(msg)
			}
		case msg := <-s.routingMgr.ChOut:
			msg1 := msg.(*routing.RoutingMessage)
			if msg1.ReceiverID == nil {
				peerLog.Critical("msg1.GetReceiverID() == nil")
				continue
			}
			receiverID := msg1.ReceiverID.String()

			var targetPeer *peer
			for _, peer := range s.peers { // TODO: threadsafe api
				nodePub := peer.identityPub.SerializeCompressed()
				idStr := hex.EncodeToString(nodePub)

				// We found the the target
				if receiverID == idStr {
					targetPeer = peer
					break
				}
			}

			if targetPeer != nil {
				targetPeer.queueMsg(msg1.Msg, nil)
			} else {
				srvrLog.Errorf("Can't find peer to send message %v",
					receiverID)
			}
		case <-s.quit:
			break out
		}
	}

	s.wg.Done()
}

// handleListPeers sends a lice of all currently active peers to the original
// caller.
func (s *server) handleListPeers(msg *listPeersMsg) {
	peers := make([]*peer, 0, len(s.peers))
	for _, peer := range s.peers {
		peers = append(peers, peer)
	}

	msg.resp <- peers
}

// handleConnectPeer attempts to establish a connection to the address enclosed
// within the passed connectPeerMsg. This function is *async*, a goroutine will
// be spawned in order to finish the request, and respond to the caller.
func (s *server) handleConnectPeer(msg *connectPeerMsg) {
	addr := msg.addr

	// Ensure we're not already connected to this
	// peer.
	for _, peer := range s.peers {
		if peer.lightningAddr.String() == addr.String() {
			msg.err <- fmt.Errorf(
				"already connected to peer: %v",
				peer.lightningAddr,
			)
			msg.resp <- -1
			return
		}
	}

	// Launch a goroutine to connect to the requested
	// peer so we can continue to handle queries.
	// TODO(roasbeef): semaphore to limit the number of goroutines for
	// async requests.
	go func() {
		// For the lndc crypto handshake, we
		// either need a compressed pubkey, or a
		// 20-byte pkh.
		var remoteId []byte
		if addr.PubKey == nil {
			remoteId = addr.Base58Adr.ScriptAddress()
		} else {
			remoteId = addr.PubKey.SerializeCompressed()
		}

		srvrLog.Debugf("connecting to %v", hex.EncodeToString(remoteId))
		// Attempt to connect to the remote
		// node. If the we can't make the
		// connection, or the crypto negotation
		// breaks down, then return an error to the
		// caller.
		ipAddr := addr.NetAddr.String()
		conn := lndc.NewConn(nil)
		if err := conn.Dial(
			s.identityPriv, ipAddr, remoteId); err != nil {
			msg.err <- err
			msg.resp <- -1
			return
		}

		// Now that we've established a connection,
		// create a peer, and it to the set of
		// currently active peers.
		peer, err := newPeer(conn, s, activeNetParams.Net, false)
		if err != nil {
			srvrLog.Errorf("unable to create peer %v", err)
			conn.Close()
			msg.resp <- -1
			msg.err <- err
			return
		}

		peer.Start()
		s.newPeers <- peer

		msg.resp <- peer.id
		msg.err <- nil
	}()
}

// handleOpenChanReq first locates the target peer, and if found hands off the
// request to the funding manager allowing it to initiate the channel funding
// workflow.
func (s *server) handleOpenChanReq(req *openChanReq) {
	// First attempt to locate the target peer to open a channel with, if
	// we're unable to locate the peer then this request will fail.
	var targetPeer *peer
	for _, peer := range s.peers { // TODO(roasbeef): threadsafe api
		// We found the the target
		if req.targetPeerID == peer.id ||
			bytes.Equal(req.targetNodeID[:], peer.lightningID[:]) {
			targetPeer = peer
			break
		}
	}

	if targetPeer == nil {
		req.err <- fmt.Errorf("unable to find peer lightningID(%v), "+
			"peerID(%v)", req.targetNodeID, req.targetPeerID)
		return
	}

	// Spawn a goroutine to send the funding workflow request to the funding
	// manager. This allows the server to continue handling queries instead of
	// blocking on this request which is exporeted as a synchronous request to
	// the outside world.
	// TODO(roasbeef): server semaphore to restrict num goroutines
	go s.fundingMgr.initFundingWorkflow(targetPeer, req)
}

// ConnectToPeer requests that the server connect to a Lightning Network peer
// at the specified address. This function will *block* until either a
// connection is established, or the initial handshake process fails.
func (s *server) ConnectToPeer(addr *lndc.LNAdr) (int32, error) {
	reply := make(chan int32, 1)
	errChan := make(chan error, 1)

	s.queries <- &connectPeerMsg{addr, reply, errChan}

	return <-reply, <-errChan
}

// OpenChannel sends a request to the server to open a channel to the specified
// peer identified by ID with the passed channel funding paramters.
func (s *server) OpenChannel(peerID int32, nodeID []byte, localAmt, remoteAmt btcutil.Amount,
	numConfs uint32) (chan *lnrpc.OpenStatusUpdate, chan error) {

	errChan := make(chan error, 1)
	updateChan := make(chan *lnrpc.OpenStatusUpdate, 1)

	req := &openChanReq{
		targetPeerID:     peerID,
		localFundingAmt:  localAmt,
		remoteFundingAmt: remoteAmt,
		numConfs:         numConfs,
		updates:          updateChan,
		err:              errChan,
	}
	copy(req.targetNodeID[:], nodeID)

	s.queries <- req

	return updateChan, errChan
}

// Peers returns a slice of all active peers.
func (s *server) Peers() []*peer {
	resp := make(chan []*peer)

	s.queries <- &listPeersMsg{resp}

	return <-resp
}

// listener is a goroutine dedicated to accepting in coming peer connections
// from the passed listener.
//
// NOTE: This MUST be run as a goroutine.
func (s *server) listener(l net.Listener) {
	srvrLog.Infof("Server listening on %s", l.Addr())
	for atomic.LoadInt32(&s.shutdown) == 0 {
		conn, err := l.Accept()
		if err != nil {
			// Only log the error message if we aren't currently
			// shutting down.
			if atomic.LoadInt32(&s.shutdown) == 0 {
				srvrLog.Errorf("Can't accept connection: %v", err)
			}
			continue
		}

		srvrLog.Tracef("New inbound connection from %v", conn.RemoteAddr())
		peer, err := newPeer(conn, s, activeNetParams.Net, true)
		if err != nil {
			srvrLog.Errorf("unable to create peer: %v", err)
			conn.Close()
			continue
		}

		peer.Start()
		s.newPeers <- peer
	}

	s.wg.Done()
}