feat(ledger): make stateless

Earthfile

@@ -60,16 +60,14 @@ tests:
     LET goFlags="-race"
     IF [ "$coverage" = "true" ]
         SET goFlags="$goFlags -covermode=atomic"
-        SET goFlags="$goFlags -coverpkg=github.com/formancehq/stack/components/ledger/internal/..."
-        SET goFlags="$goFlags,github.com/formancehq/stack/components/ledger/cmd/..."
+        SET goFlags="$goFlags -coverpkg=github.com/formancehq/ledger/internal/..."
+        SET goFlags="$goFlags,github.com/formancehq/ledger/pkg/events/..."
+        SET goFlags="$goFlags,github.com/formancehq/ledger/cmd/..."
         SET goFlags="$goFlags -coverprofile cover.out"
     END
     IF [ "$includeIntegrationTests" = "true" ]
         SET goFlags="$goFlags -tags it"
-        WITH DOCKER \
-            --pull=postgres:15-alpine \
-            --pull=clickhouse/clickhouse-server:head \
-            --pull=elasticsearch:8.14.3
+        WITH DOCKER --pull=postgres:15-alpine
             RUN --mount type=cache,id=gopkgcache,target=${GOPATH}/pkg/mod \
                 --mount type=cache,id=gobuildcache,target=/root/.cache/go-build \
                 ginkgo -r -p $goFlags
@@ -81,7 +79,8 @@ tests:
     END
     IF [ "$coverage" = "true" ]
         # exclude files suffixed with _generated.go, these are mocks used by tests
-        RUN cat cover.out | grep -v "_generated.go" > cover2.out
+        # toremovelater: also exclude machine code as it will be updated soon
+        RUN cat cover.out | grep -v "_generated.go" | grep -v "/machine/" > cover2.out
         RUN mv cover2.out cover.out
         SAVE ARTIFACT cover.out AS LOCAL cover.out
     END
@@ -192,3 +191,17 @@ generate-client:
     COPY --dir pkg/client client
     RUN --secret SPEAKEASY_API_KEY speakeasy generate sdk -s ./final.json -o ./client -l go
     SAVE ARTIFACT client AS LOCAL ./pkg/client
