feat: global swap info stream (#124)

* feat: global swap info stream

* fix: place test db in tempdir

boltzrpc/boltzrpc.proto

@@ -51,6 +51,7 @@ service Boltz {
 
     /*
     Returns the entire history of the swap if is still pending and streams updates in real time.
+    If the swap id is empty or "*" updates for all swaps will be streamed.
     */
     rpc GetSwapInfoStream (GetSwapInfoRequest) returns (stream GetSwapInfoResponse);
 

cmd/boltzd/boltzd_test.go

@@ -47,7 +47,7 @@ func loadConfig(t *testing.T) *config.Config {
 	dataDir := "test"
 	cfg, err := config.LoadConfig(dataDir)
 	require.NoError(t, err)
-	cfg.Database.Path = "file:test.db?cache=shared&mode=memory"
+	cfg.Database.Path = t.TempDir() + "/boltz.db"
 	cfg.Node = "cln"
 	cfg.Node = "lnd"
 	return cfg
@@ -220,6 +220,46 @@ func TestGetInfo(t *testing.T) {
 	}
 }
 
+func TestGetSwapInfoStream(t *testing.T) {
+	client, _, stop := setup(t, nil, "")
+	defer stop()
+
+	stream, err := client.GetSwapInfoStream("")
+	require.NoError(t, err)
+
+	updates := make(chan *boltzrpc.GetSwapInfoResponse)
+	go func() {
+		for {
+			status, err := stream.Recv()
+			if err != nil {
+				close(updates)
+				return
+			}
+			updates <- status
+		}
+	}()
+
+	swap, err := client.CreateSwap(&boltzrpc.CreateSwapRequest{})
+	require.NoError(t, err)
+
+	select {
+	case info := <-updates:
+		require.Equal(t, swap.Id, info.Swap.Id)
+	case <-time.After(2 * time.Second):
+		require.Fail(t, "no swap update received")
+	}
+
+	reverseSwap, err := client.CreateReverseSwap(&boltzrpc.CreateReverseSwapRequest{Amount: 100000})
+	require.NoError(t, err)
+
+	select {
+	case info := <-updates:
+		require.Equal(t, reverseSwap.Id, info.ReverseSwap.Id)
+	case <-time.After(2 * time.Second):
+		require.Fail(t, "no reverse swap update received")
+	}
+}
+
 func TestGetPairs(t *testing.T) {
 	cfg := loadConfig(t)
 	client, _, stop := setup(t, cfg, "")
@@ -378,12 +418,12 @@ func TestSwap(t *testing.T) {
 				SendFromInternal: true,
 			})
 			require.NoError(t, err)
-			info, err := client.GetSwapInfo(swap.Id)
-			require.NoError(t, err)
+
+			stream := swapStream(t, client, swap.Id)
+			info := stream(boltzrpc.SwapState_PENDING)
 			require.Equal(t, invoice.PaymentRequest, info.Swap.Invoice)
 
 			test.MineBlock()
-			stream := swapStream(t, client, swap.Id)
 			stream(boltzrpc.SwapState_SUCCESSFUL)
 
 			paid, err := node.CheckInvoicePaid(invoice.PaymentHash)

docs/grpc.md

@@ -79,7 +79,7 @@ Gets all available information about a swap from the database.
 
 #### GetSwapInfoStream
 
-Returns the entire history of the swap if is still pending and streams updates in real time.
+Returns the entire history of the swap if is still pending and streams updates in real time. If the swap id is empty or "*" updates for all swaps will be streamed.
 
 | Request | Response |
 | ------- | -------- |

nursery/nursery.go

@@ -28,6 +28,7 @@ type Nursery struct {
 
 	eventListeners     map[string]swapListener
 	eventListenersLock sync.RWMutex
+	globalListener     swapListener
 	waitGroup          sync.WaitGroup
 	stop               *utils.ChannelForwarder[bool]
 
@@ -45,6 +46,7 @@ type SwapUpdate struct {
 type swapListener = *utils.ChannelForwarder[SwapUpdate]
 
 func (nursery *Nursery) sendUpdate(id string, update SwapUpdate) {
+	nursery.globalListener.Send(update)
 	if listener, ok := nursery.eventListeners[id]; ok {
 		listener.Send(update)
 		logger.Debugf("Sent update for swap %s", id)
@@ -69,6 +71,13 @@ func (nursery *Nursery) SwapUpdates(id string) (<-chan SwapUpdate, func()) {
 	return nil, nil
 }
 
+func (nursery *Nursery) GlobalSwapUpdates() (<-chan SwapUpdate, func()) {
+	updates := nursery.globalListener.Get()
+	return updates, func() {
+		nursery.globalListener.Remove(updates)
+	}
+}
+
 func (nursery *Nursery) Init(
 	network *boltz.Network,
 	lightning lightning.LightningNode,
@@ -82,6 +91,7 @@ func (nursery *Nursery) Init(
 	nursery.database = database
 	nursery.onchain = chain
 	nursery.eventListeners = make(map[string]swapListener)
+	nursery.globalListener = utils.ForwardChannel(make(chan SwapUpdate), 0, false)
 	nursery.stop = utils.ForwardChannel(make(chan bool), 0, false)
 	nursery.boltzWs = boltz.NewBoltzWebsocket(boltzClient.URL)
 
@@ -106,6 +116,7 @@ func (nursery *Nursery) Stop() {
 	for id := range nursery.eventListeners {
 		nursery.removeSwapListener(id)
 	}
+	nursery.globalListener.Close()
 	logger.Debugf("Closed all event listeners")
 	nursery.boltzWs.Close()
 	nursery.waitGroup.Wait()

rpcserver/router.go

@@ -227,25 +227,33 @@ func (server *routedBoltzServer) GetSwapInfo(_ context.Context, request *boltzrp
 }
 
 func (server *routedBoltzServer) GetSwapInfoStream(request *boltzrpc.GetSwapInfoRequest, stream boltzrpc.Boltz_GetSwapInfoStreamServer) error {
-	logger.Info("Starting Swap info stream for " + request.Id)
-	info, err := server.GetSwapInfo(context.Background(), request)
-	if err != nil {
-		return handleError(err)
-	}
+	var updates <-chan nursery.SwapUpdate
+	var stop func()
 
-	updates, stop := server.nursery.SwapUpdates(request.Id)
-	if updates != nil {
-		for update := range updates {
-			if err := stream.Send(&boltzrpc.GetSwapInfoResponse{
-				Swap:        serializeSwap(update.Swap),
-				ReverseSwap: serializeReverseSwap(update.ReverseSwap),
-			}); err != nil {
-				stop()
+	if request.Id == "" || request.Id == "*" {
+		logger.Info("Starting global Swap info stream")
+		updates, stop = server.nursery.GlobalSwapUpdates()
+	} else {
+		logger.Info("Starting Swap info stream for " + request.Id)
+		updates, stop = server.nursery.SwapUpdates(request.Id)
+		if updates == nil {
+			info, err := server.GetSwapInfo(context.Background(), request)
+			if err != nil {
 				return handleError(err)
 			}
+			if err := stream.Send(info); err != nil {
+				return handleError(err)
+			}
+			return nil
 		}
-	} else {
-		if err := stream.Send(info); err != nil {
+	}
+
+	for update := range updates {
+		if err := stream.Send(&boltzrpc.GetSwapInfoResponse{
+			Swap:        serializeSwap(update.Swap),
+			ReverseSwap: serializeReverseSwap(update.ReverseSwap),
+		}); err != nil {
+			stop()
 			return handleError(err)
 		}
 	}