services: rpc payment ingestor

cmd/ingest.go

@@ -22,27 +22,11 @@ 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),
-		{
-			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.RPCURLOption(&cfg.RPCURL),
+		utils.StartLedgerOption(&cfg.StartLedger),
+		utils.EndLedgerOption(&cfg.EndLedger),
 		{
 			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.",
@@ -59,20 +43,13 @@ 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{
 		Use:   "ingest",
 		Short: "Run Ingestion service",
 		PersistentPreRunE: func(cmd *cobra.Command, args []string) error {
+			// SET UP WEBHOOK CHANNEL HERE
 			if err := cfgOpts.RequireE(); err != nil {
 				return fmt.Errorf("requiring values of config options: %w", err)
 			}
@@ -89,6 +66,8 @@ func (c *ingestCmd) Command() *cobra.Command {
 		RunE: func(_ *cobra.Command, _ []string) error {
 			return c.Run(cfg)
 		},
+		PersistentPostRun: func(_ *cobra.Command, _ []string) {
+		},
 	}
 
 	if err := cfgOpts.Init(cmd); err != nil {

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

@@ -131,6 +131,41 @@ func DistributionAccountSignatureClientProviderOption(configKey *signing.Signatu
 	}
 }
 
+func RPCURLOption(configKey *string) *config.ConfigOption {
+	return &config.ConfigOption{
+		Name:        "rpc-url",
+		Usage:       "The URL of the RPC Server.",
+		OptType:     types.String,
+		ConfigKey:   configKey,
+		FlagDefault: "localhost:8080",
+		Required:    true,
+	}
+}
+
+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{
 		{

go.mod

@@ -21,36 +21,20 @@ require (
 )
 
 require (
-	cloud.google.com/go v0.112.0 // indirect
-	cloud.google.com/go/compute v1.23.3 // indirect
-	cloud.google.com/go/compute/metadata v0.2.3 // indirect
-	cloud.google.com/go/iam v1.1.5 // indirect
-	cloud.google.com/go/storage v1.37.0 // indirect
 	github.com/BurntSushi/toml v1.3.2 // indirect
 	github.com/Masterminds/squirrel v1.5.4 // indirect
-	github.com/Microsoft/go-winio v0.6.1 // indirect
 	github.com/asaskevich/govalidator v0.0.0-20230301143203-a9d515a09cc2 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
-	github.com/cenkalti/backoff/v4 v4.2.1 // indirect
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
-	github.com/djherbis/fscache v0.10.1 // indirect
-	github.com/felixge/httpsnoop v1.0.4 // indirect
 	github.com/fsnotify/fsnotify v1.6.0 // indirect
 	github.com/gabriel-vasile/mimetype v1.4.3 // indirect
 	github.com/go-errors/errors v1.5.1 // indirect
 	github.com/go-gorp/gorp/v3 v3.1.0 // indirect
-	github.com/go-logr/logr v1.3.0 // indirect
-	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/go-playground/locales v0.14.1 // indirect
 	github.com/go-playground/universal-translator v0.18.1 // indirect
-	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
 	github.com/golang/protobuf v1.5.3 // indirect
-	github.com/google/s2a-go v0.1.7 // indirect
-	github.com/googleapis/enterprise-certificate-proxy v0.3.2 // indirect
-	github.com/googleapis/gax-go/v2 v2.12.0 // indirect
 	github.com/gorilla/schema v1.2.0 // indirect
-	github.com/hashicorp/golang-lru v1.0.2 // indirect
 	github.com/hashicorp/hcl v1.0.0 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/jmespath/go-jmespath v0.4.0 // indirect
@@ -61,7 +45,6 @@ require (
 	github.com/manucorporat/sse v0.0.0-20160126180136-ee05b128a739 // indirect
 	github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect
 	github.com/mitchellh/mapstructure v1.5.0 // indirect
-	github.com/pelletier/go-toml v1.9.5 // indirect
 	github.com/pelletier/go-toml/v2 v2.1.0 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
@@ -79,32 +62,13 @@ require (
 	github.com/spf13/pflag v1.0.5 // indirect
 	github.com/stretchr/objx v0.5.2 // indirect
 	github.com/subosito/gotenv v1.6.0 // indirect
-	go.opencensus.io v0.24.0 // indirect
-	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.46.1 // indirect
-	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.46.1 // indirect
-	go.opentelemetry.io/otel v1.21.0 // indirect
-	go.opentelemetry.io/otel/metric v1.21.0 // indirect
-	go.opentelemetry.io/otel/trace v1.21.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	golang.org/x/crypto v0.21.0 // indirect
 	golang.org/x/exp v0.0.0-20231006140011-7918f672742d // indirect
-	golang.org/x/mod v0.13.0 // indirect
 	golang.org/x/net v0.23.0 // indirect
-	golang.org/x/oauth2 v0.16.0 // indirect
-	golang.org/x/sync v0.6.0 // indirect
 	golang.org/x/sys v0.18.0 // indirect
 	golang.org/x/text v0.14.0 // indirect
-	golang.org/x/time v0.5.0 // indirect
-	golang.org/x/tools v0.14.0 // indirect
-	google.golang.org/api v0.157.0 // indirect
-	google.golang.org/appengine v1.6.8 // indirect
-	google.golang.org/genproto v0.0.0-20240116215550-a9fa1716bcac // indirect
-	google.golang.org/genproto/googleapis/api v0.0.0-20240122161410-6c6643bf1457 // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20240116215550-a9fa1716bcac // indirect
-	google.golang.org/grpc v1.60.1 // indirect
 	google.golang.org/protobuf v1.33.0 // indirect
-	gopkg.in/djherbis/atime.v1 v1.0.0 // indirect
-	gopkg.in/djherbis/stream.v1 v1.3.1 // indirect
 	gopkg.in/ini.v1 v1.67.0 // indirect
 	gopkg.in/tylerb/graceful.v1 v1.2.15 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect