Add pre transactions balance to the log

pkg/api/controllers/account_controller_test.go

@@ -178,10 +178,9 @@ func TestGetAccount(t *testing.T) {
 					Balances: map[string]int64{
 						"USD": 100,
 					},
-					Volumes: map[string]map[string]int64{
+					Volumes: core.Volumes{
 						"USD": {
-							"input":  100,
-							"output": 0,
+							Input: 100,
 						},
 					},
 					Metadata: core.Metadata{

pkg/api/controllers/swagger.yaml

@@ -781,6 +781,10 @@ components:
           format: date-time
         txid:
           type: integer
+        preCommitVolumes:
+          $ref: '#/components/schemas/AggregatedVolumes'
+        postCommitVolumes:
+          $ref: '#/components/schemas/AggregatedVolumes'
       required:
         - postings
         - timestamp
@@ -864,6 +868,50 @@ components:
       type: object
       required:
         - data
+    Volume:
+      type: object
+      properties:
+        input:
+          type: number
+        output:
+          type: number
+        balance:
+          type: number
+      required:
+      - input
+      - output
+      example:
+        input: 100
+        output: 20
+        balance: 80
+    Volumes:
+      type: object
+      additionalProperties:
+        $ref: '#/components/schemas/Volume'
+      example:
+        USD:
+          input: 100
+          output: 10
+          balance: 90
+        EUR:
+          input: 100
+          output: 10
+          balance: 90
+    AggregatedVolumes:
+      type: object
+      additionalProperties:
+        $ref: '#/components/schemas/Volumes'
+      example:
+        "orders:1":
+          "USD":
+            input: 100
+            output: 10
+            balance: 90
+        "orders:2":
+          "USD":
+            input: 100
+            output: 10
+            balance: 90
     ErrorCode:
       type: string
       enum:

pkg/api/controllers/transaction_controller_test.go

@@ -191,25 +191,66 @@ func TestGetTransaction(t *testing.T) {
 				})
 				require.Equal(t, http.StatusOK, rsp.Result().StatusCode)
 
-				tx, _ := internal.DecodeSingleResponse[[]core.Transaction](t, rsp.Body)
+				txs, _ := internal.DecodeSingleResponse[[]core.Transaction](t, rsp.Body)
+				tx := txs[0]
+				assert.EqualValues(t, core.AggregatedVolumes{
+					"world": core.Volumes{
+						"USD": {},
+					},
+					"central_bank": core.Volumes{
+						"USD": {},
+					},
+				}, tx.PreCommitVolumes)
+				assert.EqualValues(t, core.AggregatedVolumes{
+					"world": core.Volumes{
+						"USD": {
+							Output: 1000,
+						},
+					},
+					"central_bank": core.Volumes{
+						"USD": {
+							Input: 1000,
+						},
+					},
+				}, tx.PostCommitVolumes)
 
-				rsp = internal.GetTransaction(api, tx[0].ID)
+				rsp = internal.GetTransaction(api, tx.ID)
 				assert.Equal(t, http.StatusOK, rsp.Result().StatusCode)
 
 				ret, _ := internal.DecodeSingleResponse[core.Transaction](t, rsp.Body)
 
-				assert.EqualValues(t, ret.Postings, core.Postings{
+				assert.EqualValues(t, core.Postings{
 					{
 						Source:      "world",
 						Destination: "central_bank",
 						Amount:      1000,
 						Asset:       "USD",
 					},
-				})
+				}, ret.Postings)
 				assert.EqualValues(t, 0, ret.ID)
 				assert.EqualValues(t, core.Metadata{}, ret.Metadata)
 				assert.EqualValues(t, "ref", ret.Reference)
 				assert.NotEmpty(t, ret.Timestamp)
+				assert.EqualValues(t, core.AggregatedVolumes{
+					"world": core.Volumes{
+						"USD": {},
+					},
+					"central_bank": core.Volumes{
+						"USD": {},
+					},
+				}, ret.PreCommitVolumes)
+				assert.EqualValues(t, core.AggregatedVolumes{
+					"world": core.Volumes{
+						"USD": {
+							Output: 1000,
+						},
+					},
+					"central_bank": core.Volumes{
+						"USD": {
+							Input: 1000,
+						},
+					},
+				}, ret.PostCommitVolumes)
 				return nil
 			},
 		})

pkg/bus/message.go

