feat(payments): small improvments + worker pool (#1151)

components/payments/cmd/connectors/internal/api/connectormodule.go

@@ -50,7 +50,7 @@ func addConnector[ConnectorConfig models.ConnectorConfigObject](loader manager.L
 					container := dig.New()
 
 					if err := container.Provide(func() ingestion.Ingester {
-						return ingestion.NewDefaultIngester(loader.Name(), descriptor, store, publisher, messages)
+						return ingestion.NewDefaultIngester(loader.Name(), connectorID, descriptor, store, publisher, messages)
 					}); err != nil {
 						return nil, err
 					}

components/payments/cmd/connectors/internal/connectors/adyen/task_standard_webhooks.go

@@ -189,9 +189,7 @@ func handleAuthorisation(
 
 	if err := ingester.IngestPayments(
 		ctx,
-		connectorID,
 		ingestion.PaymentBatch{{Payment: payment}},
-		struct{}{},
 	); err != nil {
 		return err
 	}
@@ -223,9 +221,7 @@ func handleAuthorisationAdjustment(
 
 		if err := ingester.IngestPayments(
 			ctx,
-			connectorID,
 			ingestion.PaymentBatch{{Payment: payment}},
-			struct{}{},
 		); err != nil {
 			return err
 		}
@@ -257,9 +253,7 @@ func handleCancellation(
 
 		if err := ingester.IngestPayments(
 			ctx,
-			connectorID,
 			ingestion.PaymentBatch{{Payment: payment}},
-			struct{}{},
 		); err != nil {
 			return err
 		}
@@ -292,9 +286,7 @@ func handleCapture(
 
 		if err := ingester.IngestPayments(
 			ctx,
-			connectorID,
 			ingestion.PaymentBatch{{Payment: payment}},
-			struct{}{},
 		); err != nil {
 			return err
 		}
@@ -326,9 +318,7 @@ func handleCaptureFailed(
 
 		if err := ingester.IngestPayments(
 			ctx,
-			connectorID,
 			ingestion.PaymentBatch{{Payment: payment}},
-			struct{}{},
 		); err != nil {
 			return err
 		}
@@ -363,9 +353,7 @@ func handleRefund(
 
 		if err := ingester.IngestPayments(
 			ctx,
-			connectorID,
 			ingestion.PaymentBatch{{Payment: payment}},
-			struct{}{},
 		); err != nil {
 			return err
 		}
@@ -402,9 +390,7 @@ func handleRefundedReversed(
 
 		if err := ingester.IngestPayments(
 			ctx,
-			connectorID,
 			ingestion.PaymentBatch{{Payment: payment}},
-			struct{}{},
 		); err != nil {
 			return err
 		}
@@ -439,9 +425,7 @@ func handleRefundWithData(
 
 		if err := ingester.IngestPayments(
 			ctx,
-			connectorID,
 			ingestion.PaymentBatch{{Payment: payment}},
-			struct{}{},
 		); err != nil {
 			return err
 		}
@@ -492,9 +476,7 @@ func handlePayoutThirdparty(
 
 	if err := ingester.IngestPayments(
 		ctx,
-		connectorID,
 		ingestion.PaymentBatch{{Payment: payment}},
-		struct{}{},
 	); err != nil {
 		return err
 	}
@@ -543,9 +525,7 @@ func handlePayoutDecline(
 
 	if err := ingester.IngestPayments(
 		ctx,
-		connectorID,
 		ingestion.PaymentBatch{{Payment: payment}},
-		struct{}{},
 	); err != nil {
 		return err
 	}
@@ -594,9 +574,7 @@ func handlePayoutExpire(
 
 	if err := ingester.IngestPayments(
 		ctx,
-		connectorID,
 		ingestion.PaymentBatch{{Payment: payment}},
-		struct{}{},
 	); err != nil {
 		return err
 	}

components/payments/cmd/connectors/internal/connectors/atlar/task_fetch_transactions.go

@@ -133,7 +133,7 @@ func ingestPaymentsBatch(
 		batch = append(batch, batchElement)
 	}
 
-	if err := ingester.IngestPayments(ctx, connectorID, batch, struct{}{}); err != nil {
+	if err := ingester.IngestPayments(ctx, batch); err != nil {
 		return err
 	}
 

components/payments/cmd/connectors/internal/connectors/bankingcircle/task_fetch_payments.go

@@ -127,7 +127,7 @@ func ingestBatch(
 		batch = append(batch, batchElement)
 	}
 
-	if err := ingester.IngestPayments(ctx, connectorID, batch, struct{}{}); err != nil {
+	if err := ingester.IngestPayments(ctx, batch); err != nil {
 		return err
 	}
 

components/payments/cmd/connectors/internal/connectors/currencycloud/task_fetch_transactions.go

@@ -119,7 +119,7 @@ func ingestTransactions(
 			batch = append(batch, batchElement)
 		}
 
-		err = ingester.IngestPayments(ctx, connectorID, batch, struct{}{})
+		err = ingester.IngestPayments(ctx, batch)
 		if err != nil {
 			return err
 		}

components/payments/cmd/connectors/internal/connectors/dummypay/connector_test.go

@@ -86,14 +86,18 @@ func (m *MockIngester) IngestAccounts(ctx context.Context, batch ingestion.Accou
 	return nil
 }
 
-func (m *MockIngester) IngestPayments(ctx context.Context, connectorID models.ConnectorID, batch ingestion.PaymentBatch, commitState any) error {
+func (m *MockIngester) IngestPayments(ctx context.Context, batch ingestion.PaymentBatch) error {
 	return nil
 }
 
 func (m *MockIngester) IngestBalances(ctx context.Context, batch ingestion.BalanceBatch, checkIfAccountExists bool) error {
 	return nil
 }
 
+func (m *MockIngester) UpdateTaskState(ctx context.Context, state any) error {
+	return nil
+}
+
 func (m *MockIngester) UpdateTransferInitiationPaymentsStatus(ctx context.Context, tf *models.TransferInitiation, paymentID *models.PaymentID, status models.TransferInitiationStatus, errorMessage string, updatedAt time.Time) error {
 	return nil
 }

components/payments/cmd/connectors/internal/connectors/dummypay/task_ingest.go

@@ -70,7 +70,7 @@ func handleFile(
 			return err
 		}
 
-		err = ingester.IngestPayments(ctx, connectorID, batch, struct{}{})
+		err = ingester.IngestPayments(ctx, batch)
 		if err != nil {
 			return fmt.Errorf("failed to ingest file '%s': %w", descriptor.FileName, err)
 		}

components/payments/cmd/connectors/internal/connectors/dummypay/task_init_directory.go

@@ -179,7 +179,7 @@ func generatePaymentsFile(
 	if err != nil {
 		return fmt.Errorf("failed to marshal payment: %w", err)
 	}
-	if err := ingester.IngestPayments(ctx, connectorID, ingestion.PaymentBatch{
+	if err := ingester.IngestPayments(ctx, ingestion.PaymentBatch{
 		{
 			Payment: &models.Payment{
 				ID: models.PaymentID{
@@ -203,7 +203,7 @@ func generatePaymentsFile(
 				DestinationAccountID: destinationAccountID,
 			},
 		},
-	}, struct{}{}); err != nil {
+	}); err != nil {
 		return fmt.Errorf("failed to ingest payments: %w", err)
 	}
 

components/payments/cmd/connectors/internal/connectors/mangopay/task_fetch_transactions.go

@@ -124,7 +124,7 @@ func ingestBatch(
 		batch = append(batch, batchElement)
 	}
 
-	return ingester.IngestPayments(ctx, connectorID, batch, struct{}{})
+	return ingester.IngestPayments(ctx, batch)
 }
 
 func matchPaymentType(paymentType string) models.PaymentType {

components/payments/cmd/connectors/internal/connectors/modulr/task_fetch_transactions.go

@@ -64,7 +64,7 @@ func fetchTransactions(
 			return err
 		}
 
-		if err := ingester.IngestPayments(ctx, connectorID, batch, struct{}{}); err != nil {
+		if err := ingester.IngestPayments(ctx, batch); err != nil {
 			return err
 		}
 

components/payments/cmd/connectors/internal/connectors/moneycorp/task_fetch_transactions.go

@@ -161,7 +161,7 @@ func ingestBatch(
 		batch = append(batch, batchElement)
 	}
 
-	return ingester.IngestPayments(ctx, connectorID, batch, struct{}{})
+	return ingester.IngestPayments(ctx, batch)
 }
 
 func matchPaymentType(transactionType string, transactionDirection string) (models.PaymentType, bool) {

components/payments/cmd/connectors/internal/connectors/stripe/task_fetch_payments_for_connected_account.go

@@ -46,7 +46,12 @@ func ingestBatch(
 		"state": commitState,
 	}).Debugf("updating state")
 
-	err := ingester.IngestPayments(ctx, connectorID, batch, commitState)
+	err := ingester.IngestPayments(ctx, batch)
+	if err != nil {
+		return err
+	}
+
+	err = ingester.UpdateTaskState(ctx, commitState)
 	if err != nil {
 		return err
 	}

components/payments/cmd/connectors/internal/connectors/wise/task_fetch_transfers.go

@@ -126,7 +126,7 @@ func fetchTransfers(
 		paymentBatch = append(paymentBatch, batchElement)
 	}
 
-	if err := ingester.IngestPayments(ctx, connectorID, paymentBatch, struct{}{}); err != nil {
+	if err := ingester.IngestPayments(ctx, paymentBatch); err != nil {
 		return err
 	}
 

components/payments/cmd/connectors/internal/ingestion/ingester.go

@@ -13,20 +13,22 @@ import (
 
 type Ingester interface {
 	IngestAccounts(ctx context.Context, batch AccountBatch) error
-	IngestPayments(ctx context.Context, connectorID models.ConnectorID, batch PaymentBatch, commitState any) error
+	IngestPayments(ctx context.Context, batch PaymentBatch) error
 	IngestBalances(ctx context.Context, batch BalanceBatch, checkIfAccountExists bool) error
+	UpdateTaskState(ctx context.Context, state any) error
 	UpdateTransferInitiationPaymentsStatus(ctx context.Context, tf *models.TransferInitiation, paymentID *models.PaymentID, status models.TransferInitiationStatus, errorMessage string, updatedAt time.Time) error
 	UpdateTransferReversalStatus(ctx context.Context, transfer *models.TransferInitiation, transferReversal *models.TransferReversal) error
 	AddTransferInitiationPaymentID(ctx context.Context, tf *models.TransferInitiation, paymentID *models.PaymentID, updatedAt time.Time) error
 	LinkBankAccountWithAccount(ctx context.Context, bankAccount *models.BankAccount, accountID *models.AccountID) error
 }
 
 type DefaultIngester struct {
-	provider   models.ConnectorProvider
-	store      Store
-	descriptor models.TaskDescriptor
-	publisher  message.Publisher
-	messages   *messages.Messages
+	provider    models.ConnectorProvider
+	connectorID models.ConnectorID
+	store       Store
+	descriptor  models.TaskDescriptor
+	publisher   message.Publisher
+	messages    *messages.Messages
 }
 
 type Store interface {
@@ -42,17 +44,19 @@ type Store interface {
 
 func NewDefaultIngester(
 	provider models.ConnectorProvider,
+	connectorID models.ConnectorID,
 	descriptor models.TaskDescriptor,
 	repo Store,
 	publisher message.Publisher,
 	messages *messages.Messages,
 ) *DefaultIngester {
 	return &DefaultIngester{
-		provider:   provider,
-		descriptor: descriptor,
-		store:      repo,
-		publisher:  publisher,
-		messages:   messages,
+		provider:    provider,
+		connectorID: connectorID,
+		descriptor:  descriptor,
+		store:       repo,
+		publisher:   publisher,
+		messages:    messages,
 	}
 }
 

components/payments/cmd/connectors/internal/ingestion/payments.go

@@ -2,7 +2,6 @@ package ingestion
 
 import (
 	"context"
-	"encoding/json"
 	"fmt"
 	"time"
 
@@ -29,9 +28,7 @@ func (fn IngesterFn) IngestPayments(ctx context.Context, batch PaymentBatch, com
 
 func (i *DefaultIngester) IngestPayments(
 	ctx context.Context,
-	connectorID models.ConnectorID,
 	batch PaymentBatch,
-	commitState any,
 ) error {
 	startingAt := time.Now()
 
@@ -62,15 +59,6 @@ func (i *DefaultIngester) IngestPayments(
 		idsInsertedMap[idsInserted[idx].String()] = struct{}{}
 	}
 
-	taskState, err := json.Marshal(commitState)
-	if err != nil {
-		return fmt.Errorf("error marshaling task state: %w", err)
-	}
-
-	if err = i.store.UpdateTaskState(ctx, connectorID, i.descriptor, taskState); err != nil {
-		return fmt.Errorf("error updating task state: %w", err)
-	}
-
 	for paymentIdx := range allPayments {
 		_, ok := idsInsertedMap[allPayments[paymentIdx].ID.String()]
 		if !ok {

components/payments/cmd/connectors/internal/ingestion/payments_test.go

@@ -166,13 +166,14 @@ func TestIngestPayments(t *testing.T) {
 
 			ingester := NewDefaultIngester(
 				models.ConnectorProviderDummyPay,
+				connectorID,
 				nil,
 				NewMockStore().WithPaymentIDsNotModified(tc.paymentIDsNotModified),
 				publisher,
 				messages.NewMessages(""),
 			)
 
-			err := ingester.IngestPayments(context.Background(), connectorID, tc.batch, nil)
+			err := ingester.IngestPayments(context.Background(), tc.batch)
 			publisher.Close()
 			require.NoError(t, err)
 

components/payments/cmd/connectors/internal/ingestion/task.go

@@ -0,0 +1,20 @@
+package ingestion
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+)
+
+func (i *DefaultIngester) UpdateTaskState(ctx context.Context, state any) error {
+	taskState, err := json.Marshal(state)
+	if err != nil {
+		return fmt.Errorf("error marshaling task state: %w", err)
+	}
+
+	if err = i.store.UpdateTaskState(ctx, i.connectorID, i.descriptor, taskState); err != nil {
+		return fmt.Errorf("error updating task state: %w", err)
+	}
+
+	return nil
+}