feat(ledger): import/export (#1586)

components/fctl/cmd/ledger/export.go

@@ -0,0 +1,79 @@
+package ledger
+
+import (
+	"context"
+	"io"
+	"net/http"
+	"os"
+
+	"github.com/formancehq/fctl/cmd/ledger/internal"
+	fctl "github.com/formancehq/fctl/pkg"
+	"github.com/formancehq/formance-sdk-go/v2/pkg/models/operations"
+	"github.com/spf13/cobra"
+)
+
+type ExportStore struct {
+	response *http.Response
+}
+type ExportController struct {
+	store          *ExportStore
+	outputFileFlag string
+}
+
+var _ fctl.Controller[*ExportStore] = (*ExportController)(nil)
+
+func NewDefaultExportStore() *ExportStore {
+	return &ExportStore{}
+}
+
+func NewExportController() *ExportController {
+	return &ExportController{
+		store:          NewDefaultExportStore(),
+		outputFileFlag: "file",
+	}
+}
+
+func NewExportCommand() *cobra.Command {
+	c := NewExportController()
+	return fctl.NewCommand("export",
+		fctl.WithShortDescription("Export a ledger"),
+		fctl.WithStringFlag(c.outputFileFlag, "", "Export to file"),
+		fctl.WithController[*ExportStore](c),
+	)
+}
+
+func (c *ExportController) GetStore() *ExportStore {
+	return c.store
+}
+
+func (c *ExportController) Run(cmd *cobra.Command, args []string) (fctl.Renderable, error) {
+	store := fctl.GetStackStore(cmd.Context())
+
+	ctx := cmd.Context()
+	out := fctl.GetString(cmd, "file")
+	if out != "" {
+		ctx = context.WithValue(ctx, "path", out)
+	}
+
+	ret, err := store.Client().Ledger.V2ExportLogs(ctx, operations.V2ExportLogsRequest{
+		Ledger: fctl.GetString(cmd, internal.LedgerFlag),
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	c.store.response = ret.RawResponse
+
+	return c, nil
+}
+
+func (c *ExportController) Render(cmd *cobra.Command, args []string) error {
+	out := fctl.GetString(cmd, "file")
+	if out == "" {
+		_, err := io.Copy(os.Stdout, c.store.response.Body)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}

components/fctl/cmd/ledger/import.go

@@ -0,0 +1,59 @@
+package ledger
+
+import (
+	"fmt"
+	fctl "github.com/formancehq/fctl/pkg"
+	"github.com/formancehq/formance-sdk-go/v2/pkg/models/operations"
+	"github.com/formancehq/stack/libs/go-libs/pointer"
+	"github.com/pterm/pterm"
+	"github.com/spf13/cobra"
+)
+
+type ImportStore struct {}
+type ImportController struct {
+	store         *ImportStore
+	inputFileFlag string
+}
+
+var _ fctl.Controller[*ImportStore] = (*ImportController)(nil)
+
+func NewDefaultImportStore() *ImportStore {
+	return &ImportStore{}
+}
+
+func NewImportController() *ImportController {
+	return &ImportController{
+		store:         NewDefaultImportStore(),
+		inputFileFlag: "file",
+	}
+}
+
+func NewImportCommand() *cobra.Command {
+	c := NewImportController()
+	return fctl.NewCommand("import <ledger name> <file path>",
+		fctl.WithArgs(cobra.ExactArgs(2)),
+		fctl.WithShortDescription("Import a ledger"),
+		fctl.WithStringFlag(c.inputFileFlag, "", "Import from stdin or file"),
+		fctl.WithController[*ImportStore](c),
+	)
+}
+
+func (c *ImportController) GetStore() *ImportStore {
+	return c.store
+}
+
+func (c *ImportController) Run(cmd *cobra.Command, args []string) (fctl.Renderable, error) {
+	store := fctl.GetStackStore(cmd.Context())
+
+	_, err := store.Client().Ledger.V2ImportLogs(cmd.Context(), operations.V2ImportLogsRequest{
+		Ledger: args[0],
+		RequestBody: pointer.For(fmt.Sprintf("file:%s", args[1])),
+	})
+
+	return c, err
+}
+
+func (c *ImportController) Render(cmd *cobra.Command, args []string) error {
+	pterm.Success.WithWriter(cmd.OutOrStdout()).Printfln("Ledger imported!")
+	return nil
+}

components/fctl/cmd/ledger/root.go

@@ -22,6 +22,8 @@ func NewCommand() *cobra.Command {
 			NewListCommand(),
 			NewSetMetadataCommand(),
 			NewDeleteMetadataCommand(),
+			NewExportCommand(),
+			NewImportCommand(),
 			transactions.NewLedgerTransactionsCommand(),
 			accounts.NewLedgerAccountsCommand(),
 			volumes.NewLedgerVolumesCommand(),

components/fctl/go.sum

@@ -88,7 +88,7 @@ github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHk
 github.com/frankban/quicktest v1.14.6/go.mod h1:4ptaffx2x8+WTWXmUCuVU6aPUX1/Mz7zb5vbUoiM6w0=
 github.com/fsnotify/fsnotify v1.7.0 h1:8JEhPFa5W2WU7YfeZzPNqzMP6Lwt7L2715Ggo0nosvA=
 github.com/fsnotify/fsnotify v1.7.0/go.mod h1:40Bi/Hjc2AVfZrqy+aj+yEI+/bRxZnMJyTJwOpGvigM=
-github.com/go-chi/chi v4.1.2+incompatible h1:fGFk2Gmi/YKXk0OmGfBh0WgmN3XB8lVnEyNz34tQRec=
+github.com/go-chi/chi v4.0.2+incompatible h1:maB6vn6FqCxrpz4FqWdh4+lwpyZIQS7YEAUcHlgXVRs=
 github.com/go-chi/chi/v5 v5.0.12 h1:9euLV5sTrTNTRUU9POmDUvfxyj6LAABLUcEWO+JJb4s=
 github.com/go-chi/chi/v5 v5.0.12/go.mod h1:DslCQbL2OYiznFReuXYUmQ2hGd1aDpCnlMNITLSKoi8=
 github.com/go-logr/logr v1.4.1 h1:pKouT5E8xu9zeFC39JXRDukb6JFQPXM5p5I91188VAQ=

components/ledger/internal/api/backend/backend.go

@@ -34,6 +34,8 @@ type Ledger interface {
 	RevertTransaction(ctx context.Context, parameters command.Parameters, id *big.Int, force, atEffectiveDate bool) (*ledger.Transaction, error)
 	SaveMeta(ctx context.Context, parameters command.Parameters, targetType string, targetID any, m metadata.Metadata) error
 	DeleteMetadata(ctx context.Context, parameters command.Parameters, targetType string, targetID any, key string) error
+	Import(ctx context.Context, stream chan *ledger.ChainedLog) error
+	Export(ctx context.Context, w engine.ExportWriter) error
 
 	IsDatabaseUpToDate(ctx context.Context) (bool, error)
 

components/ledger/internal/api/v2/controller_export_logs.go

@@ -0,0 +1,23 @@
+package v2
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+
+	ledger "github.com/formancehq/ledger/internal"
+	"github.com/formancehq/ledger/internal/api/backend"
+	"github.com/formancehq/ledger/internal/engine"
+	"github.com/formancehq/stack/libs/go-libs/api"
+)
+
+func exportLogs(w http.ResponseWriter, r *http.Request) {
+	enc := json.NewEncoder(w)
+	w.Header().Set("Content-Type", "application/octet-stream")
+	if err := backend.LedgerFromContext(r.Context()).Export(r.Context(), engine.ExportWriterFn(func(ctx context.Context, log *ledger.ChainedLog) error {
+		return enc.Encode(log)
+	})); err != nil {
+		api.InternalServerError(w, r, err)
+		return
+	}
+}

components/ledger/internal/api/v2/controller_import_logs.go

@@ -0,0 +1,65 @@
+package v2
+
+import (
+	"encoding/json"
+	"io"
+	"net/http"
+
+	"github.com/formancehq/ledger/internal/engine"
+
+	ledger "github.com/formancehq/ledger/internal"
+	"github.com/formancehq/ledger/internal/api/backend"
+	"github.com/formancehq/stack/libs/go-libs/api"
+	"github.com/pkg/errors"
+)
+
+func importLogs(w http.ResponseWriter, r *http.Request) {
+
+	stream := make(chan *ledger.ChainedLog)
+	errChan := make(chan error, 1)
+	go func() {
+		errChan <- backend.LedgerFromContext(r.Context()).Import(r.Context(), stream)
+	}()
+	dec := json.NewDecoder(r.Body)
+	handleError := func(err error) {
+		switch {
+		case errors.Is(err, engine.ImportError{}):
+			api.WriteErrorResponse(w, http.StatusBadRequest, "IMPORT", err)
+		default:
+			api.InternalServerError(w, r, err)
+		}
+	}
+	for {
+		l := &ledger.ChainedLog{}
+		if err := dec.Decode(l); err != nil {
+			if errors.Is(err, io.EOF) {
+				close(stream)
+				break
+			} else {
+				api.InternalServerError(w, r, err)
+				return
+			}
+		}
+		select {
+		case stream <- l:
+		case <-r.Context().Done():
+			api.InternalServerError(w, r, r.Context().Err())
+			return
+		case err := <-errChan:
+			handleError(err)
+			return
+		}
+	}
+	select {
+	case err := <-errChan:
+		if err != nil {
+			handleError(err)
+			return
+		}
+	case <-r.Context().Done():
+		api.InternalServerError(w, r, r.Context().Err())
+		return
+	}
+
+	api.NoContent(w)
+}

components/ledger/internal/api/v2/controllers_get_ledger.go

@@ -15,7 +15,7 @@ import (
 
 func getLedger(b backend.Backend) http.HandlerFunc {
 	return func(w http.ResponseWriter, r *http.Request) {
-		configuration := driver.LedgerConfiguration{}
+		configuration := driver.LedgerState{}
 
 		data, err := io.ReadAll(r.Body)
 		if err != nil && !errors.Is(err, io.EOF) {

components/ledger/internal/api/v2/controllers_get_logs.go

@@ -0,0 +1,52 @@
+package v2
+
+import (
+	"fmt"
+	"net/http"
+
+	"github.com/formancehq/ledger/internal/api/backend"
+	"github.com/formancehq/ledger/internal/storage/ledgerstore"
+	sharedapi "github.com/formancehq/stack/libs/go-libs/api"
+	"github.com/formancehq/stack/libs/go-libs/bun/bunpaginate"
+)
+
+func getLogs(w http.ResponseWriter, r *http.Request) {
+	l := backend.LedgerFromContext(r.Context())
+
+	query := ledgerstore.GetLogsQuery{}
+
+	if r.URL.Query().Get(QueryKeyCursor) != "" {
+		err := bunpaginate.UnmarshalCursor(r.URL.Query().Get(QueryKeyCursor), &query)
+		if err != nil {
+			sharedapi.BadRequest(w, ErrValidation, fmt.Errorf("invalid '%s' query param", QueryKeyCursor))
+			return
+		}
+	} else {
+		var err error
+
+		pageSize, err := bunpaginate.GetPageSize(r)
+		if err != nil {
+			sharedapi.BadRequest(w, ErrValidation, err)
+			return
+		}
+
+		qb, err := getQueryBuilder(r)
+		if err != nil {
+			sharedapi.BadRequest(w, ErrValidation, err)
+			return
+		}
+
+		query = ledgerstore.NewGetLogsQuery(ledgerstore.PaginatedQueryOptions[any]{
+			QueryBuilder: qb,
+			PageSize:     pageSize,
+		})
+	}
+
+	cursor, err := l.GetLogs(r.Context(), query)
+	if err != nil {
+		sharedapi.InternalServerError(w, r, err)
+		return
+	}
+
+	sharedapi.RenderCursor(w, *cursor)
+}