feat: add log export

Author: gfyrag

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/serve.go

@@ -3,6 +3,9 @@ package cmd
 import (
 	"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"
@@ -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,
@@ -152,7 +157,11 @@ func NewServeCommand() *cobra.Command {
 
 			workerEnabled, _ := cmd.Flags().GetBool(WorkerEnabledFlag)
 			if workerEnabled {
-				options = append(options, newWorkerModule())
+				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)
@@ -190,6 +199,7 @@ type serveConfiguration struct {
 	numscriptCacheMaxCount uint
 	autoUpgrade            bool
 	bind                   string
+	Worker                 workerConfiguration
 }
 
 func discoverServeConfiguration(cmd *cobra.Command) serveConfiguration {
@@ -199,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/worker.go

@@ -6,19 +6,49 @@ import (
 	"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
@@ -31,7 +61,7 @@ func NewWorkerCommand() *cobra.Command {
 				otlpmetrics.FXModuleFromFlags(cmd),
 				bunconnect.Module(*connectionOptions, service.IsDebug(cmd)),
 				storage.NewFXModule(false),
-				newWorkerModule(),
+				newWorkerModule(configuration),
 			).Run(cmd)
 		},
 	}
@@ -45,6 +75,14 @@ func NewWorkerCommand() *cobra.Command {
 	return cmd
 }
 
-func newWorkerModule() fx.Option {
-	return fx.Options()
+func newWorkerModule(configuration workerConfiguration) fx.Option {
+	return fx.Options(
+		drivers.NewFXModule(),
+		fx.Invoke(all.Register),
+		runner.NewFXModule(runner.ModuleConfig{
+			SyncPeriod:      configuration.pipelinesSyncPeriod,
+			PullInterval:    configuration.pipelinesPullInterval,
+			PushRetryPeriod: configuration.pipelinesPushRetryPeriod,
+		}),
+	)
 }