feat: add log export

Author: gfyrag

Justfile

@@ -17,7 +17,9 @@ tidy:
   @cd {{justfile_directory()}}/deployments/pulumi && go mod tidy
 
 generate:
+  @rm $(find ./internal -name '*_generated_test.go') || true
   @go generate ./...
+g: generate
 
 export-docs-events:
   @go run . docs events --write-dir docs/events
@@ -35,7 +37,7 @@ openapi:
   @yq eval-all '. as $item ireduce ({}; . * $item)' openapi/v1.yaml openapi/v2.yaml openapi/overlay.yaml > openapi.yaml
   @npx -y widdershins {{justfile_directory()}}/openapi/v2.yaml -o {{justfile_directory()}}/docs/api/README.md --search false --language_tabs 'http:HTTP' --summary --omitHeader
 
-generate-client:
+generate-client: openapi
   @speakeasy generate sdk -s openapi.yaml -o ./pkg/client -l go
 
 release-local:

cmd/buckets.go

@@ -4,7 +4,7 @@ import (
 	"github.com/spf13/cobra"
 )
 
-func NewBucket() *cobra.Command {
+func NewBucketsCommand() *cobra.Command {
 	ret := &cobra.Command{
 		Use:     "buckets",
 		Aliases: []string{"storage"},

cmd/docs_events.go

@@ -3,7 +3,7 @@ package cmd
 import (
 	"encoding/json"
 	"fmt"
-	"github.com/formancehq/ledger/internal/bus"
+	"github.com/formancehq/ledger/pkg/events"
 	"github.com/invopop/jsonschema"
 	"github.com/spf13/cobra"
 	"os"
@@ -30,10 +30,10 @@ func NewDocEventsCommand() *cobra.Command {
 			}
 
 			for _, o := range []any{
-				bus.CommittedTransactions{},
-				bus.DeletedMetadata{},
-				bus.SavedMetadata{},
-				bus.RevertedTransaction{},
+				events.CommittedTransactions{},
+				events.DeletedMetadata{},
+				events.SavedMetadata{},
+				events.RevertedTransaction{},
 			} {
 				schema := jsonschema.Reflect(o)
 				data, err := json.MarshalIndent(schema, "", "  ")

cmd/root.go

@@ -31,14 +31,12 @@ func NewRootCommand() *cobra.Command {
 		Version:           Version,
 	}
 
-	serve := NewServeCommand()
-	version := NewVersion()
-
-	buckets := NewBucket()
+	root.AddCommand(NewServeCommand())
+	root.AddCommand(NewBucketsCommand())
+	root.AddCommand(NewVersionCommand())
+	root.AddCommand(NewWorkerCommand())
+	root.AddCommand(NewDocsCommand())
 
-	root.AddCommand(serve)
-	root.AddCommand(buckets)
-	root.AddCommand(version)
 	root.AddCommand(bunmigrate.NewDefaultCommand(func(cmd *cobra.Command, _ []string, db *bun.DB) error {
 		logger := logging.NewDefaultLogger(cmd.OutOrStdout(), service.IsDebug(cmd), false, false)
 		cmd.SetContext(logging.ContextWithLogger(cmd.Context(), logger))
@@ -54,7 +52,6 @@ func NewRootCommand() *cobra.Command {
 
 		return driver.UpgradeAllBuckets(cmd.Context())
 	}))
-	root.AddCommand(NewDocsCommand())
 
 	return root
 }

cmd/serve.go

@@ -1,13 +1,15 @@
 package cmd
 
 import (
-	"net/http"
-	"net/http/pprof"
-	"time"
-
 	"github.com/formancehq/go-libs/v2/logging"
 	"github.com/formancehq/ledger/internal/api/common"
+	"github.com/formancehq/ledger/internal/replication/drivers"
+	"github.com/formancehq/ledger/internal/replication/drivers/all"
+	"github.com/formancehq/ledger/internal/replication/runner"
 	systemstore "github.com/formancehq/ledger/internal/storage/system"
+	"net/http"
+	"net/http/pprof"
+	"time"
 
 	apilib "github.com/formancehq/go-libs/v2/api"
 	"github.com/formancehq/go-libs/v2/health"
@@ -48,6 +50,7 @@ const (
 	NumscriptInterpreterFlagsToPass = "numscript-interpreter-flags"
 	DefaultPageSizeFlag             = "default-page-size"
 	MaxPageSizeFlag                 = "max-page-size"
+	WorkerEnabledFlag               = "worker"
 )
 
 func NewServeCommand() *cobra.Command {
@@ -98,6 +101,8 @@ func NewServeCommand() *cobra.Command {
 				auth.FXModuleFromFlags(cmd),
 				bunconnect.Module(*connectionOptions, service.IsDebug(cmd)),
 				storage.NewFXModule(serveConfiguration.autoUpgrade),
+				drivers.NewFXModule(),
+				fx.Invoke(all.Register),
 				systemcontroller.NewFXModule(systemcontroller.ModuleConfiguration{
 					NumscriptInterpreter:      numscriptInterpreter,
 					NumscriptInterpreterFlags: numscriptInterpreterFlags,
@@ -150,6 +155,15 @@ func NewServeCommand() *cobra.Command {
 				}),
 			}
 
+			workerEnabled, _ := cmd.Flags().GetBool(WorkerEnabledFlag)
+			if workerEnabled {
+				options = append(options, runner.NewFXModule(runner.ModuleConfig{
+					SyncPeriod:      serveConfiguration.Worker.pipelinesSyncPeriod,
+					PullInterval:    serveConfiguration.Worker.pipelinesPullInterval,
+					PushRetryPeriod: serveConfiguration.Worker.pipelinesPushRetryPeriod,
+				}))
+			}
+
 			return service.New(cmd.OutOrStdout(), options...).Run(cmd)
 		},
 	}
@@ -164,7 +178,9 @@ func NewServeCommand() *cobra.Command {
 	cmd.Flags().String(NumscriptInterpreterFlagsToPass, "", "Feature flags to pass to the experimental numscript interpreter")
 	cmd.Flags().Uint64(MaxPageSizeFlag, 100, "Max page size")
 	cmd.Flags().Uint64(DefaultPageSizeFlag, 15, "Default page size")
+	cmd.Flags().Bool(WorkerEnabledFlag, false, "Enable worker")
 
+	addWorkerFlags(cmd)
 	service.AddFlags(cmd.Flags())
 	bunconnect.AddFlags(cmd.Flags())
 	otlpmetrics.AddFlags(cmd.Flags())
@@ -183,6 +199,7 @@ type serveConfiguration struct {
 	numscriptCacheMaxCount uint
 	autoUpgrade            bool
 	bind                   string
+	Worker                 workerConfiguration
 }
 
 func discoverServeConfiguration(cmd *cobra.Command) serveConfiguration {
@@ -192,6 +209,8 @@ func discoverServeConfiguration(cmd *cobra.Command) serveConfiguration {
 	ret.autoUpgrade, _ = cmd.Flags().GetBool(AutoUpgradeFlag)
 	ret.bind, _ = cmd.Flags().GetString(BindFlag)
 
+	ret.Worker = discoverWorkerConfiguration(cmd)
+
 	return ret
 }
 

cmd/version.go

@@ -12,7 +12,7 @@ func PrintVersion(_ *cobra.Command, _ []string) {
 	fmt.Printf("Commit: %s \n", Commit)
 }
 
-func NewVersion() *cobra.Command {
+func NewVersionCommand() *cobra.Command {
 	return &cobra.Command{
 		Use:   "version",
 		Short: "Get version",

cmd/worker.go

@@ -0,0 +1,82 @@
+package cmd
+
+import (
+	"github.com/formancehq/go-libs/v2/bun/bunconnect"
+	"github.com/formancehq/go-libs/v2/otlp"
+	"github.com/formancehq/go-libs/v2/otlp/otlpmetrics"
+	"github.com/formancehq/go-libs/v2/otlp/otlptraces"
+	"github.com/formancehq/go-libs/v2/service"
+	"github.com/formancehq/ledger/internal/replication/drivers"
+	"github.com/formancehq/ledger/internal/replication/drivers/all"
+	"github.com/formancehq/ledger/internal/replication/runner"
+	"github.com/formancehq/ledger/internal/storage"
+	"github.com/spf13/cobra"
+	"go.uber.org/fx"
+	"time"
+)
+
+const (
+	PipelinesSyncPeriodFlag      = "pipelines-sync-period"
+	PipelinesPullIntervalFlag    = "pipelines-pull-interval"
+	PipelinesPushRetryPeriodFlag = "pipelines-push-retry-period"
+)
+
+func addWorkerFlags(cmd *cobra.Command) {
+	cmd.Flags().Duration(PipelinesSyncPeriodFlag, 5*time.Second, "Pipelines sync period")
+	cmd.Flags().Duration(PipelinesPullIntervalFlag, runner.DefaultPullInterval, "Pipelines pull interval")
+	cmd.Flags().Duration(PipelinesPushRetryPeriodFlag, runner.DefaultPushRetryPeriod, "Pipelines push retry period")
+}
+
+type workerConfiguration struct {
+	pipelinesSyncPeriod      time.Duration
+	pipelinesPullInterval    time.Duration
+	pipelinesPushRetryPeriod time.Duration
+}
+
+func discoverWorkerConfiguration(cmd *cobra.Command) workerConfiguration {
+	ret := workerConfiguration{}
+	ret.pipelinesSyncPeriod, _ = cmd.Flags().GetDuration(PipelinesSyncPeriodFlag)
+	ret.pipelinesPullInterval, _ = cmd.Flags().GetDuration(PipelinesPullIntervalFlag)
+	ret.pipelinesPushRetryPeriod, _ = cmd.Flags().GetDuration(PipelinesPushRetryPeriodFlag)
+
+	return ret
+}
+
+func NewWorkerCommand() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:          "worker",
+		SilenceUsage: true,
+		RunE: func(cmd *cobra.Command, _ []string) error {
+			configuration := discoverWorkerConfiguration(cmd)
+
+			connectionOptions, err := bunconnect.ConnectionOptionsFromFlags(cmd)
+			if err != nil {
+				return err
+			}
+
+			return service.New(cmd.OutOrStdout(),
+				fx.NopLogger,
+				otlp.FXModuleFromFlags(cmd),
+				otlptraces.FXModuleFromFlags(cmd),
+				otlpmetrics.FXModuleFromFlags(cmd),
+				bunconnect.Module(*connectionOptions, service.IsDebug(cmd)),
+				storage.NewFXModule(false),
+				drivers.NewFXModule(),
+				fx.Invoke(all.Register),
+				runner.NewFXModule(runner.ModuleConfig{
+					SyncPeriod:      configuration.pipelinesSyncPeriod,
+					PullInterval:    configuration.pipelinesPullInterval,
+					PushRetryPeriod: configuration.pipelinesPushRetryPeriod,
+				}),
+			).Run(cmd)
+		},
+	}
+
+	addWorkerFlags(cmd)
+	service.AddFlags(cmd.Flags())
+	bunconnect.AddFlags(cmd.Flags())
+	otlpmetrics.AddFlags(cmd.Flags())
+	otlptraces.AddFlags(cmd.Flags())
+
+	return cmd
+}

deployments/pulumi/main.go

@@ -3,6 +3,7 @@ package main
 import (
 	"errors"
 	"fmt"
+	"github.com/formancehq/go-libs/v2/pointer"
 	"github.com/formancehq/ledger/deployments/pulumi/pkg"
 	"github.com/pulumi/pulumi/sdk/v3/go/pulumi"
 	"github.com/pulumi/pulumi/sdk/v3/go/pulumi/config"
@@ -44,10 +45,11 @@ func deploy(ctx *pulumi.Context) error {
 		Postgres: pulumi_ledger.PostgresArgs{
 			URI: pulumi.String(postgresURI),
 		},
-		Debug:                pulumi.Bool(conf.GetBool("debug")),
-		ReplicaCount:         pulumi.Int(conf.GetInt("replicaCount")),
-		ExperimentalFeatures: pulumi.Bool(conf.GetBool("experimentalFeatures")),
-		Upgrade: pulumix.Val(pulumi_ledger.UpgradeMode(config.Get(ctx, "upgrade-mode"))),
+		Debug: pulumi.Bool(conf.GetBool("debug")),
+		API: pulumi_ledger.APIArgs{
+			ReplicaCount:         pulumix.Val(pointer.For(conf.GetInt("replicaCount"))),
+			ExperimentalFeatures: pulumi.Bool(conf.GetBool("experimentalFeatures")),
+		},
 	})
 
 	return err