feat: add ledger partitionning (#867)

Earthfile

@@ -86,6 +86,7 @@ bench:
     COPY (+sources/*) /src
     WORKDIR /src/components/ledger/internal/storage/ledgerstore
     ARG numberOfTransactions=10000
+    ARG ledgers=10
     ARG benchTime=1s
     ARG count=1
     ARG GOPROXY
@@ -104,6 +105,7 @@ bench:
             go test -timeout $testTimeout -bench=$bench -run ^$ $additionalArgs \
             -benchtime=$benchTime \
             -count=$count \
+            -ledgers=$ledgers \
             -transactions=$numberOfTransactions | tee -a /results.txt
     END
     RUN benchstat /results.txt

cmd/buckets.go

@@ -0,0 +1,72 @@
+package cmd
+
+import (
+	"github.com/formancehq/ledger/internal/storage"
+	"github.com/formancehq/ledger/internal/storage/driver"
+	"github.com/formancehq/stack/libs/go-libs/logging"
+	"github.com/formancehq/stack/libs/go-libs/service"
+	"github.com/spf13/cobra"
+	"github.com/spf13/viper"
+)
+
+func NewBucket() *cobra.Command {
+	return &cobra.Command{
+		Use:     "buckets",
+		Aliases: []string{"storage"},
+	}
+}
+
+func NewBucketUpgrade() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:          "upgrade",
+		Args:         cobra.ExactArgs(1),
+		SilenceUsage: true,
+		RunE: func(cmd *cobra.Command, args []string) error {
+
+			driver := driver.New(storage.ConnectionOptionsFromFlags(viper.GetViper(), cmd.OutOrStdout(), viper.GetBool(service.DebugFlag)))
+			if err := driver.Initialize(cmd.Context()); err != nil {
+				return err
+			}
+			defer func() {
+				_ = driver.Close()
+			}()
+
+			name := args[0]
+
+			bucket, err := driver.OpenBucket(name)
+			if err != nil {
+				return err
+			}
+
+			logger := service.GetDefaultLogger(cmd.OutOrStdout(), viper.GetBool(service.DebugFlag), false)
+
+			return bucket.Migrate(logging.ContextWithLogger(cmd.Context(), logger))
+		},
+	}
+	return cmd
+}
+
+func NewBucketUpgradeAll() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:          "upgrade-all",
+		Args:         cobra.ExactArgs(0),
+		SilenceUsage: true,
+		RunE: func(cmd *cobra.Command, args []string) error {
+
+			logger := service.GetDefaultLogger(cmd.OutOrStdout(), viper.GetBool(service.DebugFlag), false)
+			ctx := logging.ContextWithLogger(cmd.Context(), logger)
+
+			driver := driver.New(storage.ConnectionOptionsFromFlags(viper.GetViper(), cmd.OutOrStdout(), viper.GetBool(service.DebugFlag)))
+			defer func() {
+				_ = driver.Close()
+			}()
+
+			if err := driver.Initialize(ctx); err != nil {
+				return err
+			}
+
+			return driver.UpgradeAllBuckets(ctx)
+		},
+	}
+	return cmd
+}

cmd/root.go

@@ -37,18 +37,12 @@ func NewRootCommand() *cobra.Command {
 	serve := NewServe()
 	version := NewVersion()
 
-	conf := NewConfig()
-	conf.AddCommand(NewConfigInit())
-	store := NewStorage()
-	store.AddCommand(NewStorageInit())
-	store.AddCommand(NewStorageList())
-	store.AddCommand(NewStorageUpgrade())
-	store.AddCommand(NewStorageUpgradeAll())
-	store.AddCommand(NewStorageDelete())
+	buckets := NewBucket()
+	buckets.AddCommand(NewBucketUpgrade())
+	buckets.AddCommand(NewBucketUpgradeAll())
 
 	root.AddCommand(serve)
-	root.AddCommand(conf)
-	root.AddCommand(store)
+	root.AddCommand(buckets)
 	root.AddCommand(version)
 
 	root.AddCommand(NewDocCommand())

cmd/serve.go

@@ -40,7 +40,7 @@ func NewServe() *cobra.Command {
 				fx.Invoke(func(lc fx.Lifecycle, driver *driver.Driver) {
 					if viper.GetBool(autoUpgradeFlag) {
 						lc.Append(fx.Hook{
-							OnStart: driver.UpgradeAllLedgersSchemas,
+							OnStart: driver.UpgradeAllBuckets,
 						})
 					}
 				}),