fix: consolidate reliability and safety fixes

cmd/root.go

@@ -5,16 +5,17 @@ import (
 	"fmt"
 	"net/http"
 
-	"github.com/formancehq/go-libs/v3/auth"
-	"github.com/formancehq/go-libs/v3/bun/bunconnect"
-	"github.com/formancehq/go-libs/v3/bun/bunmigrate"
-	"github.com/formancehq/go-libs/v3/licence"
-	"github.com/formancehq/go-libs/v3/otlp"
-	"github.com/formancehq/go-libs/v3/otlp/otlpmetrics"
-	"github.com/formancehq/go-libs/v3/otlp/otlptraces"
-	"github.com/formancehq/go-libs/v3/publish"
-	"github.com/formancehq/go-libs/v3/service"
-	"github.com/formancehq/go-libs/v3/temporal"
+	"github.com/formancehq/go-libs/v5/pkg/fx/authnfx"
+	"github.com/formancehq/go-libs/v5/pkg/fx/messagingfx"
+	"github.com/formancehq/go-libs/v5/pkg/fx/observefx"
+	"github.com/formancehq/go-libs/v5/pkg/fx/storagefx"
+	"github.com/formancehq/go-libs/v5/pkg/fx/workflowfx"
+	otlp "github.com/formancehq/go-libs/v5/pkg/observe"
+	otlptraces "github.com/formancehq/go-libs/v5/pkg/observe/traces"
+	"github.com/formancehq/go-libs/v5/pkg/service"
+	bunconnect "github.com/formancehq/go-libs/v5/pkg/storage/bun/connect"
+	bunmigrate "github.com/formancehq/go-libs/v5/pkg/storage/bun/migrate"
+	"github.com/formancehq/go-libs/v5/pkg/workflow/temporal"
 	"github.com/formancehq/orchestration/internal/storage"
 	"github.com/formancehq/orchestration/internal/temporalworker"
 	"github.com/formancehq/orchestration/internal/tracer"
@@ -43,6 +44,7 @@ const (
 	topicsFlag            = "topics"
 	listenFlag            = "listen"
 	workerFlag            = "worker"
+	stackHTTPClientName   = "stack"
 )
 
 func NewRootCommand() *cobra.Command {
@@ -69,19 +71,44 @@ func Execute() {
 	service.Execute(NewRootCommand())
 }
 
+func stackHTTPClientModule(cmd *cobra.Command) fx.Option {
+	return fx.Provide(fx.Annotate(func() *http.Client {
+		httpClient := &http.Client{
+			Transport: otlp.NewRoundTripper(http.DefaultTransport, service.IsDebug(cmd)),
+		}
+
+		stackClientID, _ := cmd.Flags().GetString(stackClientIDFlag)
+		stackClientSecret, _ := cmd.Flags().GetString(stackClientSecretFlag)
+		stackURL, _ := cmd.Flags().GetString(stackURLFlag)
+
+		if stackClientID == "" {
+			return httpClient
+		}
+		oauthConfig := clientcredentials.Config{
+			ClientID:     stackClientID,
+			ClientSecret: stackClientSecret,
+			TokenURL:     fmt.Sprintf("%s/api/auth/oauth/token", stackURL),
+			Scopes:       []string{"openid", "ledger:read", "ledger:write", "wallets:read", "wallets:write", "payments:read", "payments:write"},
+		}
+		return oauthConfig.Client(context.WithValue(context.Background(),
+			oauth2.HTTPClient, httpClient))
+	}, fx.ResultTags(`name:"stack"`)))
+}
+
 func commonOptions(cmd *cobra.Command) (fx.Option, error) {
-	connectionOptions, err := bunconnect.ConnectionOptionsFromFlags(cmd)
+	connectionOptions, err := bunconnect.ConnectionOptionsFromFlags(cmd.Flags(), cmd.Context())
 	if err != nil {
 		return nil, err
 	}
 
 	stack, _ := cmd.Flags().GetString(stackFlag)
+	stackURL, _ := cmd.Flags().GetString(stackURLFlag)
 	temporalTaskQueue, _ := cmd.Flags().GetString(temporal.TemporalTaskQueueFlag)
 
 	return fx.Options(
-		otlp.FXModuleFromFlags(cmd),
-		otlptraces.FXModuleFromFlags(cmd),
-		temporal.FXModuleFromFlags(
+		observefx.ResourceModuleFromFlags(cmd),
+		observefx.TracesModuleFromFlags(cmd),
+		workflowfx.TemporalClientModuleFromFlags(
 			cmd,
 			tracer.Tracer,
 			temporal.SearchAttributes{
@@ -91,36 +118,16 @@ func commonOptions(cmd *cobra.Command) (fx.Option, error) {
 				),
 			},
 		),
-		otlpmetrics.FXModuleFromFlags(cmd),
-		bunconnect.Module(*connectionOptions, service.IsDebug(cmd)),
-		publish.FXModuleFromFlags(cmd, service.IsDebug(cmd)),
-		auth.FXModuleFromFlags(cmd),
-		licence.FXModuleFromFlags(cmd, ServiceName),
+		observefx.MetricsModuleFromFlags(cmd),
+		storagefx.BunConnectModule(*connectionOptions, service.IsDebug(cmd)),
+		messagingfx.PublishModuleFromFlags(cmd, service.IsDebug(cmd)),
+		authnfx.JWTModuleFromFlags(cmd),
+		authnfx.LicenceModuleFromFlags(cmd, ServiceName),
 		workflow.NewModule(stack, temporalTaskQueue),
-		triggers.NewModule(stack, temporalTaskQueue),
+		triggers.NewModule(stack, stackURL, temporalTaskQueue, stackHTTPClientName),
 		fx.Provide(func() *bunconnect.ConnectionOptions {
 			return connectionOptions
 		}),
-		fx.Provide(func() *http.Client {
-			httpClient := &http.Client{
-				Transport: otlp.NewRoundTripper(http.DefaultTransport, service.IsDebug(cmd)),
-			}
-
-			stackClientID, _ := cmd.Flags().GetString(stackClientIDFlag)
-			stackClientSecret, _ := cmd.Flags().GetString(stackClientSecretFlag)
-			stackURL, _ := cmd.Flags().GetString(stackURLFlag)
-
-			if stackClientID == "" {
-				return httpClient
-			}
-			oauthConfig := clientcredentials.Config{
-				ClientID:     stackClientID,
-				ClientSecret: stackClientSecret,
-				TokenURL:     fmt.Sprintf("%s/api/auth/oauth/token", stackURL),
-				Scopes:       []string{"openid", "ledger:read", "ledger:write", "wallets:read", "wallets:write", "payments:read", "payments:write"},
-			}
-			return oauthConfig.Client(context.WithValue(context.Background(),
-				oauth2.HTTPClient, httpClient))
-		}),
+		stackHTTPClientModule(cmd),
 	), nil
 }

cmd/root_test.go

@@ -0,0 +1,26 @@
+package cmd
+
+import (
+	"net/http"
+	"testing"
+
+	"github.com/formancehq/go-libs/v5/pkg/fx/authnfx"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/fx"
+)
+
+func TestCommonOptionsBuildsWithJWTAndStackHTTPClients(t *testing.T) {
+	cmd := newServeCommand()
+	app := fx.New(
+		fx.NopLogger,
+		authnfx.JWTModuleFromFlags(cmd),
+		stackHTTPClientModule(cmd),
+		fx.Invoke(fx.Annotate(
+			func(stackClient *http.Client) {
+				require.NotNil(t, stackClient)
+			},
+			fx.ParamTags(`name:"stack"`),
+		)),
+	)
+	require.NoError(t, app.Err())
+}

cmd/serve.go

@@ -3,16 +3,18 @@ package cmd
 import (
 	"context"
 
-	"github.com/formancehq/go-libs/v3/auth"
-	"github.com/formancehq/go-libs/v3/aws/iam"
-	"github.com/formancehq/go-libs/v3/bun/bunconnect"
-	"github.com/formancehq/go-libs/v3/health"
-	"github.com/formancehq/go-libs/v3/httpserver"
-	"github.com/formancehq/go-libs/v3/licence"
-	"github.com/formancehq/go-libs/v3/otlp/otlpmetrics"
-	"github.com/formancehq/go-libs/v3/publish"
-	"github.com/formancehq/go-libs/v3/service"
-	"github.com/formancehq/go-libs/v3/temporal"
+	auth "github.com/formancehq/go-libs/v5/pkg/authn/jwt"
+	"github.com/formancehq/go-libs/v5/pkg/authn/licence"
+	"github.com/formancehq/go-libs/v5/pkg/cloud/aws/iam"
+	"github.com/formancehq/go-libs/v5/pkg/fx/servicefx"
+	"github.com/formancehq/go-libs/v5/pkg/fx/transportfx"
+	"github.com/formancehq/go-libs/v5/pkg/messaging/publish"
+	otlpmetrics "github.com/formancehq/go-libs/v5/pkg/observe/metrics"
+	"github.com/formancehq/go-libs/v5/pkg/service"
+	"github.com/formancehq/go-libs/v5/pkg/service/health"
+	bunconnect "github.com/formancehq/go-libs/v5/pkg/storage/bun/connect"
+	"github.com/formancehq/go-libs/v5/pkg/transport/httpserver"
+	"github.com/formancehq/go-libs/v5/pkg/workflow/temporal"
 	"github.com/formancehq/orchestration/internal/api"
 	v1 "github.com/formancehq/orchestration/internal/api/v1"
 	v2 "github.com/formancehq/orchestration/internal/api/v2"
@@ -25,8 +27,8 @@ import (
 
 func healthCheckModule() fx.Option {
 	return fx.Options(
-		health.Module(),
-		health.ProvideHealthCheck(func() health.NamedCheck {
+		servicefx.HealthModule(),
+		servicefx.ProvideHealthCheck(func() health.NamedCheck {
 			return health.NewNamedCheck("default", health.CheckFn(func(ctx context.Context) error {
 				return nil
 			}))
@@ -64,12 +66,16 @@ func newServeCommand() *cobra.Command {
 				}),
 				api.NewModule(service.IsDebug(cmd)),
 				fx.Invoke(func(lc fx.Lifecycle, router *chi.Mux) {
-					lc.Append(httpserver.NewHook(router, httpserver.WithAddress(listen)))
+					lc.Append(transportfx.FXHook(httpserver.NewHook(router, httpserver.WithAddress(listen))))
 				}),
 			}
 			worker, _ := cmd.Flags().GetBool(workerFlag)
 			if worker {
-				options = append(options, workerOptions(cmd))
+				workerOptions, err := workerOptions(cmd)
+				if err != nil {
+					return err
+				}
+				options = append(options, workerOptions)
 			}
 
 			return service.New(cmd.OutOrStdout(), options...).Run(cmd)

cmd/worker.go

@@ -1,16 +1,19 @@
 package cmd
 
 import (
+	"fmt"
+	"math"
 	"net/http"
+	"strconv"
 
 	sdk "github.com/formancehq/formance-sdk-go/v3"
-	"github.com/formancehq/go-libs/v3/aws/iam"
-	"github.com/formancehq/go-libs/v3/bun/bunconnect"
-	"github.com/formancehq/go-libs/v3/licence"
-	"github.com/formancehq/go-libs/v3/otlp/otlpmetrics"
-	"github.com/formancehq/go-libs/v3/publish"
-	"github.com/formancehq/go-libs/v3/service"
-	"github.com/formancehq/go-libs/v3/temporal"
+	"github.com/formancehq/go-libs/v5/pkg/authn/licence"
+	"github.com/formancehq/go-libs/v5/pkg/cloud/aws/iam"
+	"github.com/formancehq/go-libs/v5/pkg/messaging/publish"
+	otlpmetrics "github.com/formancehq/go-libs/v5/pkg/observe/metrics"
+	"github.com/formancehq/go-libs/v5/pkg/service"
+	bunconnect "github.com/formancehq/go-libs/v5/pkg/storage/bun/connect"
+	"github.com/formancehq/go-libs/v5/pkg/workflow/temporal"
 	"github.com/formancehq/orchestration/internal/temporalworker"
 	"github.com/formancehq/orchestration/internal/triggers"
 	"github.com/spf13/cobra"
@@ -22,26 +25,40 @@ func stackClientModule(cmd *cobra.Command) fx.Option {
 	stackURL, _ := cmd.Flags().GetString(stackURLFlag)
 
 	return fx.Options(
-		fx.Provide(func(httpClient *http.Client) *sdk.Formance {
+		fx.Provide(fx.Annotate(func(httpClient *http.Client) *sdk.Formance {
 			return sdk.New(
 				sdk.WithClient(httpClient),
 				sdk.WithServerURL(stackURL),
 			)
-		}),
+		}, fx.ParamTags(`name:"stack"`))),
 	)
 }
 
-func workerOptions(cmd *cobra.Command) fx.Option {
+func workerOptions(cmd *cobra.Command) (fx.Option, error) {
 
 	stack, _ := cmd.Flags().GetString(stackFlag)
 	temporalTaskQueue, _ := cmd.Flags().GetString(temporal.TemporalTaskQueueFlag)
-	temporalMaxParallelActivities, _ := cmd.Flags().GetInt(temporal.TemporalMaxParallelActivitiesFlag)
+	// The flag is registered as a float64 in go-libs; reading it with GetInt
+	// silently fails and yields 0, so the configured limit was never applied.
+	temporalMaxParallelActivities, err := cmd.Flags().GetFloat64(temporal.TemporalMaxParallelActivitiesFlag)
+	if err != nil {
+		return nil, err
+	}
+	maxIntExclusive := math.Exp2(float64(strconv.IntSize - 1))
+	if temporalMaxParallelActivities <= 0 ||
+		math.Trunc(temporalMaxParallelActivities) != temporalMaxParallelActivities ||
+		temporalMaxParallelActivities >= maxIntExclusive {
+		return nil, fmt.Errorf("%s must be a positive whole number", temporal.TemporalMaxParallelActivitiesFlag)
+	}
 	topics, _ := cmd.Flags().GetStringSlice(topicsFlag)
 
 	return fx.Options(
 		stackClientModule(cmd),
 		temporalworker.NewWorkerModule(temporalTaskQueue, worker.Options{
-			TaskQueueActivitiesPerSecond: float64(temporalMaxParallelActivities),
+			// "max parallel activities" caps concurrency, which maps to
+			// MaxConcurrentActivityExecutionSize, not the queue-wide rate limit
+			// TaskQueueActivitiesPerSecond it was previously wired to.
+			MaxConcurrentActivityExecutionSize: int(temporalMaxParallelActivities),
 		}),
 		triggers.NewListenerModule(
 			stack,
@@ -50,7 +67,7 @@ func workerOptions(cmd *cobra.Command) fx.Option {
 			true,
 			topics,
 		),
-	)
+	), nil
 }
 
 func newWorkerCommand() *cobra.Command {
@@ -61,8 +78,12 @@ func newWorkerCommand() *cobra.Command {
 			if err != nil {
 				return err
 			}
+			workerOptions, err := workerOptions(cmd)
+			if err != nil {
+				return err
+			}
 
-			return service.New(cmd.OutOrStdout(), commonOptions, workerOptions(cmd)).Run(cmd)
+			return service.New(cmd.OutOrStdout(), commonOptions, workerOptions).Run(cmd)
 		},
 	}
 	ret.Flags().String(stackURLFlag, "", "Stack url")

cmd/worker_test.go

@@ -0,0 +1,33 @@
+package cmd
+
+import (
+	"testing"
+
+	"github.com/formancehq/go-libs/v5/pkg/workflow/temporal"
+	"github.com/stretchr/testify/require"
+)
+
+func TestWorkerOptionsValidatesMaxParallelActivities(t *testing.T) {
+	for _, testCase := range []struct {
+		name    string
+		value   string
+		wantErr bool
+	}{
+		{name: "positive integer", value: "10"},
+		{name: "zero", value: "0", wantErr: true},
+		{name: "negative", value: "-1", wantErr: true},
+		{name: "fractional", value: "0.5", wantErr: true},
+		{name: "int overflow", value: "9223372036854775808", wantErr: true},
+	} {
+		t.Run(testCase.name, func(t *testing.T) {
+			cmd := newWorkerCommand()
+			require.NoError(t, cmd.Flags().Set(temporal.TemporalMaxParallelActivitiesFlag, testCase.value))
+			_, err := workerOptions(cmd)
+			if testCase.wantErr {
+				require.Error(t, err)
+				return
+			}
+			require.NoError(t, err)
+		})
+	}
+}