+
+export-database-schema:
+    FROM core+builder-image
+    RUN go install github.com/roerohan/wait-for-it@latest
+    COPY (+sources/*) /src
+    WORKDIR /src/components/ledger
+    COPY --dir scripts scripts
+    WITH DOCKER --pull postgres:15-alpine --pull schemaspy/schemaspy:6.2.4
+        RUN --mount type=cache,id=gopkgcache,target=${GOPATH}/pkg/mod \
+            --mount type=cache,id=gobuild,target=/root/.cache/go-build \
+            ./scripts/export-database-schema.sh
+    END
+    SAVE ARTIFACT docs/database AS LOCAL docs/database
+

cmd/buckets.go

@@ -1,70 +1,15 @@
 package cmd
 
 import (
-	"github.com/formancehq/go-libs/bun/bunconnect"
-	"github.com/formancehq/go-libs/logging"
-	"github.com/formancehq/go-libs/service"
-	"github.com/formancehq/ledger/internal/storage/driver"
 	"github.com/spf13/cobra"
 )
 
 func NewBucket() *cobra.Command {
-	return &cobra.Command{
+	ret := &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 {
-			connectionOptions, err := bunconnect.ConnectionOptionsFromFlags(cmd)
-			if err != nil {
-				return err
-			}
-
-			driver := driver.New(*connectionOptions)
-			if err := driver.Initialize(cmd.Context()); err != nil {
-				return err
-			}
-			defer func() {
-				_ = driver.Close()
-			}()
-
-			name := args[0]
-
-			bucket, err := driver.OpenBucket(cmd.Context(), name)
-			if err != nil {
-				return err
-			}
-
-			logger := logging.NewDefaultLogger(cmd.OutOrStdout(), service.IsDebug(cmd), false)
-
-			return bucket.Migrate(logging.ContextWithLogger(cmd.Context(), logger))
-		},
-	}
-	return cmd
-}
-
-func upgradeAll(cmd *cobra.Command, _ []string) error {
-	logger := logging.NewDefaultLogger(cmd.OutOrStdout(), service.IsDebug(cmd), false)
-	ctx := logging.ContextWithLogger(cmd.Context(), logger)
-
-	connectionOptions, err := bunconnect.ConnectionOptionsFromFlags(cmd)
-	if err != nil {
-		return err
-	}
-
-	driver := driver.New(*connectionOptions)
-	if err := driver.Initialize(ctx); err != nil {
-		return err
-	}
-	defer func() {
-		_ = driver.Close()
-	}()
 
-	return driver.UpgradeAllBuckets(ctx)
+	ret.AddCommand(NewBucketUpgrade())
+	return ret
 }

cmd/buckets_upgrade.go

@@ -0,0 +1,74 @@
+package cmd
+
+import (
+	"github.com/formancehq/go-libs/bun/bunconnect"
+	"github.com/formancehq/go-libs/logging"
+	"github.com/formancehq/go-libs/service"
+	"github.com/formancehq/ledger/internal/storage/driver"
+	"github.com/spf13/cobra"
+)
+
+func NewBucketUpgrade() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:          "upgrade",
+		Args:         cobra.ExactArgs(1),
+		SilenceUsage: true,
+		RunE: func(cmd *cobra.Command, args []string) error {
+			connectionOptions, err := bunconnect.ConnectionOptionsFromFlags(cmd)
+			if err != nil {
+				return err
+			}
+
+			db, err := bunconnect.OpenSQLDB(cmd.Context(), *connectionOptions)
+			if err != nil {
+				return err
+			}
+			defer func() {
+				_ = db.Close()
+			}()
+
+			driver := driver.New(db)
+			if err := driver.Initialize(cmd.Context()); err != nil {
+				return err
+			}
+
+			logger := logging.NewDefaultLogger(cmd.OutOrStdout(), service.IsDebug(cmd), false)
+
+			if args[0] == "*" {
+				return upgradeAll(cmd)
+			} else {
+				return driver.UpgradeBucket(logging.ContextWithLogger(cmd.Context(), logger), args[0])
+			}
+		},
+	}
+
+	service.AddFlags(cmd.Flags())
+	bunconnect.AddFlags(cmd.Flags())
+
+	return cmd
+}
+
+func upgradeAll(cmd *cobra.Command) error {
+	logger := logging.NewDefaultLogger(cmd.OutOrStdout(), service.IsDebug(cmd), false)
+	ctx := logging.ContextWithLogger(cmd.Context(), logger)
+
+	connectionOptions, err := bunconnect.ConnectionOptionsFromFlags(cmd)
+	if err != nil {
+		return err
+	}
+
+	db, err := bunconnect.OpenSQLDB(cmd.Context(), *connectionOptions)
+	if err != nil {
+		return err
+	}
+	defer func() {
+		_ = db.Close()
+	}()
+
+	driver := driver.New(db)
+	if err := driver.Initialize(ctx); err != nil {
+		return err
+	}
+
+	return driver.UpgradeAllBuckets(ctx)
+}

cmd/buckets_upgrade_test.go

@@ -0,0 +1,53 @@
+//go:build it
+
+package cmd
+
+import (
+	"github.com/formancehq/go-libs/bun/bunconnect"
+	"github.com/formancehq/go-libs/logging"
+	"github.com/formancehq/go-libs/testing/docker"
+	"github.com/formancehq/go-libs/testing/platform/pgtesting"
+	"github.com/stretchr/testify/require"
+	"io"
+	"testing"
+)
+
+func TestBucketsUpgrade(t *testing.T) {
+	t.Parallel()
+
+	dockerPool := docker.NewPool(t, logging.Testing())
+	srv := pgtesting.CreatePostgresServer(t, dockerPool)
+	ctx := logging.TestingContext()
+
+	type testCase struct {
+		name string
+		args []string
+	}
+
+	for _, tc := range []testCase{
+		{
+			name: "nominal",
+			args: []string{"test"},
+		},
+		{
+			name: "upgrade all",
+			args: []string{"*"},
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			db := srv.NewDatabase(t)
+
+			args := []string{
+				"--" + bunconnect.PostgresURIFlag, db.ConnString(),
+			}
+			args = append(args, tc.args...)
+
+			cmd := NewBucketUpgrade()
+			cmd.SetOut(io.Discard)
+			cmd.SetArgs(args)
+			require.NoError(t, cmd.ExecuteContext(ctx))
+		})
+	}
+}

cmd/doc.go

@@ -1,62 +1,13 @@
 package cmd
 
 import (
-	"fmt"
-	"os"
-	"sort"
-	"strings"
-	"text/tabwriter"
-
 	"github.com/spf13/cobra"
-	"github.com/spf13/pflag"
 )
 
 func NewDocCommand() *cobra.Command {
 	cmd := &cobra.Command{
 		Use: "doc",
 	}
-	cmd.AddCommand(NewDocFlagCommand())
-	return cmd
-}
-
-func NewDocFlagCommand() *cobra.Command {
-	cmd := &cobra.Command{
-		Use: "flags",
-		Run: func(cmd *cobra.Command, args []string) {
-
-			w := tabwriter.NewWriter(os.Stdout, 0, 0, 1, ' ', tabwriter.Debug)
-			defer func(w *tabwriter.Writer) {
-				if err := w.Flush(); err != nil {
-					panic(err)
-				}
-			}(w)
-
-			allKeys := make([]string, 0)
-			cmd.Flags().VisitAll(func(f *pflag.Flag) {
-				allKeys = append(allKeys, f.Name)
-			})
-			sort.Strings(allKeys)
-
-			if _, err := fmt.Fprintf(w,
-				"\tFlag\tEnv var\tDefault value\tDescription\t\r\n"); err != nil {
-				panic(err)
-			}
-			if _, err := fmt.Fprintf(w,
-				"\t-\t-\t-\t-\t\r\n"); err != nil {
-				panic(err)
-			}
-			for _, key := range allKeys {
-				asEnvVar := strings.ToUpper(strings.Replace(key, "-", "_", -1))
-				flag := cmd.Parent().Parent().PersistentFlags().Lookup(key)
-				if flag == nil {
-					continue
-				}
-				if _, err := fmt.Fprintf(w,
-					"\t--%s\t%s\t%s\t%s\t\r\n", key, asEnvVar, flag.DefValue, flag.Usage); err != nil {
-					panic(err)
-				}
-			}
-		},
-	}
+	cmd.AddCommand(NewDocFlagsCommand())
 	return cmd
 }