feat: add accounts balances (#433)

components/operator/internal/controllers/stack/testdata/monopod-latest/results/migrations-stack.formance.com-v1beta3/payments-v0.9.0-pre-upgrade.yaml

@@ -0,0 +1,18 @@
+apiVersion: stack.formance.com/v1beta3
+kind: Migration
+metadata:
+    annotations:
+        reloader.stakater.com/auto: "true"
+    generation: 1
+    labels:
+        stack: "true"
+    name: payments-v0.9.0-pre-upgrade
+    namespace: monopod-latest
+spec:
+    configuration: monopod-latest
+    module: payments
+    postUpgrade: false
+    targetedVersion: v0.9.0
+    version: monopod-latest
+status:
+    terminated: true

components/operator/internal/controllers/stack/testdata/monopod-ledgerv1/results/migrations-stack.formance.com-v1beta3/payments-v0.9.0-pre-upgrade.yaml

@@ -0,0 +1,18 @@
+apiVersion: stack.formance.com/v1beta3
+kind: Migration
+metadata:
+    annotations:
+        reloader.stakater.com/auto: "true"
+    generation: 1
+    labels:
+        stack: "true"
+    name: payments-v0.9.0-pre-upgrade
+    namespace: monopod-ledgerv1
+spec:
+    configuration: monopod-ledgerv1
+    module: payments
+    postUpgrade: false
+    targetedVersion: v0.9.0
+    version: monopod-ledgerv1
+status:
+    terminated: true

components/operator/internal/controllers/stack/testdata/monopod-search-before-v0.7.0/results/migrations-stack.formance.com-v1beta3/payments-v0.9.0-pre-upgrade.yaml

@@ -0,0 +1,18 @@
+apiVersion: stack.formance.com/v1beta3
+kind: Migration
+metadata:
+    annotations:
+        reloader.stakater.com/auto: "true"
+    generation: 1
+    labels:
+        stack: "true"
+    name: payments-v0.9.0-pre-upgrade
+    namespace: monopod-search-before-v0-7-0
+spec:
+    configuration: monopod-search-before-v0-7-0
+    module: payments
+    postUpgrade: false
+    targetedVersion: v0.9.0
+    version: monopod-search-before-v0-7-0
+status:
+    terminated: true

components/operator/internal/controllers/stack/testdata/multipod-latest/results/migrations-stack.formance.com-v1beta3/payments-v0.9.0-pre-upgrade.yaml

@@ -0,0 +1,18 @@
+apiVersion: stack.formance.com/v1beta3
+kind: Migration
+metadata:
+    annotations:
+        reloader.stakater.com/auto: "true"
+    generation: 1
+    labels:
+        stack: "true"
+    name: payments-v0.9.0-pre-upgrade
+    namespace: multipod-latest
+spec:
+    configuration: multipod-latest
+    module: payments
+    postUpgrade: false
+    targetedVersion: v0.9.0
+    version: multipod-latest
+status:
+    terminated: true

components/operator/internal/handlers/handler_payments.go

@@ -116,6 +116,15 @@ func init() {
 					return paymentsServices(ctx, env)
 				},
 			},
+			"v0.9.0": {
+				PreUpgrade: func(ctx modules.Context) error {
+					// Add payment accounts
+					return paymentsPreUpgradeMigration(ctx)
+				},
+				Services: func(ctx modules.ModuleContext) modules.Services {
+					return paymentsServices(ctx, env)
+				},
+			},
 		},
 	})
 }

components/payments/internal/app/api/accounts.go

@@ -14,7 +14,7 @@ import (
 	"github.com/pkg/errors"
 )
 