@@ -14,8 +14,11 @@ type baseEvent struct {
 }
 
 type committedTransactions struct {
-	Transactions []core.Transaction     `json:"transactions"`
-	Volumes      core.AggregatedVolumes `json:"volumes"`
+	Transactions []core.Transaction `json:"transactions"`
+	// Deprecated (use postCommitVolumes)
+	Volumes           core.AggregatedVolumes `json:"volumes"`
+	PostCommitVolumes core.AggregatedVolumes `json:"postCommitVolumes"`
+	PreCommitVolumes  core.AggregatedVolumes `json:"preCommitVolumes"`
 }
 
 type savedMetadata struct {

pkg/bus/monitor.go

@@ -35,10 +35,12 @@ func (l *ledgerMonitor) publish(ctx context.Context, ledger string, et string, d
 	}
 }
 
-func (l *ledgerMonitor) CommittedTransactions(ctx context.Context, ledger string, results []core.Transaction, volumes core.AggregatedVolumes) {
+func (l *ledgerMonitor) CommittedTransactions(ctx context.Context, ledger string, result *ledger.CommitResult) {
 	l.publish(ctx, ledger, CommittedTransactions, committedTransactions{
-		Transactions: results,
-		Volumes:      volumes,
+		Transactions:      result.GeneratedTransactions,
+		Volumes:           result.PostCommitVolumes,
+		PostCommitVolumes: result.PostCommitVolumes,
+		PreCommitVolumes:  result.PreCommitVolumes,
 	})
 }
 

pkg/bus/monitor_test.go

@@ -8,6 +8,7 @@ import (
 	"github.com/ThreeDotsLabs/watermill"
 	"github.com/ThreeDotsLabs/watermill/pubsub/gochannel"
 	"github.com/numary/go-libs/sharedpublish"
+	"github.com/numary/ledger/pkg/ledger"
 	"github.com/pborman/uuid"
 	"github.com/stretchr/testify/assert"
 )
@@ -28,7 +29,7 @@ func TestMonitor(t *testing.T) {
 		"*": "testing",
 	})
 	m := NewLedgerMonitor(p)
