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,8 +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"
 	systemstore "github.com/formancehq/ledger/internal/storage/system"
-	"github.com/formancehq/ledger/internal/worker"
 	"net/http"
 	"net/http/pprof"
 	"time"
@@ -56,7 +57,10 @@ func NewServeCommand() *cobra.Command {
 		Use:          "serve",
 		SilenceUsage: true,
 		RunE: func(cmd *cobra.Command, _ []string) error {
-			serveConfiguration := discoverServeConfiguration(cmd)
+			serveConfiguration, err := discoverServeConfiguration(cmd)
+			if err != nil {
+				return err
+			}
 
 			connectionOptions, err := bunconnect.ConnectionOptionsFromFlags(cmd)
 			if err != nil {
@@ -101,6 +105,8 @@ func NewServeCommand() *cobra.Command {
 				storage.NewFXModule(storage.ModuleConfig{
 					AutoUpgrade: serveConfiguration.autoUpgrade,
 				}),
+				drivers.NewFXModule(),
+				fx.Invoke(all.Register),
 				systemcontroller.NewFXModule(systemcontroller.ModuleConfiguration{
 					NumscriptInterpreter:      numscriptInterpreter,
 					NumscriptInterpreterFlags: numscriptInterpreterFlags,
@@ -129,15 +135,15 @@ func NewServeCommand() *cobra.Command {
 				}),
 				fx.Decorate(func(
 					params struct {
-					fx.In
+						fx.In
 
-					Handler          chi.Router
-					HealthController *health.HealthController
-					Logger           logging.Logger
+						Handler          chi.Router
+						HealthController *health.HealthController
+						Logger           logging.Logger
 
-					MeterProvider *metric.MeterProvider         `optional:"true"`
-					Exporter      *otlpmetrics.InMemoryExporter `optional:"true"`
-				},
+						MeterProvider *metric.MeterProvider         `optional:"true"`
+						Exporter      *otlpmetrics.InMemoryExporter `optional:"true"`
+					},
 				) chi.Router {
 					return assembleFinalRouter(
 						service.IsDebug(cmd),
@@ -155,10 +161,7 @@ func NewServeCommand() *cobra.Command {
 
 			workerEnabled, _ := cmd.Flags().GetBool(WorkerEnabledFlag)
 			if workerEnabled {
-				options = append(options, worker.NewFXModule(worker.ModuleConfig{
-					Schedule:     serveConfiguration.workerConfiguration.hashLogsBlockCRONSpec,
-					MaxBlockSize: serveConfiguration.workerConfiguration.hashLogsBlockMaxSize,
-				}))
+				options = append(options, newWorkerModule(serveConfiguration.workerConfiguration))
 			}
 
 			return service.New(cmd.OutOrStdout(), options...).Run(cmd)
@@ -199,15 +202,20 @@ type serveConfiguration struct {
 	workerConfiguration    workerConfiguration
 }
 
-func discoverServeConfiguration(cmd *cobra.Command) serveConfiguration {
-	ret := serveConfiguration{}
+func discoverServeConfiguration(cmd *cobra.Command) (*serveConfiguration, error) {
+	ret := &serveConfiguration{}
 	ret.ballastSize, _ = cmd.Flags().GetUint(BallastSizeInBytesFlag)
 	ret.numscriptCacheMaxCount, _ = cmd.Flags().GetUint(NumscriptCacheMaxCountFlag)
 	ret.autoUpgrade, _ = cmd.Flags().GetBool(AutoUpgradeFlag)
 	ret.bind, _ = cmd.Flags().GetString(BindFlag)
-	ret.workerConfiguration = discoverWorkerConfiguration(cmd)
 
-	return ret
+	workerConfiguration, err := discoverWorkerConfiguration(cmd)
+	if err != nil {
+		return nil, err
+	}
+	ret.workerConfiguration = *workerConfiguration
+
+	return ret, nil
 }
 
 func assembleFinalRouter(

cmd/worker.go

@@ -6,31 +6,57 @@ import (
 	"github.com/formancehq/go-libs/v2/otlp/otlpmetrics"
 	"github.com/formancehq/go-libs/v2/otlp/otlptraces"
 	"github.com/formancehq/go-libs/v2/service"
+	replication "github.com/formancehq/ledger/internal/replication"
+	"github.com/formancehq/ledger/internal/replication/drivers"
+	"github.com/formancehq/ledger/internal/replication/drivers/all"
 	"github.com/formancehq/ledger/internal/storage"
 	"github.com/formancehq/ledger/internal/worker"
+	"github.com/robfig/cron/v3"
 	"github.com/spf13/cobra"
 	"go.uber.org/fx"
+	"time"
 )
 
 const (
+	WorkerPipelinesSyncPeriodFlag      = "worker-pipelines-sync-period"
+	WorkerPipelinesPullIntervalFlag    = "worker-pipelines-pull-interval"
+	WorkerPipelinesPushRetryPeriodFlag = "worker-pipelines-push-retry-period"
+
 	WorkerAsyncBlockHasherMaxBlockSizeFlag = "worker-async-block-hasher-max-block-size"
 	WorkerAsyncBlockHasherScheduleFlag     = "worker-async-block-hasher-schedule"
 )
 
 type workerConfiguration struct {
-	hashLogsBlockMaxSize   int
-	hashLogsBlockCRONSpec  string
+	AsyncBlockRunnerConfig storage.AsyncBlockRunnerConfig
+	ReplicationConfig 	replication.ModuleConfig
 }
 
-func discoverWorkerConfiguration(cmd *cobra.Command) workerConfiguration {
-	ret := workerConfiguration{}
-	ret.hashLogsBlockCRONSpec, _ = cmd.Flags().GetString(WorkerAsyncBlockHasherScheduleFlag)
-	ret.hashLogsBlockMaxSize, _ = cmd.Flags().GetInt(WorkerAsyncBlockHasherMaxBlockSizeFlag)
+func discoverWorkerConfiguration(cmd *cobra.Command) (*workerConfiguration, error) {
+	ret := &workerConfiguration{}
+	hashLogsBlockCRONSpec, _ := cmd.Flags().GetString(WorkerAsyncBlockHasherScheduleFlag)
+	if hashLogsBlockCRONSpec == "" {
+		hashLogsBlockCRONSpec = "0 * * * * *"
+	}
+	parser := cron.NewParser(cron.Second | cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow)
+
+	var err error
+	ret.AsyncBlockRunnerConfig.Schedule, err = parser.Parse(hashLogsBlockCRONSpec)
+	if err != nil {
+		return nil, err
+	}
+
+	ret.AsyncBlockRunnerConfig.MaxBlockSize, _ = cmd.Flags().GetInt(WorkerAsyncBlockHasherMaxBlockSizeFlag)
+	ret.ReplicationConfig.SyncPeriod, _ = cmd.Flags().GetDuration(WorkerPipelinesSyncPeriodFlag)
+	ret.ReplicationConfig.PullInterval, _ = cmd.Flags().GetDuration(WorkerPipelinesPullIntervalFlag)
+	ret.ReplicationConfig.PushRetryPeriod, _ = cmd.Flags().GetDuration(WorkerPipelinesPushRetryPeriodFlag)
 
-	return ret
+	return ret, nil
 }
 
 func addWorkerFlags(cmd *cobra.Command) {
+	cmd.Flags().Duration(WorkerPipelinesSyncPeriodFlag, 5*time.Second, "Pipelines sync period")
+	cmd.Flags().Duration(WorkerPipelinesPullIntervalFlag, replication.DefaultPullInterval, "Pipelines pull interval")
+	cmd.Flags().Duration(WorkerPipelinesPushRetryPeriodFlag, replication.DefaultPushRetryPeriod, "Pipelines push retry period")
 	cmd.Flags().Int(WorkerAsyncBlockHasherMaxBlockSizeFlag, 1000, "Max block size")
 	cmd.Flags().String(WorkerAsyncBlockHasherScheduleFlag, "0 * * * * *", "Schedule")
 }
@@ -45,7 +71,10 @@ func NewWorkerCommand() *cobra.Command {
 				return err
 			}
 
-			workerConfiguration := discoverWorkerConfiguration(cmd)
+			workerConfiguration, err := discoverWorkerConfiguration(cmd)
+			if err != nil {
+				return err
+			}
 
 			return service.New(cmd.OutOrStdout(),
 				fx.NopLogger,
@@ -54,10 +83,9 @@ func NewWorkerCommand() *cobra.Command {
 				otlpmetrics.FXModuleFromFlags(cmd),
 				bunconnect.Module(*connectionOptions, service.IsDebug(cmd)),
 				storage.NewFXModule(storage.ModuleConfig{}),
-				worker.NewFXModule(worker.ModuleConfig{
-					MaxBlockSize: workerConfiguration.hashLogsBlockMaxSize,
-					Schedule: workerConfiguration.hashLogsBlockCRONSpec,
-				}),
+				drivers.NewFXModule(),
+				fx.Invoke(all.Register),
+				newWorkerModule(*workerConfiguration),
 			).Run(cmd)
 		},
 	}
@@ -71,6 +99,9 @@ func NewWorkerCommand() *cobra.Command {
 	return cmd
 }
 
-func newWorkerModule() fx.Option {
-	return fx.Options()
-}
+func newWorkerModule(configuration workerConfiguration) fx.Option {
+	return worker.NewFXModule(worker.ModuleConfig{
+		AsyncBlockRunnerConfig: configuration.AsyncBlockRunnerConfig,
+		ReplicationConfig: configuration.ReplicationConfig,
+	})
+}