feat: add coinbase prime connector

go.mod

@@ -12,6 +12,7 @@ require (
 	github.com/ThreeDotsLabs/watermill v1.4.7
 	github.com/adyen/adyen-go-api-library/v7 v7.3.1
 	github.com/bombsimon/logrusr/v3 v3.1.0
+	github.com/coinbase-samples/prime-sdk-go v0.5.3
 	github.com/emvi/iso-639-1 v1.1.1
 	github.com/formancehq/go-libs/v3 v3.0.2-0.20250814071617-0f5bb98d939b
 	github.com/formancehq/payments/genericclient v0.0.0-00010101000000-000000000000
@@ -95,6 +96,7 @@ require (
 	github.com/buger/jsonparser v1.1.1 // indirect
 	github.com/cenkalti/backoff/v4 v4.3.0 // indirect
 	github.com/cenkalti/backoff/v5 v5.0.2 // indirect
+	github.com/coinbase-samples/core-go v0.2.1 // indirect
 	github.com/containerd/continuity v0.4.3 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/decred/dcrd/dcrec/secp256k1/v4 v4.4.0 // indirect
@@ -138,6 +140,7 @@ require (
 	github.com/gorilla/mux v1.8.1 // indirect
 	github.com/gorilla/schema v1.4.1 // indirect
 	github.com/gorilla/securecookie v1.1.2 // indirect
+	github.com/gorilla/websocket v1.5.3 // indirect
 	github.com/grpc-ecosystem/go-grpc-middleware v1.4.0 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.1 // indirect
 	github.com/hashicorp/errwrap v1.1.0 // indirect
@@ -196,6 +199,7 @@ require (
 	github.com/robfig/cron v1.2.0 // indirect
 	github.com/rs/cors v1.11.1 // indirect
 	github.com/shirou/gopsutil/v4 v4.25.5 // indirect
+	github.com/shopspring/decimal v1.4.0 // indirect
 	github.com/spf13/pflag v1.0.7 // indirect
 	github.com/stretchr/objx v0.5.2 // indirect
 	github.com/tklauser/go-sysconf v0.3.15 // indirect

go.sum

@@ -710,6 +710,10 @@ github.com/cncf/xds/go v0.0.0-20211011173535-cb28da3451f1/go.mod h1:eXthEFrGJvWH
 github.com/cncf/xds/go v0.0.0-20220314180256-7f1daf1720fc/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
 github.com/cncf/xds/go v0.0.0-20230105202645-06c439db220b/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
 github.com/cncf/xds/go v0.0.0-20230310173818-32f1caf87195/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
+github.com/coinbase-samples/core-go v0.2.1 h1:O5V7je5D95C2000GRC0CM8tNFBfRkaITvu56KHeZirc=
+github.com/coinbase-samples/core-go v0.2.1/go.mod h1:Owx2Pv2gQIUODJ5Ck+g3h/MQ8bftv9OuoTVP8VVH8SI=
+github.com/coinbase-samples/prime-sdk-go v0.5.3 h1:CHj902sunlMAYLN5IsleWRNmnkMp4QstHrpwPdo5GhE=
+github.com/coinbase-samples/prime-sdk-go v0.5.3/go.mod h1:orFTxU1U6RTFXDHam3NTDqx8qYbZ+KunDjh3EW6YJeo=
 github.com/containerd/continuity v0.4.3 h1:6HVkalIp+2u1ZLH1J/pYX2oBVXlJZvh1X1A7bEZ9Su8=
 github.com/containerd/continuity v0.4.3/go.mod h1:F6PTNCKepoxEaXLQp3wDAjygEnImnZ/7o4JzpodfroQ=
 github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g=
@@ -982,6 +986,8 @@ github.com/gorilla/securecookie v1.1.1/go.mod h1:ra0sb63/xPlUeL+yeDciTfxMRAA+MP+
 github.com/gorilla/securecookie v1.1.2 h1:YCIWL56dvtr73r6715mJs5ZvhtnY73hBvEF8kXD8ePA=
 github.com/gorilla/securecookie v1.1.2/go.mod h1:NfCASbcHqRSY+3a8tlWJwsQap2VX5pwzwo4h3eOamfo=
 github.com/gorilla/sessions v1.2.1/go.mod h1:dk2InVEVJ0sfLlnXv9EAgkf6ecYs/i80K/zI+bUmuGM=
+github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
+github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/grpc-ecosystem/go-grpc-middleware v1.4.0 h1:UH//fgunKIs4JdUbpDl1VZCDaL56wXCB/5+wF6uHfaI=
 github.com/grpc-ecosystem/go-grpc-middleware v1.4.0/go.mod h1:g5qyo/la0ALbONm6Vbp88Yd8NsDy6rZz+RcrMPxvld8=
 github.com/grpc-ecosystem/grpc-gateway v1.16.0/go.mod h1:BDjrQk3hbvj6Nolgz8mAMFbcEtjT1g+wF4CSlocrBnw=
@@ -1192,6 +1198,8 @@ github.com/ruudk/golang-pdf417 v0.0.0-20181029194003-1af4ab5afa58/go.mod h1:6lfF
 github.com/ruudk/golang-pdf417 v0.0.0-20201230142125-a7e3863a1245/go.mod h1:pQAZKsJ8yyVxGRWYNEm9oFB8ieLgKFnamEyDmSA0BRk=
 github.com/shirou/gopsutil/v4 v4.25.5 h1:rtd9piuSMGeU8g1RMXjZs9y9luK5BwtnG7dZaQUJAsc=
 github.com/shirou/gopsutil/v4 v4.25.5/go.mod h1:PfybzyydfZcN+JMMjkF6Zb8Mq1A/VcogFFg7hj50W9c=
+github.com/shopspring/decimal v1.4.0 h1:bxl37RwXBklmTi0C79JfXCEBD1cqqHt0bbgBAGFp81k=
+github.com/shopspring/decimal v1.4.0/go.mod h1:gawqmDU56v4yIKSwfBSFip1HdCCXN8/+DMd9qYNcwME=
 github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
 github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ=
 github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=

internal/connectors/plugins/public/coinbaseprime/accounts.go

@@ -0,0 +1,63 @@
+package coinbaseprime
+
+import (
+	"context"
+	"encoding/json"
+	"time"
+
+	"github.com/formancehq/payments/internal/models"
+)
+
+type accountsState struct {
+	LastPage int `json:"lastPage"`
+}
+
+func (p *Plugin) fetchNextAccounts(ctx context.Context, req models.FetchNextAccountsRequest) (models.FetchNextAccountsResponse, error) {
+	var oldState accountsState
+	if req.State != nil {
+		if err := json.Unmarshal(req.State, &oldState); err != nil {
+			return models.FetchNextAccountsResponse{}, err
+		}
+	}
+
+	// Fetch a single SDK page based on state
+	page := oldState.LastPage
+	pagedAccounts, err := p.client.GetAccounts(ctx, page, req.PageSize)
+	if err != nil {
+		return models.FetchNextAccountsResponse{}, err
+	}
+
+	accounts := make([]models.PSPAccount, 0, len(pagedAccounts))
+	for _, a := range pagedAccounts {
+		raw, _ := json.Marshal(a)
+		var namePtr *string
+		if a.Name != "" {
+			n := a.Name
+			namePtr = &n
+		}
+		accounts = append(accounts, models.PSPAccount{
+			Reference: a.ID,
+			CreatedAt: time.Now(),
+			Name:      namePtr,
+			Metadata:  a.Metadata,
+			Raw:       raw,
+		})
+	}
+
+	hasMore := len(pagedAccounts) == req.PageSize
+	newState := accountsState{LastPage: page}
+	if hasMore {
+		newState.LastPage = page + 1
+	}
+
+	payload, err := json.Marshal(newState)
+	if err != nil {
+		return models.FetchNextAccountsResponse{}, err
+	}
+
+	return models.FetchNextAccountsResponse{
+		Accounts: accounts,
+		NewState: payload,
+		HasMore:  hasMore,
+	}, nil
+}

internal/connectors/plugins/public/coinbaseprime/balances.go

@@ -0,0 +1,65 @@
+package coinbaseprime
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/formancehq/go-libs/v3/currency"
+	"github.com/formancehq/payments/internal/connectors/plugins/public/coinbaseprime/client"
+	"github.com/formancehq/payments/internal/models"
+)
+
+func (p *Plugin) fetchNextBalances(ctx context.Context, req models.FetchNextBalancesRequest) (models.FetchNextBalancesResponse, error) {
+	var from models.PSPAccount
+	if req.FromPayload == nil {
+		return models.FetchNextBalancesResponse{}, models.ErrMissingFromPayloadInRequest
+	}
+	if err := json.Unmarshal(req.FromPayload, &from); err != nil {
+		return models.FetchNextBalancesResponse{}, err
+	}
+
+	kind := from.Metadata["spec.coinbase.com/type"]
+	var sdkBalances []*client.Balance
+	var err error
+	if kind == "wallet" {
+		portfolioID := from.Metadata["spec.coinbase.com/portfolio_id"]
+		sdkBalances, err = p.client.GetWalletBalance(ctx, portfolioID, from.Reference)
+	} else {
+		sdkBalances, err = p.client.GetAccountBalances(ctx, from.Reference)
+	}
+	if err != nil {
+		return models.FetchNextBalancesResponse{}, err
+	}
+
+	res := make([]models.PSPBalance, 0, len(sdkBalances))
+	for _, b := range sdkBalances {
+		symbol := strings.ToUpper(b.Symbol)
+		precision, ok := supportedCurrenciesWithDecimal[symbol]
+		if !ok {
+			precision = 8
+		}
+		amount, err := currency.GetAmountWithPrecisionFromString(b.Amount, precision)
+		if err != nil {
+			return models.FetchNextBalancesResponse{}, fmt.Errorf("failed to parse balance amount: %w", err)
+		}
+		asset := currency.FormatAsset(supportedCurrenciesWithDecimal, symbol)
+		if asset == "" {
+			asset = fmt.Sprintf("%s/%d", symbol, precision)
+		}
+
+		res = append(res, models.PSPBalance{
+			AccountReference: from.Reference,
+			CreatedAt:        time.Now().UTC(),
+			Amount:           amount,
+			Asset:            asset,
+		})
+	}
+
+	return models.FetchNextBalancesResponse{
+		Balances: res,
+		HasMore:  false,
+	}, nil
+}

internal/connectors/plugins/public/coinbaseprime/capabilities.go

@@ -0,0 +1,10 @@
+package coinbaseprime
+
+import "github.com/formancehq/payments/internal/models"
+
+var capabilities = []models.Capability{
+	models.CAPABILITY_FETCH_ACCOUNTS,
+	models.CAPABILITY_FETCH_BALANCES,
+	models.CAPABILITY_FETCH_PAYMENTS,
+	models.CAPABILITY_CREATE_TRANSFER,
+}

internal/connectors/plugins/public/coinbaseprime/client/_poc.go

@@ -0,0 +1,185 @@
+package main
+
+import (
+	"context"
+	"crypto/tls"
+	"encoding/json"
+	"fmt"
+	"log"
+	"math/big"
+	"net/http"
+	"os"
+	"strings"
+	"time"
+
+	_ "embed"
+
+	"github.com/coinbase-samples/prime-sdk-go/balances"
+	"github.com/coinbase-samples/prime-sdk-go/client"
+	"github.com/coinbase-samples/prime-sdk-go/credentials"
+	"github.com/coinbase-samples/prime-sdk-go/portfolios"
+	"github.com/coinbase-samples/prime-sdk-go/transactions"
+	"github.com/coinbase-samples/prime-sdk-go/wallets"
+)
+
+func LoadCredentials() string {
+	return os.Getenv("COINBASE_CREDENTIALS")
+}
+
+type Export struct {
+	Accounts []Account `json:"accounts"`
+}
+
+type Account struct {
+	Id          string            `json:"id"`
+	AccountName string            `json:"accountName"`
+	CreatedAt   string            `json:"createdAt"`
+	Metadata    map[string]string `json:"metadata"`
+	Balances    []Balance         `json:"balances"`
+}
+
+type Balance struct {
+	Currency string `json:"currency"`
+	Amount   string `json:"amount"`
+}
+
+func main() {
+	_credentials := LoadCredentials()
+	if _credentials == "" {
+		log.Fatalf("unable to load prime credentials")
+	}
+
+	primeCredentials, err := credentials.UnmarshalCredentials([]byte(_credentials))
+	if err != nil {
+		log.Fatalf("unable to load prime credentials: %v", err)
+	}
+
+	fmt.Println(primeCredentials)
+
+	httpClient, err := client.DefaultHttpClient()
+	httpClient.Transport = &http.Transport{
+		TLSClientConfig: &tls.Config{
+			InsecureSkipVerify: true,
+		},
+	}
+	if err != nil {
+		log.Fatalf("unable to load default http client: %v", err)
+	}
+
+	client := client.NewRestClient(primeCredentials, httpClient)
+
+	portfoliosSvc := portfolios.NewPortfoliosService(client)
+	balancesSvc := balances.NewBalancesService(client)
+	transactionsSvc := transactions.NewTransactionsService(client)
+	walletsSvc := wallets.NewWalletsService(client)
+
+	export := Export{
+		Accounts: []Account{},
+	}
+
+	res, err := portfoliosSvc.ListPortfolios(context.Background(), &portfolios.ListPortfoliosRequest{})
+
+	if err != nil {
+		log.Fatalf("unable to list portfolios: %v", err)
+	}
+
+	for _, p := range res.Portfolios {
+		account := Account{
+			Id:          p.Id,
+			AccountName: p.Name,
+			CreatedAt:   time.Now().Format(time.RFC3339),
+			Metadata: map[string]string{
+				"spec.coinbase.com/type":         "portfolio",
+				"spec.coinbase.com/portfolio_id": p.Id,
+			},
+		}
+
+		fmt.Printf("|-- %s: %s\n", p.Id, p.Name)
+
+		{
+			fmt.Println("\t|-- balances")
+			r, _ := balancesSvc.ListPortfolioBalances(context.Background(), &balances.ListPortfolioBalancesRequest{
+				PortfolioId: p.Id,
+			})
+
+			for _, b := range r.Balances {
+				fmt.Printf("\t\t|-- %s: %s\n", b.Symbol, b.Amount)
+
+				amount, _ := big.NewFloat(0).SetString(b.Amount)
+				amount = amount.Mul(amount, big.NewFloat(100))
+
+				account.Balances = append(account.Balances, Balance{
+					Currency: fmt.Sprintf("%s/2", strings.ToUpper(b.Symbol)),
+					Amount:   amount.String(),
+				})
+			}
+		}
+
+		{
+			fmt.Println("\t|-- transactions")
+			r, _ := transactionsSvc.ListPortfolioTransactions(context.Background(), &transactions.ListPortfolioTransactionsRequest{
+				PortfolioId: p.Id,
+			})
+
+			for _, t := range r.Transactions {
+				fmt.Printf(
+					"\t\t|-- %s: %s [%s > %s] [wallet: %s] %s\n",
+					t.Symbol,
+					t.Amount,
+					t.TransferFrom.Type,
+					t.TransferTo.Type,
+					t.WalletId,
+					t.Type,
+				)
+			}
+		}
+
+		{
+			fmt.Println("\t|-- wallets")
+			r, _ := walletsSvc.ListWallets(context.Background(), &wallets.ListWalletsRequest{
+				PortfolioId: p.Id,
+			})
+
+			for _, w := range r.Wallets {
+				account := Account{
+					Id:          w.Id,
+					AccountName: w.Name,
+					CreatedAt:   time.Now().Format(time.RFC3339),
+					Metadata: map[string]string{
+						"spec.coinbase.com/type":        "wallet",
+						"spec.coinbase.com/wallet_type": w.Type,
+					},
+				}
+				export.Accounts = append(export.Accounts, account)
+
+				fmt.Printf("\t\t|-- %s: %s (%s)\n", w.Id, w.Name, w.Type)
+
+				// {
+				// 	r, err := balancesSvc.GetWalletBalance(
+				// 		context.Background(),
+				// 		&balances.GetWalletBalanceRequest{
+				// 			Id: w.Id,
+				// 		},
+				// 	)
+
+				// 	if err != nil {
+				// 		log.Printf("unable to get wallet balance: %v", err)
+				// 		continue
+				// 	}
+
+				// 	fmt.Println(r.Balance.Amount, r.Balance.Symbol)
+				// }
+			}
+		}
+
+		export.Accounts = append(export.Accounts, account)
+	}
+
+	file, err := os.Create("export.json")
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer file.Close()
+
+	json.NewEncoder(file).Encode(export)
+}