loopd: add assets manager/server

Author: sputn1ck

loopd/config.go

@@ -200,6 +200,8 @@ type Config struct {
 	Tapd *assets.TapdConfig `group:"tapd" namespace:"tapd"`
 
 	View viewParameters `command:"view" alias:"v" description:"View all swaps in the database. This command can only be executed when loopd is not running."`
+
+	TapdConfig *assets.TapdConfig `group:"tapd" namespace:"tapd"`
 }
 
 const (
@@ -247,6 +249,7 @@ func DefaultConfig() Config {
 			MacaroonPath: DefaultLndMacaroonPath,
 			RPCTimeout:   DefaultLndRPCTimeout,
 		},
+		TapdConfig: assets.DefaultTapdConfig(),
 	}
 }
 

loopd/daemon.go

@@ -249,6 +249,8 @@ func (d *Daemon) startWebServers() error {
 	)
 	loop_looprpc.RegisterSwapClientServer(d.grpcServer, d)
 
+	loop_looprpc.RegisterAssetsClientServer(d.grpcServer, d.assetsServer)
+
 	// Register our debug server if it is compiled in.
 	d.registerDebugServer()
 
@@ -332,7 +334,7 @@ func (d *Daemon) startWebServers() error {
 		d.wg.Add(1)
 		go func() {
 			defer d.wg.Done()
-
+			defer log.Info("REST proxy stopped")
 			log.Infof("REST proxy listening on %s",
 				d.restListener.Addr())
 			err := d.restServer.Serve(d.restListener)
@@ -354,7 +356,7 @@ func (d *Daemon) startWebServers() error {
 	d.wg.Add(1)
 	go func() {
 		defer d.wg.Done()
-
+		defer log.Info("RPC server stopped")
 		log.Infof("RPC server listening on %s", d.grpcListener.Addr())
 		err = d.grpcServer.Serve(d.grpcListener)
 		if err != nil && !errors.Is(err, grpc.ErrServerStopped) {
@@ -487,6 +489,11 @@ func (d *Daemon) initialize(withMacaroonService bool) error {
 		swapClient.Conn,
 	)
 
+	// Create a assets server client.
+	assetsClient := loop_swaprpc.NewAssetsSwapServerClient(
+		swapClient.Conn,
+	)
+
 	// Both the client RPC server and the swap server client should stop
 	// on main context cancel. So we create it early and pass it down.
 	d.mainCtx, d.mainCtxCancel = context.WithCancel(context.Background())
@@ -636,6 +643,8 @@ func (d *Daemon) initialize(withMacaroonService bool) error {
 	var (
 		reservationManager *reservation.Manager
 		instantOutManager  *instantout.Manager
+		assetManager       *assets.AssetsSwapManager
+		assetClientServer  *assets.AssetsClientServer
 	)
 
 	// Create the reservation and instantout managers.
@@ -676,6 +685,27 @@ func (d *Daemon) initialize(withMacaroonService bool) error {
 		instantOutManager = instantout.NewInstantOutManager(
 			instantOutConfig,
 		)
+
+		tapdClient, err := assets.NewTapdClient(
+			d.cfg.TapdConfig,
+		)
+		if err != nil {
+			return err
+		}
+		assetsStore := assets.NewPostgresStore(baseDb)
+		assetsConfig := &assets.Config{
+			ServerClient:         assetsClient,
+			Store:                assetsStore,
+			AssetClient:          tapdClient,
+			LndClient:            d.lnd.Client,
+			Router:               d.lnd.Router,
+			ChainNotifier:        d.lnd.ChainNotifier,
+			Signer:               d.lnd.Signer,
+			Wallet:               d.lnd.WalletKit,
+			ExchangeRateProvider: assets.NewFixedExchangeRateProvider(),
+		}
+		assetManager = assets.NewAssetSwapServer(assetsConfig)
+		assetClientServer = assets.NewAssetsServer(assetManager)
 	}
 
 	// Now finally fully initialize the swap client RPC server instance.
@@ -696,6 +726,8 @@ func (d *Daemon) initialize(withMacaroonService bool) error {
 		withdrawalManager:    withdrawalManager,
 		staticLoopInManager:  staticLoopInManager,
 		assetClient:          d.assetClient,
+		assetManager:         assetManager,
+		assetsServer:         assetClientServer,
 	}
 
 	// Retrieve all currently existing swaps from the database.
@@ -801,6 +833,10 @@ func (d *Daemon) initialize(withMacaroonService bool) error {
 			cancel()
 		}
 	}
+	getInfo, err := d.lnd.Client.GetInfo(d.mainCtx)
+	if err != nil {
+		return err
+	}
 
 	// Start the instant out manager.
 	if d.instantOutManager != nil {
@@ -809,12 +845,6 @@ func (d *Daemon) initialize(withMacaroonService bool) error {
 		go func() {
 			defer d.wg.Done()
 
-			getInfo, err := d.lnd.Client.GetInfo(d.mainCtx)
-			if err != nil {
-				d.internalErrChan <- err
-				return
-			}
-
 			log.Info("Starting instantout manager")
 			defer log.Info("Instantout manager stopped")
 
@@ -933,6 +963,20 @@ func (d *Daemon) initialize(withMacaroonService bool) error {
 		staticLoopInManager.WaitInitComplete()
 	}
 
+	// Start the asset manager.
+	if d.assetManager != nil {
+		d.wg.Add(1)
+		go func() {
+			defer d.wg.Done()
+			log.Infof("Starting asset manager")
+			defer log.Infof("Asset manager stopped")
+			err := d.assetManager.Run(d.mainCtx, int32(getInfo.BlockHeight))
+			if err != nil && !errors.Is(err, context.Canceled) {
+				d.internalErrChan <- err
+			}
+		}()
+	}
+
 	// Last, start our internal error handler. This will return exactly one
 	// error or nil on the main error channel to inform the caller that
 	// something went wrong or that shutdown is complete. We don't add to
@@ -978,6 +1022,9 @@ func (d *Daemon) Stop() {
 
 // stop does the actual shutdown and blocks until all goroutines have exit.
 func (d *Daemon) stop() {
+	// Sleep a second in order to fix a blocking issue when having a
+	// startup error.
+	<-time.After(time.Second)
 	// First of all, we can cancel the main context that all event handlers
 	// are using. This should stop all swap activity and all event handlers
 	// should exit.
@@ -995,6 +1042,7 @@ func (d *Daemon) stop() {
 	if d.restServer != nil {
 		// Don't return the error here, we first want to give everything
 		// else a chance to shut down cleanly.
+
 		err := d.restServer.Close()
 		if err != nil {
 			log.Errorf("Error stopping REST server: %v", err)

loopd/log.go

@@ -5,6 +5,7 @@ import (
 	"github.com/lightninglabs/aperture/l402"
 	"github.com/lightninglabs/lndclient"
 	"github.com/lightninglabs/loop"
+	"github.com/lightninglabs/loop/assets"
 	"github.com/lightninglabs/loop/fsm"
 	"github.com/lightninglabs/loop/instantout"
 	"github.com/lightninglabs/loop/instantout/reservation"
@@ -14,6 +15,7 @@ import (
 	"github.com/lightninglabs/loop/staticaddr"
 	"github.com/lightninglabs/loop/sweep"
 	"github.com/lightninglabs/loop/sweepbatcher"
+	"github.com/lightninglabs/loop/utils"
 	"github.com/lightningnetwork/lnd"
 	"github.com/lightningnetwork/lnd/build"
 	"github.com/lightningnetwork/lnd/signal"
@@ -58,6 +60,13 @@ func SetupLoggers(root *build.RotatingLogWriter, intercept signal.Interceptor) {
 	lnd.AddSubLogger(
 		root, sweep.Subsystem, intercept, sweep.UseLogger,
 	)
+
+	lnd.AddSubLogger(
+		root, assets.Subsystem, intercept, assets.UseLogger,
+	)
+	lnd.AddSubLogger(
+		root, utils.Subsystem, intercept, utils.UseLogger,
+	)
 }
 
 // genSubLogger creates a logger for a subsystem. We provide an instance of

loopd/perms/perms.go

@@ -169,4 +169,20 @@ var RequiredPermissions = map[string][]bakery.Op{
 		Entity: "swap",
 		Action: "read",
 	}},
+	"/looprpc.AssetsClient/SwapOut": {{
+		Entity: "swap",
+		Action: "execute",
+	}},
+	"/looprpc.AssetsClient/ListAssetSwaps": {{
+		Entity: "swap",
+		Action: "read",
+	}},
+	"/looprpc.AssetsClient/ClientListAvailableAssets": {{
+		Entity: "swap",
+		Action: "read",
+	}},
+	"/looprpc.AssetsClient/ClientGetAssetSwapOutQuote": {{
+		Entity: "swap",
+		Action: "read",
+	}},
 }

loopd/swapclient_server.go

@@ -96,6 +96,8 @@ type swapClientServer struct {
 	withdrawalManager    *withdraw.Manager
 	staticLoopInManager  *loopin.Manager
 	assetClient          *assets.TapdClient
+	assetManager         *assets.AssetsSwapManager
+	assetsServer         *assets.AssetsClientServer
 	swaps                map[lntypes.Hash]loop.SwapInfo
 	subscribers          map[int]chan<- interface{}
 	statusChan           chan loop.SwapInfo