tss: Populate tss worker pools

cmd/ingest.go

@@ -3,6 +3,8 @@ package cmd
 import (
 	"fmt"
 	"go/types"
+	"net/http"
+	"time"
 
 	_ "github.com/lib/pq"
 	"github.com/spf13/cobra"
@@ -11,6 +13,7 @@ import (
 	"github.com/stellar/wallet-backend/cmd/utils"
 	"github.com/stellar/wallet-backend/internal/apptracker/sentry"
 	"github.com/stellar/wallet-backend/internal/ingest"
+	tsschannels "github.com/stellar/wallet-backend/internal/tss/channels"
 )
 
 type ingestCmd struct{}
@@ -22,28 +25,15 @@ func (c *ingestCmd) Command() *cobra.Command {
 	cfgOpts := config.ConfigOptions{
 		utils.DatabaseURLOption(&cfg.DatabaseURL),
 		utils.LogLevelOption(&cfg.LogLevel),
-		utils.NetworkPassphraseOption(&cfg.NetworkPassphrase),
 		utils.SentryDSNOption(&sentryDSN),
 		utils.StellarEnvironmentOption(&stellarEnvironment),
 		utils.RPCURLOption(&cfg.RPCURL),
-		{
-			Name:           "captive-core-bin-path",
-			Usage:          "Path to Captive Core's binary file.",
-			OptType:        types.String,
-			CustomSetValue: utils.SetConfigOptionCaptiveCoreBinPath,
-			ConfigKey:      &cfg.CaptiveCoreBinPath,
-			FlagDefault:    "/usr/local/bin/stellar-core",
-			Required:       true,
-		},
-		{
-			Name:           "captive-core-config-dir",
-			Usage:          "Path to Captive Core's configuration files directory.",
-			OptType:        types.String,
-			CustomSetValue: utils.SetConfigOptionCaptiveCoreConfigDir,
-			ConfigKey:      &cfg.CaptiveCoreConfigDir,
-			FlagDefault:    "./internal/ingest/config",
-			Required:       true,
-		},
+		utils.StartLedgerOption(&cfg.StartLedger),
+		utils.EndLedgerOption(&cfg.EndLedger),
+		utils.WebhookHandlerServiceChannelMaxBufferSizeOption(&cfg.WebhookChannelMaxBufferSize),
+		utils.WebhookHandlerServiceChannelMaxWorkersOptions(&cfg.WebhookChannelMaxWorkers),
+		utils.WebhookHandlerServiceChannelMaxRetriesOption(&cfg.WebhookChannelMaxRetries),
+		utils.WebhookHandlerServiceChannelMinWaitBtwnRetriesMSOption(&cfg.WebhookChannelWaitBtwnTriesMS),
 		{
 			Name:        "ledger-cursor-name",
 			Usage:       "Name of last synced ledger cursor, used to keep track of the last ledger ingested by the service. When starting up, ingestion will resume from the ledger number stored in this record. It should be an unique name per container as different containers would overwrite the cursor value of its peers when using the same cursor name.",
@@ -60,14 +50,6 @@ func (c *ingestCmd) Command() *cobra.Command {
 			FlagDefault: 0,
 			Required:    false,
 		},
-		{
-			Name:        "end",
-			Usage:       "Ledger number up to which ingestion should run. When not present, ingestion run indefinitely (live ingestion requires it to be empty).",
-			OptType:     types.Int,
-			ConfigKey:   &cfg.EndLedger,
-			FlagDefault: 0,
-			Required:    false,
-		},
 	}
 
 	cmd := &cobra.Command{
@@ -85,11 +67,21 @@ func (c *ingestCmd) Command() *cobra.Command {
 				return fmt.Errorf("initializing app tracker: %w", err)
 			}
 			cfg.AppTracker = appTracker
+			cfg.WebhookChannel = tsschannels.NewWebhookChannel(tsschannels.WebhookChannelConfigs{
+				HTTPClient:           &http.Client{Timeout: 30 * time.Second},
+				MaxBufferSize:        cfg.WebhookChannelMaxBufferSize,
+				MaxWorkers:           cfg.WebhookChannelMaxWorkers,
+				MaxRetries:           cfg.WebhookChannelMaxRetries,
+				MinWaitBtwnRetriesMS: cfg.WebhookChannelWaitBtwnTriesMS,
+			})
 			return nil
 		},
 		RunE: func(_ *cobra.Command, _ []string) error {
 			return c.Run(cfg)
 		},
+		PersistentPostRun: func(_ *cobra.Command, _ []string) {
+			cfg.WebhookChannel.Stop()
+		},
 	}
 
 	if err := cfgOpts.Init(cmd); err != nil {

cmd/serve.go

@@ -43,6 +43,11 @@ func (c *serveCmd) Command() *cobra.Command {
 		utils.ErrorHandlerServiceNonJitterChannelWaitBtwnRetriesMSOption(&cfg.ErrorHandlerServiceNonJitterChannelWaitBtwnRetriesMS),
 		utils.ErrorHandlerServiceJitterChannelMaxRetriesOptions(&cfg.ErrorHandlerServiceJitterChannelMaxRetries),
 		utils.ErrorHandlerServiceNonJitterChannelMaxRetriesOption(&cfg.ErrorHandlerServiceNonJitterChannelMaxRetries),
+		utils.WebhookHandlerServiceChannelMaxBufferSizeOption(&cfg.WebhookHandlerServiceChannelMaxBufferSize),
+		utils.WebhookHandlerServiceChannelMaxWorkersOptions(&cfg.WebhookHandlerServiceChannelMaxWorkers),
+		utils.WebhookHandlerServiceChannelMaxRetriesOption(&cfg.WebhookHandlerServiceChannelMaxRetries),
+		utils.WebhookHandlerServiceChannelMinWaitBtwnRetriesMSOption(&cfg.WebhookHandlerServiceChannelMinWaitBtwnRetriesMS),
+
 		{
 			Name:        "port",
 			Usage:       "Port to listen and serve on",

cmd/utils/custom_set_value.go

@@ -2,9 +2,7 @@ package utils
 
 import (
 	"encoding/json"
-	"errors"
 	"fmt"
-	"os"
 	"strings"
 
 	"github.com/sirupsen/logrus"
@@ -83,48 +81,6 @@ func SetConfigOptionStellarPrivateKey(co *config.ConfigOption) error {
 	return nil
 }
 
-func SetConfigOptionCaptiveCoreBinPath(co *config.ConfigOption) error {
-	binPath := viper.GetString(co.Name)
-
-	fileInfo, err := os.Stat(binPath)
-	if errors.Is(err, os.ErrNotExist) {
-		return fmt.Errorf("binary file %s does not exist", binPath)
-	}
-
-	if fileInfo.IsDir() {
-		return fmt.Errorf("binary file path %s is a directory, not a file", binPath)
-	}
-
-	key, ok := co.ConfigKey.(*string)
-	if !ok {
-		return unexpectedTypeError(key, co)
-	}
-	*key = binPath
-
-	return nil
-}
-
-func SetConfigOptionCaptiveCoreConfigDir(co *config.ConfigOption) error {
-	dirPath := viper.GetString(co.Name)
-
-	fileInfo, err := os.Stat(dirPath)
-	if errors.Is(err, os.ErrNotExist) {
-		return fmt.Errorf("captive core configuration files dir %s does not exist", dirPath)
-	}
-
-	if !fileInfo.IsDir() {
-		return fmt.Errorf("captive core configuration files dir %s is not a directory", dirPath)
-	}
-
-	key, ok := co.ConfigKey.(*string)
-	if !ok {
-		return unexpectedTypeError(key, co)
-	}
-	*key = dirPath
-
-	return nil
-}
-
 func SetConfigOptionAssets(co *config.ConfigOption) error {
 	assetsJSON := viper.GetString(co.Name)
 

cmd/utils/custom_set_value_test.go

@@ -220,97 +220,6 @@ func Test_SetConfigOptionLogLevel(t *testing.T) {
 	}
 }
 
-func TestSetConfigOptionCaptiveCoreBinPath(t *testing.T) {
-	opts := struct{ binPath string }{}
-
-	co := config.ConfigOption{
-		Name:           "captive-core-bin-path",
-		OptType:        types.String,
-		CustomSetValue: SetConfigOptionCaptiveCoreBinPath,
-		ConfigKey:      &opts.binPath,
-	}
-
-	testCases := []customSetterTestCase[string]{
-		{
-			name:            "returns an error if the file path is not set, should be caught by the Require() function",
-			wantErrContains: "binary file  does not exist",
-		},
-		{
-			name:            "returns an error if the path is invalid",
-			args:            []string{"--captive-core-bin-path", "/a/random/path/bin"},
-			wantErrContains: "binary file /a/random/path/bin does not exist",
-		},
-		{
-			name:            "returns an error if the path format is invalid",
-			args:            []string{"--captive-core-bin-path", "^7JcrS8J4q@V0$c"},
-			wantErrContains: "binary file ^7JcrS8J4q@V0$c does not exist",
-		},
-		{
-			name:            "returns an error if the path is a directory, not a file",
-			args:            []string{"--captive-core-bin-path", "./"},
-			wantErrContains: "binary file path ./ is a directory, not a file",
-		},
-		{
-			name:       "sets to ENV var value",
-			envValue:   "./custom_set_value_test.go",
-			wantResult: "./custom_set_value_test.go",
-		},
-	}
-
-	for _, tc := range testCases {
-		t.Run(tc.name, func(t *testing.T) {
-			opts.binPath = ""
-			customSetterTester(t, tc, co)
-		})
-	}
-}
-
-func TestSetConfigOptionCaptiveCoreConfigDir(t *testing.T) {
-	opts := struct{ binPath string }{}
-
-	co := config.ConfigOption{
-		Name:           "captive-core-config-dir",
-		OptType:        types.String,
-		CustomSetValue: SetConfigOptionCaptiveCoreConfigDir,
-		ConfigKey:      &opts.binPath,
-	}
-
-	testCases := []customSetterTestCase[string]{
-		{
-			name:            "returns an error if the file path is not set, should be caught by the Require() function",
-			wantErrContains: "captive core configuration files dir  does not exist",
-		},
-		{
-			name:            "returns an error if the path is invalid",
-			envValue:        "/a/random/path",
-			wantErrContains: "captive core configuration files dir /a/random/path does not exist",
-		},
-		{
-			name:            "returns an error if the path format is invalid",
-			envValue:        "^7JcrS8J4q@V0$c",
-			wantErrContains: "captive core configuration files dir ^7JcrS8J4q@V0$c does not exist",
-		},
-
-		{
-			name:            "returns an error if the path is a file, not a directory",
-			envValue:        "./custom_set_value_test.go",
-			wantErrContains: "captive core configuration files dir ./custom_set_value_test.go is not a directory",
-		},
-		{
-			name:       "sets to ENV var value",
-			envValue:   "../../internal/ingest/config",
-			wantResult: "../../internal/ingest/config",
-		},
-	}
-
-	for _, tc := range testCases {
-		t.Run(tc.name, func(t *testing.T) {
-			opts.binPath = ""
-			customSetterTester(t, tc, co)
-		})
-	}
-}
-
 func TestSetConfigOptionAssets(t *testing.T) {
 	opts := struct{ assets []entities.Asset }{}
 

cmd/utils/global_options.go

@@ -142,6 +142,30 @@ func DistributionAccountSignatureClientProviderOption(configKey *signing.Signatu
 	}
 }
 
+func StartLedgerOption(configKey *int) *config.ConfigOption {
+	return &config.ConfigOption{
+		Name:        "start-ledger",
+		Usage:       "ledger number to start getting transactions from",
+		OptType:     types.Int,
+		ConfigKey:   configKey,
+		FlagDefault: 0,
+		Required:    true,
+	}
+
+}
+
+func EndLedgerOption(configKey *int) *config.ConfigOption {
+	return &config.ConfigOption{
+		Name:        "end-ledger",
+		Usage:       "ledger number to end on",
+		OptType:     types.Int,
+		ConfigKey:   configKey,
+		FlagDefault: 0,
+		Required:    true,
+	}
+
+}
+
 func AWSOptions(awsRegionConfigKey *string, kmsKeyARN *string, required bool) config.ConfigOptions {
 	awsOpts := config.ConfigOptions{
 		{

cmd/utils/tss_options.go

@@ -116,3 +116,47 @@ func ErrorHandlerServiceNonJitterChannelMaxRetriesOption(configKey *int) *config
 		Required:    true,
 	}
 }
+
+func WebhookHandlerServiceChannelMaxBufferSizeOption(configKey *int) *config.ConfigOption {
+	return &config.ConfigOption{
+		Name:        "webhook-service-channel-max-buffer-size",
+		Usage:       "Set the buffer size of the webhook serive channel.",
+		OptType:     types.Int,
+		ConfigKey:   configKey,
+		FlagDefault: 100,
+		Required:    true,
+	}
+}
+
+func WebhookHandlerServiceChannelMaxWorkersOptions(configKey *int) *config.ConfigOption {
+	return &config.ConfigOption{
+		Name:        "webhook-service-channel-max-workers",
+		Usage:       "Set the max number of workers for the webhook serive channel.",
+		OptType:     types.Int,
+		ConfigKey:   configKey,
+		FlagDefault: 10,
+		Required:    true,
+	}
+}
+
+func WebhookHandlerServiceChannelMaxRetriesOption(configKey *int) *config.ConfigOption {
+	return &config.ConfigOption{
+		Name:        "webhook-service-channel-max-retries",
+		Usage:       "Set the max number of times to ping a webhook before quitting.",
+		OptType:     types.Int,
+		ConfigKey:   configKey,
+		FlagDefault: 3,
+		Required:    true,
+	}
+}
+
+func WebhookHandlerServiceChannelMinWaitBtwnRetriesMSOption(configKey *int) *config.ConfigOption {
+	return &config.ConfigOption{
+		Name:        "webhook-service-channel-min-wait-between-retries",
+		Usage:       "The minumum amout of time to wait before resending the payload to the webhook url",
+		OptType:     types.Int,
+		ConfigKey:   configKey,
+		FlagDefault: 10,
+		Required:    true,
+	}
+}