feat: make transactions pcv computed on the fly

formancehq · Oct 23, 2024 · c5b0510 · c5b0510
1 parent 53d5611
commit c5b0510
Show file tree

Hide file tree

Showing 15 changed files with 238 additions and 165 deletions.
diff --git a/internal/api/v1/controllers_transactions_create.go b/internal/api/v1/controllers_transactions_create.go
@@ -119,6 +119,7 @@ func createTransaction(w http.ResponseWriter, r *http.Request) {
 		Metadata:  payload.Metadata,
 	}
 
+	// todo: handle missing error cases
 	res, err := l.CreateTransaction(r.Context(), getCommandParameters(r), runScript)
 	if err != nil {
 		switch {

diff --git a/internal/api/v1/controllers_transactions_create_test.go b/internal/api/v1/controllers_transactions_create_test.go
@@ -2,6 +2,7 @@ package v1
 
 import (
 	"encoding/json"
+	"github.com/formancehq/go-libs/time"
 	"math/big"
 	"net/http"
 	"net/http/httptest"
@@ -221,6 +222,7 @@ func TestTransactionsCreate(t *testing.T) {
 
 			systemController, ledgerController := newTestingSystemController(t, true)
 			if testCase.expectedStatusCode < 300 && testCase.expectedStatusCode >= 200 {
+				testCase.expectedRunScript.Timestamp = time.Time{}
 				ledgerController.EXPECT().
 					CreateTransaction(gomock.Any(), ledgercontroller.Parameters{
 						DryRun: tc.expectedPreview,

diff --git a/internal/api/v2/controllers_transactions_create_test.go b/internal/api/v2/controllers_transactions_create_test.go
@@ -1,6 +1,7 @@
 package v2
 
 import (
+	"github.com/formancehq/go-libs/time"
 	"math/big"
 	"net/http"
 	"net/http/httptest"
@@ -401,6 +402,7 @@ func TestTransactionCreate(t *testing.T) {
 
 			systemController, ledgerController := newTestingSystemController(t, true)
 			if testCase.expectControllerCall {
+				testCase.expectedRunScript.Timestamp = time.Time{}
 				expect := ledgerController.EXPECT().
 					CreateTransaction(gomock.Any(), ledgercontroller.Parameters{
 						DryRun: tc.expectedDryRun,

diff --git a/internal/storage/bucket/migrations/0-init-schema.sql b/internal/storage/bucket/migrations/0-init-schema.sql
@@ -551,11 +551,13 @@ as
 $$
 begin
     insert into "{{.Bucket}}".accounts_metadata (ledger, accounts_seq, revision, date, metadata)
-    values (new.ledger, new.seq, (select revision + 1
-                                  from "{{.Bucket}}".accounts_metadata
-                                  where accounts_metadata.accounts_seq = new.seq
-                                  order by revision desc
-                                  limit 1), new.updated_at, new.metadata);
+    values (new.ledger, new.seq, (
+        select revision + 1
+		from "{{.Bucket}}".accounts_metadata
+		where accounts_metadata.accounts_seq = new.seq
+		order by revision desc
+		limit 1
+	), new.updated_at, new.metadata);
 
     return new;
 end;

diff --git a/internal/storage/bucket/migrations/11-stateless.sql b/internal/storage/bucket/migrations/11-stateless.sql
@@ -21,6 +21,9 @@ rename attribute inputs to input;
 alter type "{{.Bucket}}".volumes
 rename attribute outputs to output;
 
+alter table "{{.Bucket}}".transactions
+add column post_commit_volumes jsonb not null ;
+
 alter table "{{.Bucket}}".moves
 alter column post_commit_volumes
 drop not null,
@@ -119,7 +122,6 @@ create function "{{.Bucket}}".set_volumes()
 as
 $$
 begin
-	--todo: use balances table directly...
     new.post_commit_volumes = coalesce((
         select (
             (post_commit_volumes).input + case when new.is_source then 0 else new.amount end,

diff --git a/internal/storage/ledger/accounts_test.go b/internal/storage/ledger/accounts_test.go
@@ -290,6 +290,7 @@ func TestGetAccount(t *testing.T) {
 
 	t.Run("find account in past", func(t *testing.T) {
 		t.Parallel()
+
 		account, err := store.GetAccount(ctx, ledgercontroller.NewGetAccountQuery("multi").WithPIT(now.Add(-30*time.Second)))
 		require.NoError(t, err)
 		require.Equal(t, ledger.ExpandedAccount{

diff --git a/internal/storage/ledger/balances_test.go b/internal/storage/ledger/balances_test.go
@@ -281,7 +281,7 @@ func TestUpdateBalances(t *testing.T) {
 		AccountsSeq: world.Seq,
 	})
 	require.NoError(t, err)
-	require.Equal(t, map[string]map[string]ledger.Volumes{
+	require.Equal(t, ledger.PostCommitVolumes{
 		"world": {
 			"USD/2": ledger.NewVolumesInt64(0, 100),
 		},
@@ -295,7 +295,7 @@ func TestUpdateBalances(t *testing.T) {
 		Output:  big.NewInt(0),
 	})
 	require.NoError(t, err)
-	require.Equal(t, map[string]map[string]ledger.Volumes{
+	require.Equal(t, ledger.PostCommitVolumes{
 		"world": {
 			"USD/2": ledger.NewVolumesInt64(50, 100),
 		},
@@ -310,7 +310,7 @@ func TestUpdateBalances(t *testing.T) {
 		AccountsSeq: world.Seq,
 	})
 	require.NoError(t, err)
-	require.Equal(t, map[string]map[string]ledger.Volumes{
+	require.Equal(t, ledger.PostCommitVolumes{
 		"world": {
 			"USD/2": ledger.NewVolumesInt64(100, 150),
 		},

diff --git a/internal/storage/ledger/moves.go b/internal/storage/ledger/moves.go
@@ -126,29 +126,13 @@ func (m Moves) volumeUpdates() []AccountsVolumes {
 	return ret
 }
 
-func (m Moves) ComputePostCommitVolumes() TransactionsPostCommitVolumes {
-	ret := TransactionsPostCommitVolumes{}
+func (m Moves) ComputePostCommitEffectiveVolumes() ledger.PostCommitVolumes {
+	ret := ledger.PostCommitVolumes{}
 	for _, move := range m {
-		ret = append(ret, TransactionPostCommitVolume{
-			AggregatedAccountVolume: AggregatedAccountVolume{
-				Volumes: *move.PostCommitVolumes,
-				Asset:   move.Asset,
+		ret = ret.Merge(ledger.PostCommitVolumes{
+			move.Account: ledger.VolumesByAssets{
+				move.Asset: *move.PostCommitEffectiveVolumes,
 			},
-			Account: move.Account,
-		})
-	}
-	return ret
-}
-
-func (m Moves) ComputePostCommitEffectiveVolumes() TransactionsPostCommitVolumes {
-	ret := TransactionsPostCommitVolumes{}
-	for _, move := range m {
-		ret = append(ret, TransactionPostCommitVolume{
-			AggregatedAccountVolume: AggregatedAccountVolume{
-				Volumes: *move.PostCommitEffectiveVolumes,
-				Asset:   move.Asset,
-			},
-			Account: move.Account,
 		})
 	}
 	return ret