ARCO-153: Parallelise Create UTXO

cmd/broadcaster-cli/app/utxos/create/create.go

@@ -5,15 +5,14 @@ import (
 	"fmt"
 	"log"
 	"log/slog"
-
-	"github.com/bitcoin-sv/arc/pkg/keyset"
-
-	"github.com/spf13/cobra"
-	"github.com/spf13/viper"
+	"os"
+	"os/signal"
 
 	"github.com/bitcoin-sv/arc/cmd/broadcaster-cli/helper"
 	"github.com/bitcoin-sv/arc/internal/broadcaster"
 	"github.com/bitcoin-sv/arc/internal/woc_client"
+	"github.com/spf13/cobra"
+	"github.com/spf13/viper"
 )
 
 var Cmd = &cobra.Command{
@@ -78,32 +77,42 @@ var Cmd = &cobra.Command{
 			return fmt.Errorf("failed to create client: %v", err)
 		}
 
+		names := helper.GetOrderedKeys(keySetsMap)
+
 		wocClient := woc_client.New(!isTestnet, woc_client.WithAuth(wocApiKey), woc_client.WithLogger(logger))
+		creators := make([]broadcaster.Creator, 0, len(keySetsMap)) // Use the Creator interface for flexibility
+		for _, keyName := range names {
+			ks := keySetsMap[keyName]
+			creator, err := broadcaster.NewUTXOCreator(
+				logger.With(slog.String("address", ks.Address(!isTestnet)), slog.String("name", keyName)),
+				client, ks, wocClient, isTestnet, broadcaster.WithFees(miningFeeSat),
+			)
+			if err != nil {
+				return err
+			}
 
-		ks := make([]*keyset.KeySet, len(keySetsMap))
-		counter := 0
-		for _, keySet := range keySetsMap {
-			ks[counter] = keySet
-			counter++
-		}
-		rateBroadcaster, err := broadcaster.NewUTXOCreator(logger, client, ks, wocClient, isTestnet,
-			broadcaster.WithFees(miningFeeSat),
-		)
-		if err != nil {
-			return fmt.Errorf("failed to create broadcaster: %v", err)
+			creators = append(creators, creator)
 		}
 
-		err = rateBroadcaster.CreateUtxos(outputs, uint64(satoshisPerOutput))
-		if err != nil {
-			return fmt.Errorf("failed to create utxos: %v", err)
-		}
+		multiCreator := broadcaster.NewMultiKeyUTXOCreator(logger, creators)
+
+		signalChan := make(chan os.Signal, 1)
+		signal.Notify(signalChan, os.Interrupt) // Listen for Ctrl+C
+
+		// Graceful shutdown on interrupt signal
+		go func() {
+			<-signalChan
+			multiCreator.Shutdown()
+		}()
+
+		logger.Info("Starting UTXO creation")
+		multiCreator.Start(outputs, uint64(satoshisPerOutput))
 
 		return nil
 	},
 }
 
 func init() {
-
 	logger := helper.GetLogger()
 	Cmd.SetHelpFunc(func(command *cobra.Command, strings []string) {
 		// Hide unused persistent flags

internal/broadcaster/broadcaster_mocks.go

@@ -11,3 +11,6 @@ package broadcaster
 
 // from multi_rate_broadcaster.go
 //go:generate moq -pkg mocks -out ./mocks/rate_broadcaster_mock.go . RateBroadcaster
+
+// from multi_utxo_creator.go
+//go:generate moq -pkg mocks -out ./mocks/utxo_creator_mock.go . Creator

internal/broadcaster/multi_utxo_creator.go

@@ -0,0 +1,50 @@
+package broadcaster
+
+import (
+	"log/slog"
+)
+
+type MultiKeyUTXOCreator struct {
+	creators []Creator
+	logger   *slog.Logger
+}
+
+type Creator interface {
+	Start(outputs int, satoshisPerOutput uint64) error
+	Wait()
+	Shutdown()
+}
+
+func NewMultiKeyUTXOCreator(logger *slog.Logger, creators []Creator, opts ...func(p *MultiKeyUTXOCreator)) *MultiKeyUTXOCreator {
+	mkuc := &MultiKeyUTXOCreator{
+		creators: creators,
+		logger:   logger,
+	}
+
+	for _, opt := range opts {
+		opt(mkuc)
+	}
+
+	return mkuc
+}
+
+func (mkuc *MultiKeyUTXOCreator) Start(outputs int, satoshisPerOutput uint64) {
+
+	for _, creator := range mkuc.creators {
+		err := creator.Start(outputs, satoshisPerOutput)
+		if err != nil {
+			mkuc.logger.Error("failed to start UTXO creator", slog.String("err", err.Error()))
+		}
+	}
+
+	for _, creator := range mkuc.creators {
+		creator.Wait()
+	}
+}
+
+func (mkuc *MultiKeyUTXOCreator) Shutdown() {
+
+	for _, creator := range mkuc.creators {
+		creator.Shutdown()
+	}
+}

internal/broadcaster/multi_utxo_creator_test.go

@@ -0,0 +1,42 @@
+package broadcaster_test
+
+import (
+	"errors"
+	"log/slog"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/bitcoin-sv/arc/internal/broadcaster"
+	"github.com/bitcoin-sv/arc/internal/broadcaster/mocks"
+)
+
+func TestMultiKeyUTXOCreatorStart(t *testing.T) {
+	t.Run("start and shutdown", func(t *testing.T) {
+		logger := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelDebug}))
+
+		// Create mocks for creators
+		creators := []broadcaster.Creator{
+			&mocks.CreatorMock{
+				StartFunc:    func(outputs int, satoshisPerOutput uint64) error { return nil },
+				WaitFunc:     func() {},
+				ShutdownFunc: func() {},
+			},
+			&mocks.CreatorMock{
+				StartFunc:    func(outputs int, satoshisPerOutput uint64) error { return errors.New("failed to start") },
+				WaitFunc:     func() {},
+				ShutdownFunc: func() {},
+			},
+		}
+
+		// Initialize the MultiKeyUTXOCreator
+		mkuc := broadcaster.NewMultiKeyUTXOCreator(logger, creators)
+
+		// Start the MultiKeyUTXOCreator
+		mkuc.Start(100, 1000)
+		time.Sleep(50 * time.Millisecond)
+
+		// Shutdown the MultiKeyUTXOCreator
+		mkuc.Shutdown()
+	})
+}