-	go m.CommittedTransactions(context.Background(), uuid.New(), nil, nil)
+	go m.CommittedTransactions(context.Background(), uuid.New(), &ledger.CommitResult{})
 
 	select {
 	case m := <-messages:

pkg/core/account.go

@@ -4,19 +4,6 @@ const (
 	WORLD = "world"
 )
 
-type Balances map[string]int64
-type Volumes map[string]map[string]int64
-
-func (v Volumes) Balances() Balances {
-	balances := Balances{}
-	for asset, vv := range v {
-		balances[asset] = vv["input"] - vv["output"]
-	}
-	return balances
-}
-
-type AggregatedVolumes map[string]Volumes
-
 type Account struct {
 	Address  string           `json:"address" example:"users:001"`
 	Type     string           `json:"type,omitempty" example:"virtual"`

pkg/core/transaction.go

@@ -32,8 +32,10 @@ func (t *TransactionData) Reverse() TransactionData {
 
 type Transaction struct {
 	TransactionData
-	ID        uint64 `json:"txid"`
-	Timestamp string `json:"timestamp"`
+	ID                uint64            `json:"txid"`
+	Timestamp         string            `json:"timestamp"`
+	PreCommitVolumes  AggregatedVolumes `json:"preCommitVolumes,omitempty"`  // Keep omitempty to keep consistent hash
+	PostCommitVolumes AggregatedVolumes `json:"postCommitVolumes,omitempty"` // Keep omitempty to keep consistent hash
 }
 
 func (t *Transaction) AppendPosting(p Posting) {

pkg/core/volumes.go

@@ -0,0 +1,61 @@
+package core
+
+import (
+	"database/sql/driver"
+	"encoding/json"
+)
+
+type Volume struct {
+	Input  int64 `json:"input"`
+	Output int64 `json:"output"`
+}
+
+func (v Volume) MarshalJSON() ([]byte, error) {
+	type volume Volume
+	return json.Marshal(struct {
+		volume
+		Balance int64 `json:"balance"`
+	}{
+		volume:  volume(v),
+		Balance: v.Input - v.Output,
+	})
+}
+
+func (v Volume) Balance() int64 {
+	return v.Input - v.Output
+}
+
+type Balances map[string]int64
+type Volumes map[string]Volume
+
+func (v Volumes) Balances() Balances {
+	balances := Balances{}
+	for asset, vv := range v {
+		balances[asset] = vv.Input - vv.Output
+	}
+	return balances
+}
+
+type AggregatedVolumes map[string]Volumes
+
+// Scan - Implement the database/sql scanner interface
+func (m *AggregatedVolumes) Scan(value interface{}) error {
+	if value == nil {
+		return nil
+	}
+
+	v, err := driver.String.ConvertValue(value)
+	if err != nil {
+		return err
+	}
+
+	*m = AggregatedVolumes{}
+	switch vv := v.(type) {
+	case []uint8:
+		return json.Unmarshal(vv, m)
+	case string:
+		return json.Unmarshal([]byte(vv), m)
+	default:
+		panic("not handled type")
+	}
+}

pkg/ledger/executor_test.go

@@ -24,7 +24,7 @@ func assertBalance(t *testing.T, l *Ledger, account, asset string, amount int64)
 }
 
 func TestNoScript(t *testing.T) {
-	with(func(l *Ledger) {
+	runOnLedger(func(l *Ledger) {
 		script := core.Script{}
 
 		_, err := l.Execute(context.Background(), script)
@@ -34,7 +34,7 @@ func TestNoScript(t *testing.T) {
 }
 
 func TestCompilationError(t *testing.T) {
-	with(func(l *Ledger) {
+	runOnLedger(func(l *Ledger) {
 		script := core.Script{
 			Plain: "willnotcompile",
 		}
@@ -46,7 +46,7 @@ func TestCompilationError(t *testing.T) {
 }
 
 func TestTransactionInvalidScript(t *testing.T) {
-	with(func(l *Ledger) {
+	runOnLedger(func(l *Ledger) {
 		script := core.Script{
 			Plain: "this is not a valid script",
 		}
@@ -59,7 +59,7 @@ func TestTransactionInvalidScript(t *testing.T) {
 }
 
 func TestTransactionFail(t *testing.T) {
-	with(func(l *Ledger) {
+	runOnLedger(func(l *Ledger) {
 		script := core.Script{
 			Plain: "fail",
 		}
@@ -72,7 +72,7 @@ func TestTransactionFail(t *testing.T) {
 }
 
 func TestSend(t *testing.T) {
-	with(func(l *Ledger) {
+	runOnLedger(func(l *Ledger) {
 		defer func(l *Ledger, ctx context.Context) {
 			require.NoError(t, l.Close(ctx))
 		}(l, context.Background())
@@ -92,7 +92,7 @@ func TestSend(t *testing.T) {
 }
 
 func TestNoVariables(t *testing.T) {
-	with(func(l *Ledger) {
+	runOnLedger(func(l *Ledger) {
 		var script core.Script
 		err := json.Unmarshal(
 			[]byte(`{
@@ -110,7 +110,7 @@ func TestNoVariables(t *testing.T) {
 }
 
 func TestVariables(t *testing.T) {
-	with(func(l *Ledger) {
+	runOnLedger(func(l *Ledger) {
 		defer func(l *Ledger, ctx context.Context) {
 			require.NoError(t, l.Close(ctx))
 		}(l, context.Background())
@@ -141,7 +141,7 @@ func TestVariables(t *testing.T) {
 }
 
 func TestEnoughFunds(t *testing.T) {
-	with(func(l *Ledger) {
+	runOnLedger(func(l *Ledger) {
 		defer func(l *Ledger, ctx context.Context) {
 			require.NoError(t, l.Close(ctx))
 		}(l, context.Background())
@@ -174,7 +174,7 @@ func TestEnoughFunds(t *testing.T) {
 }
 
 func TestNotEnoughFunds(t *testing.T) {
-	with(func(l *Ledger) {
+	runOnLedger(func(l *Ledger) {
 		defer func(l *Ledger, ctx context.Context) {
 			require.NoError(t, l.Close(ctx))
 		}(l, context.Background())
@@ -207,7 +207,7 @@ func TestNotEnoughFunds(t *testing.T) {
 }
 
 func TestMissingMetadata(t *testing.T) {
-	with(func(l *Ledger) {
+	runOnLedger(func(l *Ledger) {
 		defer func(l *Ledger, ctx context.Context) {
 			require.NoError(t, l.Close(ctx))
 		}(l, context.Background())
@@ -237,7 +237,7 @@ func TestMissingMetadata(t *testing.T) {
 }
 
 func TestMetadata(t *testing.T) {
-	with(func(l *Ledger) {
+	runOnLedger(func(l *Ledger) {
 		defer func(l *Ledger, ctx context.Context) {
 			require.NoError(t, l.Close(ctx))
 		}(l, context.Background())
@@ -308,7 +308,7 @@ func TestMetadata(t *testing.T) {
 }
 
 func TestSetTxMeta(t *testing.T) {
-	with(func(l *Ledger) {
+	runOnLedger(func(l *Ledger) {
 		defer func(l *Ledger, ctx context.Context) {
 			require.NoError(t, l.Close(ctx))
 		}(l, context.Background())