-type listAccountsRepository interface {
+type accountsRepository interface {
 	ListAccounts(ctx context.Context, pagination storage.Paginator) ([]*models.Account, storage.PaginationDetails, error)
 }
 
@@ -28,7 +28,7 @@ type accountResponse struct {
 	Raw             interface{} `json:"raw"`
 }
 
-func listAccountsHandler(repo listAccountsRepository) http.HandlerFunc {
+func listAccountsHandler(repo accountsRepository) http.HandlerFunc {
 	return func(w http.ResponseWriter, r *http.Request) {
 		w.Header().Set("Content-Type", "application/json")
 

components/payments/internal/app/api/balances.go

@@ -0,0 +1,168 @@
+package api
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/formancehq/payments/internal/app/models"
+	"github.com/formancehq/payments/internal/app/storage"
+	"github.com/formancehq/stack/libs/go-libs/api"
+	"github.com/gorilla/mux"
+	"github.com/pkg/errors"
+)
+
+type balancesRepository interface {
+	ListBalances(ctx context.Context, query storage.BalanceQuery) ([]*models.Balance, storage.PaginationDetails, error)
+}
+
+type balancesResponse struct {
+	AccountID     string    `json:"accountId"`
+	CreatedAt     time.Time `json:"createdAt"`
+	LastUpdatedAt time.Time `json:"lastUpdatedAt"`
+	Currency      string    `json:"currency"`
+	Balance       int64     `json:"balance"`
+}
+
+func listBalancesForAccount(repo balancesRepository) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+
+		var sorter storage.Sorter
+
+		if sortParams := r.URL.Query()["sort"]; sortParams != nil {
+			for _, s := range sortParams {
+				parts := strings.SplitN(s, ":", 2)
+
+				var order storage.SortOrder
+
+				if len(parts) > 1 {
+					//nolint:goconst // allow duplicate string
+					switch parts[1] {
+					case "asc", "ASC":
+						order = storage.SortOrderAsc
+					case "dsc", "desc", "DSC", "DESC":
+						order = storage.SortOrderDesc
+					default:
+						handleValidationError(w, r, errors.New("sort order not well specified, got "+parts[1]))
+
+						return
+					}
+				}
+
+				column := parts[0]
+
+				sorter.Add(column, order)
+			}
+		}
+
+		pageSize, err := pageSizeQueryParam(r)
+		if err != nil {
+			handleValidationError(w, r, err)
+
+			return
+		}
+
+		pagination, err := storage.Paginate(pageSize, r.URL.Query().Get("cursor"), sorter)
+		if err != nil {
+			handleValidationError(w, r, err)
+
+			return
+		}
+
+		accountID, err := models.AccountIDFromString(mux.Vars(r)["accountID"])
+		if err != nil {
+			handleValidationError(w, r, err)
+
+			return
+		}
+
+		balanceQuery := storage.NewBalanceQuery(pagination).
+			WithAccountID(accountID).
+			WithCurrency(r.URL.Query().Get("currency"))
+
+		var startTimeParsed, endTimeParsed time.Time
+		from, to := r.URL.Query().Get("from"), r.URL.Query().Get("to")
+		if from != "" {
+			startTimeParsed, err = time.Parse(time.RFC3339Nano, from)
+			if err != nil {
+				handleValidationError(w, r, err)
+
+				return
+			}
+		}
+		if to != "" {
+			endTimeParsed, err = time.Parse(time.RFC3339Nano, to)
+			if err != nil {
+				handleValidationError(w, r, err)
+
+				return
+			}
+		}
+		if r.URL.Query().Get("limit") != "" {
+			limit, err := strconv.ParseInt(r.URL.Query().Get("limit"), 10, 64)
+			if err != nil {
+				handleValidationError(w, r, err)
+
+				return
+			}
+
+			if limit > 0 {
+				balanceQuery = balanceQuery.WithLimit(int(limit))
+			}
+		}
+
+		switch {
+		case startTimeParsed.IsZero() && endTimeParsed.IsZero():
+			balanceQuery = balanceQuery.
+				WithTo(time.Now())
+		case !startTimeParsed.IsZero() && endTimeParsed.IsZero():
+			balanceQuery = balanceQuery.
+				WithFrom(startTimeParsed).
+				WithTo(time.Now())
+		case startTimeParsed.IsZero() && !endTimeParsed.IsZero():
+			balanceQuery = balanceQuery.
+				WithTo(endTimeParsed)
+		default:
+			balanceQuery = balanceQuery.
+				WithFrom(startTimeParsed).
+				WithTo(endTimeParsed)
+		}
+
+		ret, paginationDetails, err := repo.ListBalances(r.Context(), balanceQuery)
+		if err != nil {
+			handleServerError(w, r, err)
+
+			return
+		}
+
+		data := make([]*balancesResponse, len(ret))
+		for i := range ret {
+			data[i] = &balancesResponse{
+				AccountID:     ret[i].AccountID.String(),
+				CreatedAt:     ret[i].CreatedAt,
+				Currency:      ret[i].Currency,
+				Balance:       ret[i].Balance,
+				LastUpdatedAt: ret[i].LastUpdatedAt,
+			}
+		}
+
+		err = json.NewEncoder(w).Encode(api.BaseResponse[*balancesResponse]{
+			Cursor: &api.Cursor[*balancesResponse]{
+				PageSize: paginationDetails.PageSize,
+				HasMore:  paginationDetails.HasMore,
+				Previous: paginationDetails.PreviousPage,
+				Next:     paginationDetails.NextPage,
+				Data:     data,
+			},
+		})
+		if err != nil {
+			handleServerError(w, r, err)
+
+			return
+		}
+	}
+}

components/payments/internal/app/api/router.go

@@ -46,6 +46,7 @@ func httpRouter(logger logging.Logger, store *storage.Storage, serviceInfo api.S
 	authGroup.Path("/payments/{paymentID}/metadata").Methods(http.MethodPatch).Handler(updateMetadataHandler(store))
 
 	authGroup.Path("/accounts").Methods(http.MethodGet).Handler(listAccountsHandler(store))
+	authGroup.Path("/accounts/{accountID}/balances").Methods(http.MethodGet).Handler(listBalancesForAccount(store))
 
 	authGroup.HandleFunc("/connectors", readConnectorsHandler(store))
 

components/payments/internal/app/connectors/bankingcircle/task_fetch_accounts.go

@@ -62,8 +62,10 @@ func ingestAccountsBatch(
 	ingester ingestion.Ingester,
 	accounts []*client.Account,
 ) error {
-	batch := ingestion.AccountBatch{}
+	accountsBatch := ingestion.AccountBatch{}
+	balanceBatch := ingestion.BalanceBatch{}
 
+	now := time.Now()
 	for _, account := range accounts {
 		raw, err := json.Marshal(account)
 		if err != nil {
@@ -75,26 +77,42 @@ func ingestAccountsBatch(
 			return fmt.Errorf("failed to parse opening date: %w", err)
 		}
 
-		batchElement := ingestion.AccountBatchElement{
-			Account: &models.Account{
-				ID: models.AccountID{
+		accountsBatch = append(accountsBatch, &models.Account{
+			ID: models.AccountID{
+				Reference: account.AccountID,
+				Provider:  models.ConnectorProviderBankingCircle,
+			},
+			CreatedAt:       openingDate,
+			Reference:       account.AccountID,
+			Provider:        models.ConnectorProviderBankingCircle,
+			DefaultCurrency: account.Currency,
+			AccountName:     account.AccountDescription,
+			Type:            models.AccountTypeInternal,
+			RawData:         raw,
+		})
+
+		for _, balance := range account.Balances {
+			balanceBatch = append(balanceBatch, &models.Balance{
+				AccountID: models.AccountID{
 					Reference: account.AccountID,
 					Provider:  models.ConnectorProviderBankingCircle,
 				},
-				CreatedAt:       openingDate,
-				Reference:       account.AccountID,
-				Provider:        models.ConnectorProviderBankingCircle,
-				DefaultCurrency: account.Currency,
-				AccountName:     account.AccountDescription,
-				Type:            models.AccountTypeInternal,
-				RawData:         raw,
-			},
+				// Note(polo): same thing as payments
+				// TODO(polo): do a complete pass on all connectors to
+				// normalize the currencies
+				Currency:      balance.Currency + "/2",
+				Balance:       int64(balance.IntraDayAmount * 100),
+				CreatedAt:     now,
+				LastUpdatedAt: now,
+			})
 		}
+	}
 
-		batch = append(batch, batchElement)
+	if err := ingester.IngestAccounts(ctx, accountsBatch); err != nil {
+		return err
 	}
 
-	if err := ingester.IngestAccounts(ctx, batch); err != nil {
+	if err := ingester.IngestBalances(ctx, balanceBatch, false); err != nil {
 		return err
 	}