loopd: add assets manager/server

Author: sputn1ck

loopd/config.go

@@ -201,6 +201,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 (

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()
 
@@ -495,6 +497,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())
@@ -644,6 +651,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.
@@ -684,6 +693,27 @@ func (d *Daemon) initialize(withMacaroonService bool) error {
 		instantOutManager = instantout.NewInstantOutManager(
 			instantOutConfig, int32(blockHeight),
 		)
+
+		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.
@@ -704,6 +734,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.
@@ -898,6 +930,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()
+			infof("Starting asset manager")
+			defer 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
@@ -943,6 +989,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.
@@ -960,6 +1009,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 {
 			errorf("Error stopping REST server: %v", err)

loopd/log.go

@@ -7,6 +7,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"
@@ -16,6 +17,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"
@@ -92,6 +94,13 @@ func SetupLoggers(root *build.SubLoggerManager, 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