feat(currenclycloud): improve loops and add tests

go.mod

@@ -36,6 +36,7 @@ require (
 	go.uber.org/fx v1.22.2
 	go.uber.org/mock v0.4.0
 	golang.org/x/oauth2 v0.23.0
+	golang.org/x/sync v0.8.0
 	google.golang.org/grpc v1.67.0
 	google.golang.org/protobuf v1.34.2
 )
@@ -197,7 +198,6 @@ require (
 	golang.org/x/crypto v0.27.0 // indirect
 	golang.org/x/exp v0.0.0-20240909161429-701f63a606c0 // indirect
 	golang.org/x/net v0.29.0 // indirect
-	golang.org/x/sync v0.8.0 // indirect
 	golang.org/x/sys v0.25.0 // indirect
 	golang.org/x/text v0.18.0 // indirect
 	golang.org/x/time v0.6.0 // indirect

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

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"time"
 
+	"github.com/formancehq/payments/internal/connectors/plugins/public/currencycloud/client"
 	"github.com/formancehq/payments/internal/models"
 )
 
@@ -43,46 +44,24 @@ func (p Plugin) fetchNextAccounts(ctx context.Context, req models.FetchNextAccou
 			break
 		}
 
-		for _, account := range pagedAccounts {
-			switch account.CreatedAt.Compare(oldState.LastCreatedAt) {
-			case -1, 0:
-				// Account already ingested, skip
-				continue
-			default:
-			}
-
-			raw, err := json.Marshal(account)
-			if err != nil {
-				return models.FetchNextAccountsResponse{}, err
-			}
-
-			accounts = append(accounts, models.PSPAccount{
-				Reference: account.ID,
-				CreatedAt: account.CreatedAt,
-				Name:      &account.AccountName,
-				Raw:       raw,
-			})
-
-			newState.LastCreatedAt = account.CreatedAt
-
-			if len(accounts) >= req.PageSize {
-				break
-			}
-		}
-
-		if len(accounts) >= req.PageSize {
-			hasMore = true
-			break
+		accounts, err = fillAccounts(accounts, pagedAccounts, oldState)
+		if err != nil {
+			return models.FetchNextAccountsResponse{}, err
 		}
 
-		if nextPage == -1 {
+		needMore := true
+		needMore, hasMore, accounts = shouldFetchMore(accounts, nextPage, req.PageSize)
+		if !needMore {
 			break
 		}
 
 		page = nextPage
 	}
 
 	newState.LastPage = page
+	if len(accounts) > 0 {
+		newState.LastCreatedAt = accounts[len(accounts)-1].CreatedAt
+	}
 
 	payload, err := json.Marshal(newState)
 	if err != nil {
@@ -95,3 +74,32 @@ func (p Plugin) fetchNextAccounts(ctx context.Context, req models.FetchNextAccou
 		HasMore:  hasMore,
 	}, nil
 }
+
+func fillAccounts(
+	accounts []models.PSPAccount,
+	pagedAccounts []*client.Account,
+	oldState accountsState,
+) ([]models.PSPAccount, error) {
+	for _, account := range pagedAccounts {
+		switch account.CreatedAt.Compare(oldState.LastCreatedAt) {
+		case -1, 0:
+			// Account already ingested, skip
+			continue
+		default:
+		}
+
+		raw, err := json.Marshal(account)
+		if err != nil {
+			return nil, err
+		}
+
+		accounts = append(accounts, models.PSPAccount{
+			Reference: account.ID,
+			CreatedAt: account.CreatedAt,
+			Name:      &account.AccountName,
+			Raw:       raw,
+		})
+	}
+
+	return accounts, nil
+}

internal/connectors/plugins/public/currencycloud/accounts_test.go

@@ -0,0 +1,176 @@
+package currencycloud
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/formancehq/payments/internal/connectors/plugins/public/currencycloud/client"
+	"github.com/formancehq/payments/internal/models"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	"go.uber.org/mock/gomock"
+)
+
+var _ = Describe("CurrencyCloud Plugin Accounts", func() {
+	var (
+		plg *Plugin
+	)
+
+	BeforeEach(func() {
+		plg = &Plugin{}
+	})
+
+	Context("fetching next accounts", func() {
+		var (
+			m              *client.MockClient
+			sampleAccounts []*client.Account
+			now            time.Time
+		)
+
+		BeforeEach(func() {
+			ctrl := gomock.NewController(GinkgoT())
+			m = client.NewMockClient(ctrl)
+			plg.client = m
+			now = time.Now().UTC()
+
+			sampleAccounts = make([]*client.Account, 0)
+			for i := 0; i < 50; i++ {
+				sampleAccounts = append(sampleAccounts, &client.Account{
+					ID:          fmt.Sprintf("%d", i),
+					AccountName: fmt.Sprintf("Account %d", i),
+					CreatedAt:   now.Add(-time.Duration(50-i) * time.Minute).UTC(),
+					UpdatedAt:   now.Add(-time.Duration(50-i) * time.Minute).UTC(),
+				})
+			}
+		})
+
+		It("should return an error - get accounts error", func(ctx SpecContext) {
+			req := models.FetchNextAccountsRequest{
+				State:    []byte(`{}`),
+				PageSize: 60,
+			}
+
+			m.EXPECT().GetAccounts(ctx, 1, 60).Return(
+				[]*client.Account{},
+				-1,
+				errors.New("test error"),
+			)
+
+			resp, err := plg.FetchNextAccounts(ctx, req)
+			Expect(err).ToNot(BeNil())
+			Expect(err).To(MatchError("test error"))
+			Expect(resp).To(Equal(models.FetchNextAccountsResponse{}))
+		})
+
+		It("should fetch next accounts - no state no results", func(ctx SpecContext) {
+			req := models.FetchNextAccountsRequest{
+				State:    []byte(`{}`),
+				PageSize: 60,
+			}
+
+			m.EXPECT().GetAccounts(ctx, 1, 60).Return(
+				[]*client.Account{},
+				-1,
+				nil,
+			)
+
+			resp, err := plg.FetchNextAccounts(ctx, req)
+			Expect(err).To(BeNil())
+			Expect(resp.Accounts).To(HaveLen(0))
+			Expect(resp.HasMore).To(BeFalse())
+			Expect(resp.NewState).ToNot(BeNil())
+
+			var state accountsState
+			err = json.Unmarshal(resp.NewState, &state)
+			Expect(err).To(BeNil())
+			// We fetched everything, state should be resetted
+			Expect(state.LastPage).To(Equal(1))
+			Expect(state.LastCreatedAt.IsZero()).To(BeTrue())
+		})
+
+		It("should fetch next accounts - no state pageSize > total accounts", func(ctx SpecContext) {
+			req := models.FetchNextAccountsRequest{
+				State:    []byte(`{}`),
+				PageSize: 60,
+			}
+
+			m.EXPECT().GetAccounts(ctx, 1, 60).Return(
+				sampleAccounts,
+				-1,
+				nil,
+			)
+
+			resp, err := plg.FetchNextAccounts(ctx, req)
+			Expect(err).To(BeNil())
+			Expect(resp.Accounts).To(HaveLen(50))
+			Expect(resp.HasMore).To(BeFalse())
+			Expect(resp.NewState).ToNot(BeNil())
+
+			var state accountsState
+			err = json.Unmarshal(resp.NewState, &state)
+			Expect(err).To(BeNil())
+			// We fetched everything, state should be resetted
+			Expect(state.LastPage).To(Equal(1))
+			Expect(state.LastCreatedAt).To(Equal(sampleAccounts[49].CreatedAt))
+		})
+
+		It("should fetch next accounts - no state pageSize < total accounts", func(ctx SpecContext) {
+			req := models.FetchNextAccountsRequest{
+				State:    []byte(`{}`),
+				PageSize: 40,
+			}
+
+			m.EXPECT().GetAccounts(ctx, 1, 40).Return(
+				sampleAccounts[:40],
+				2,
+				nil,
+			)
+
+			resp, err := plg.FetchNextAccounts(ctx, req)
+			Expect(err).To(BeNil())
+			Expect(resp.Accounts).To(HaveLen(40))
+			Expect(resp.HasMore).To(BeTrue())
+			Expect(resp.NewState).ToNot(BeNil())
+
+			var state accountsState
+			err = json.Unmarshal(resp.NewState, &state)
+			Expect(err).To(BeNil())
+			Expect(state.LastPage).To(Equal(1))
+			Expect(state.LastCreatedAt).To(Equal(sampleAccounts[39].CreatedAt))
+		})
+
+		It("should fetch next accounts - with state pageSize < total accounts", func(ctx SpecContext) {
+			req := models.FetchNextAccountsRequest{
+				State:    []byte(fmt.Sprintf(`{"lastPage": %d, "lastCreatedAt": "%s"}`, 1, sampleAccounts[38].CreatedAt.Format(time.RFC3339Nano))),
+				PageSize: 40,
+			}
+
+			m.EXPECT().GetAccounts(ctx, 1, 40).Return(
+				sampleAccounts[:40],
+				2,
+				nil,
+			)
+
+			m.EXPECT().GetAccounts(ctx, 2, 40).Return(
+				sampleAccounts[41:],
+				-1,
+				nil,
+			)
+
+			resp, err := plg.FetchNextAccounts(ctx, req)
+			Expect(err).To(BeNil())
+			Expect(resp.Accounts).To(HaveLen(10))
+			Expect(resp.HasMore).To(BeFalse())
+			Expect(resp.NewState).ToNot(BeNil())
+
+			var state accountsState
+			err = json.Unmarshal(resp.NewState, &state)
+			Expect(err).To(BeNil())
+			// We fetched everything, state should be resetted
+			Expect(state.LastPage).To(Equal(2))
+			Expect(state.LastCreatedAt).To(Equal(sampleAccounts[49].CreatedAt))
+		})
+	})
+})