feat: add capability to create transaction at controller level

formancehq · Nov 22, 2024 · d0f36bb · d0f36bb
1 parent 382c5e9
commit d0f36bb
Show file tree

Hide file tree

Showing 32 changed files with 931 additions and 435 deletions.
diff --git a/Earthfile b/Earthfile
@@ -35,6 +35,7 @@ generate:
     RUN go install github.com/princjef/gomarkdoc/cmd/gomarkdoc@latest
     COPY (+tidy/*) /src/
     COPY --dir (+sources/src/*) /src/
+
     WORKDIR /src
     RUN go generate ./...
     SAVE ARTIFACT internal AS LOCAL internal

diff --git a/docs/api/README.md b/docs/api/README.md
@@ -498,6 +498,8 @@ Accept: application/json
 |Name|In|Type|Required|Description|
 |---|---|---|---|---|
 |ledger|path|string|true|Name of the ledger.|
+|continueOnFailure|query|boolean|false|Continue on failure|
+|atomic|query|boolean|false|Make bulk atomic|
 |body|body|[V2Bulk](#schemav2bulk)|false|none|
 
 > Example responses

diff --git a/internal/api/common/mocks_ledger_controller_test.go b/internal/api/common/mocks_ledger_controller_test.go
diff --git a/internal/api/v1/mocks_ledger_controller_test.go b/internal/api/v1/mocks_ledger_controller_test.go
diff --git a/internal/api/v2/controllers_bulk.go b/internal/api/v2/controllers_bulk.go
@@ -31,7 +31,13 @@ func bulkHandler(bulkMaxSize int) http.HandlerFunc {
 
 		w.Header().Set("Content-Type", "application/json")
 
-		ret, errorsInBulk, err := ProcessBulk(r.Context(), common.LedgerFromContext(r.Context()), b, api.QueryParamBool(r, "continueOnFailure"))
+		ret, errorsInBulk, err := ProcessBulk(
+			r.Context(),
+			common.LedgerFromContext(r.Context()),
+			b,
+			api.QueryParamBool(r, "continueOnFailure"),
+			api.QueryParamBool(r, "atomic"),
+		)
 		if err != nil {
 			api.InternalServerError(w, r, err)
 			return
@@ -122,12 +128,7 @@ func (req *TransactionRequest) ToRunScript(allowUnboundedOverdrafts bool) (*ledg
 	}, nil
 }
 
-func ProcessBulk(
-	ctx context.Context,
-	l ledgercontroller.Controller,
-	bulk Bulk,
-	continueOnFailure bool,
-) ([]Result, bool, error) {
+func ProcessBulk(ctx context.Context, l ledgercontroller.Controller, bulk Bulk, continueOnFailure bool, atomic bool) ([]Result, bool, error) {
 
 	for i, element := range bulk {
 		switch element.Action {
@@ -166,6 +167,15 @@ func ProcessBulk(
 		errorsInBulk = true
 	}
 
+	if atomic {
+		if err := l.BeginTX(ctx, nil); err != nil {
+			return nil, errorsInBulk, fmt.Errorf("error starting transaction: %s", err)
+		}
+		defer func() {
+			_ = l.Rollback(ctx)
+		}()
+	}
+
 	for i, element := range bulk {
 		switch element.Action {
 		case ActionCreateTransaction:
@@ -370,5 +380,12 @@ func ProcessBulk(
 			}
 		}
 	}
+
+	if atomic {
+		if err := l.Commit(ctx); err != nil {
+			return nil, errorsInBulk, fmt.Errorf("error committing transaction: %s", err)
+		}
+	}
+
 	return ret, errorsInBulk, nil
 }
diff --git a/internal/api/v2/controllers_bulk_test.go b/internal/api/v2/controllers_bulk_test.go
@@ -349,9 +349,76 @@ func TestBulk(t *testing.T) {
 			}},
 			expectError: true,
 		},
+		{
+			name: "with atomic",
+			body: `[
+				{
+					"action": "ADD_METADATA",
+					"data": {
+						"targetId": "world",
+						"targetType": "ACCOUNT",
+						"metadata": {
+							"foo": "bar"
+						}			
+					}
+				},
+				{
+					"action": "ADD_METADATA",
+					"data": {
+						"targetId": "world",
+						"targetType": "ACCOUNT",
+						"metadata": {
+							"foo2": "bar2"
+						}			
+					}
+				}
+			]`,
+			queryParams: map[string][]string{
+				"atomic": {"true"},
+			},
+			expectations: func(mockLedger *LedgerController) {
+				mockLedger.EXPECT().
+					BeginTX(gomock.Any(), nil).
+					Return(nil)
+
+				mockLedger.EXPECT().
+					SaveAccountMetadata(gomock.Any(), ledgercontroller.Parameters[ledgercontroller.SaveAccountMetadata]{
+						Input: ledgercontroller.SaveAccountMetadata{
+							Address: "world",
+							Metadata: metadata.Metadata{
+								"foo": "bar",
+							},
+						},
+					}).
+					Return(&ledger.Log{}, nil)
+
+				mockLedger.EXPECT().
+					SaveAccountMetadata(gomock.Any(), ledgercontroller.Parameters[ledgercontroller.SaveAccountMetadata]{
+						Input: ledgercontroller.SaveAccountMetadata{
+							Address: "world",
+							Metadata: metadata.Metadata{
+								"foo2": "bar2",
+							},
+						},
+					}).
+					Return(&ledger.Log{}, nil)
+
+				mockLedger.EXPECT().
+					Commit(gomock.Any()).
+					Return(nil)
+
+				mockLedger.EXPECT().
+					Rollback(gomock.Any()).
+					Return(nil)
+			},
+			expectResults: []Result{{
+				ResponseType: ActionAddMetadata,
+			}, {
+				ResponseType: ActionAddMetadata,
+			}},
+		},
 	}
 	for _, testCase := range testCases {
-		testCase := testCase
 		t.Run(testCase.name, func(t *testing.T) {
 
 			systemController, ledgerController := newTestingSystemController(t, true)

diff --git a/internal/api/v2/mocks_ledger_controller_test.go b/internal/api/v2/mocks_ledger_controller_test.go
diff --git a/internal/controller/ledger/controller.go b/internal/controller/ledger/controller.go
@@ -2,6 +2,7 @@ package ledger
 
 import (
 	"context"
+	"database/sql"
 	"github.com/formancehq/go-libs/v2/metadata"
 	"github.com/formancehq/ledger/internal/machine/vm"
 
@@ -13,6 +14,10 @@ import (
 //go:generate mockgen -write_source_comment=false -write_package_comment=false -source controller.go -destination controller_generated_test.go -package ledger . Controller
 
 type Controller interface {
+	BeginTX(ctx context.Context, options *sql.TxOptions) error
+	Commit(ctx context.Context) error
+	Rollback(ctx context.Context) error
+
 	// IsDatabaseUpToDate check if the ledger store is up to date, including the bucket and the ledger specifics
 	// It returns true if up to date
 	IsDatabaseUpToDate(ctx context.Context) (bool, error)