From 1c20bd6d8cb4e60fa0b811b966c32c13328fc2ce Mon Sep 17 00:00:00 2001 From: Paul Nicolas Date: Fri, 1 Sep 2023 16:48:00 +0200 Subject: [PATCH] payments: add modulr transfer and payouts --- .../connectors/mangopay/client/transfer.go | 2 +- .../app/connectors/mangopay/task_payments.go | 2 + .../app/connectors/modulr/client/accounts.go | 4 +- .../connectors/modulr/client/beneficiaries.go | 3 +- .../app/connectors/modulr/client/client.go | 22 +- .../app/connectors/modulr/client/payout.go | 13 +- .../connectors/modulr/client/transactions.go | 3 +- .../app/connectors/modulr/client/transfer.go | 54 +++- .../app/connectors/modulr/connector.go | 28 +- .../app/connectors/modulr/task_payments.go | 304 ++++++++++++++++++ .../app/connectors/modulr/task_resolve.go | 22 +- 11 files changed, 417 insertions(+), 40 deletions(-) create mode 100644 components/payments/internal/app/connectors/modulr/task_payments.go diff --git a/components/payments/internal/app/connectors/mangopay/client/transfer.go b/components/payments/internal/app/connectors/mangopay/client/transfer.go index 2dd2191870..33dd2e894a 100644 --- a/components/payments/internal/app/connectors/mangopay/client/transfer.go +++ b/components/payments/internal/app/connectors/mangopay/client/transfer.go @@ -50,7 +50,7 @@ func (c *Client) InitiateWalletTransfer(ctx context.Context, transferRequest *Tr req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewBuffer(body)) if err != nil { - return nil, fmt.Errorf("failed to create login request: %w", err) + return nil, fmt.Errorf("failed to create transfer request: %w", err) } req.Header.Set("Content-Type", "application/json") diff --git a/components/payments/internal/app/connectors/mangopay/task_payments.go b/components/payments/internal/app/connectors/mangopay/task_payments.go index bcc93b93b6..3c5af42b8a 100644 --- a/components/payments/internal/app/connectors/mangopay/task_payments.go +++ b/components/payments/internal/app/connectors/mangopay/task_payments.go @@ -42,6 +42,8 @@ func taskInitiatePayment(logger logging.Logger, mangopayClient *client.Client, t var err error defer func() { if err != nil { + ctx, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() metricsRegistry.ConnectorObjectsErrors().Add(ctx, 1, attrs) if err := ingester.UpdateTransferInitiationStatus(ctx, transferInitiationID, models.TransferInitiationStatusFailed, err.Error(), time.Now()); err != nil { logger.Error("failed to update transfer initiation status: %v", err) diff --git a/components/payments/internal/app/connectors/modulr/client/accounts.go b/components/payments/internal/app/connectors/modulr/client/accounts.go index 67e89d7728..6929da841c 100644 --- a/components/payments/internal/app/connectors/modulr/client/accounts.go +++ b/components/payments/internal/app/connectors/modulr/client/accounts.go @@ -2,7 +2,6 @@ package client import ( "encoding/json" - "fmt" "net/http" ) @@ -28,11 +27,10 @@ func (m *Client) GetAccounts() ([]*Account, error) { if err != nil { return nil, err } - defer resp.Body.Close() if resp.StatusCode != http.StatusOK { - return nil, fmt.Errorf("unexpected status code: %d", resp.StatusCode) + return nil, getError(resp) } var res responseWrapper[[]*Account] diff --git a/components/payments/internal/app/connectors/modulr/client/beneficiaries.go b/components/payments/internal/app/connectors/modulr/client/beneficiaries.go index 5a67ef96be..54045dd859 100644 --- a/components/payments/internal/app/connectors/modulr/client/beneficiaries.go +++ b/components/payments/internal/app/connectors/modulr/client/beneficiaries.go @@ -2,7 +2,6 @@ package client import ( "encoding/json" - "fmt" "net/http" ) @@ -21,7 +20,7 @@ func (m *Client) GetBeneficiaries() ([]*Beneficiary, error) { defer resp.Body.Close() if resp.StatusCode != http.StatusOK { - return nil, fmt.Errorf("unexpected status code: %d", resp.StatusCode) + return nil, getError(resp) } var res responseWrapper[[]*Beneficiary] diff --git a/components/payments/internal/app/connectors/modulr/client/client.go b/components/payments/internal/app/connectors/modulr/client/client.go index 06a050de19..b9223cbd3d 100644 --- a/components/payments/internal/app/connectors/modulr/client/client.go +++ b/components/payments/internal/app/connectors/modulr/client/client.go @@ -1,8 +1,10 @@ package client import ( + "encoding/json" "fmt" "net/http" + "strings" "github.com/formancehq/payments/internal/app/connectors/modulr/hmac" "go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp" @@ -34,7 +36,8 @@ type Client struct { } func (m *Client) buildEndpoint(path string, args ...interface{}) string { - return fmt.Sprintf("%s/%s", m.endpoint, fmt.Sprintf(path, args...)) + endpoint := strings.TrimSuffix(m.endpoint, "/") + return fmt.Sprintf("%s/%s", endpoint, fmt.Sprintf(path, args...)) } const sandboxAPIEndpoint = "https://api-sandbox.modulrfinance.com/api-sandbox-token" @@ -60,3 +63,20 @@ func NewClient(apiKey, apiSecret, endpoint string) (*Client, error) { endpoint: endpoint, }, nil } + +type ErrorResponse struct { + Field string `json:"field"` + Code string `json:"code"` + Message string `json:"message"` + ErrorCode string `json:"errorCode"` + SourceService string `json:"sourceService"` +} + +func getError(resp *http.Response) error { + var errorResponse ErrorResponse + if err := json.NewDecoder(resp.Body).Decode(&errorResponse); err != nil { + return fmt.Errorf("unexpected status code: %d", resp.StatusCode) + } + + return fmt.Errorf("%s: %s", errorResponse.ErrorCode, errorResponse.Message) +} diff --git a/components/payments/internal/app/connectors/modulr/client/payout.go b/components/payments/internal/app/connectors/modulr/client/payout.go index e62adc8f00..6ebe77ff35 100644 --- a/components/payments/internal/app/connectors/modulr/client/payout.go +++ b/components/payments/internal/app/connectors/modulr/client/payout.go @@ -3,7 +3,6 @@ package client import ( "bytes" "encoding/json" - "fmt" "math/big" "net/http" ) @@ -14,10 +13,10 @@ type PayoutRequest struct { Type string `json:"type"` ID string `json:"id"` } `json:"destination"` - Currency string `json:"currency"` - Amount *big.Int `json:"amount"` - Reference string `json:"reference"` - ExternalReference string `json:"externalReference"` + Currency string `json:"currency"` + Amount *big.Float `json:"amount"` + Reference string `json:"reference"` + ExternalReference string `json:"externalReference"` } type PayoutResponse struct { @@ -42,7 +41,7 @@ func (c *Client) InitiatePayout(payoutRequest *PayoutRequest) (*PayoutResponse, defer resp.Body.Close() if resp.StatusCode != http.StatusCreated { - return nil, fmt.Errorf("unexpected status code: %d", resp.StatusCode) + return nil, getError(resp) } var res PayoutResponse @@ -61,7 +60,7 @@ func (c *Client) GetPayout(payoutID string) (*PayoutResponse, error) { defer resp.Body.Close() if resp.StatusCode != http.StatusOK { - return nil, fmt.Errorf("unexpected status code: %d", resp.StatusCode) + return nil, getError(resp) } var res PayoutResponse diff --git a/components/payments/internal/app/connectors/modulr/client/transactions.go b/components/payments/internal/app/connectors/modulr/client/transactions.go index 61345639e1..64ef0d65e9 100644 --- a/components/payments/internal/app/connectors/modulr/client/transactions.go +++ b/components/payments/internal/app/connectors/modulr/client/transactions.go @@ -2,7 +2,6 @@ package client import ( "encoding/json" - "fmt" "net/http" ) @@ -29,7 +28,7 @@ func (m *Client) GetTransactions(accountID string) ([]Transaction, error) { defer resp.Body.Close() if resp.StatusCode != http.StatusOK { - return nil, fmt.Errorf("unexpected status code: %d", resp.StatusCode) + return nil, getError(resp) } var res responseWrapper[[]Transaction] diff --git a/components/payments/internal/app/connectors/modulr/client/transfer.go b/components/payments/internal/app/connectors/modulr/client/transfer.go index 6e6f9002b1..88648411b8 100644 --- a/components/payments/internal/app/connectors/modulr/client/transfer.go +++ b/components/payments/internal/app/connectors/modulr/client/transfer.go @@ -8,16 +8,30 @@ import ( "net/http" ) +type DestinationType string + +const ( + DestinationTypeAccount DestinationType = "ACCOUNT" + DestinationTypeBeneficiary DestinationType = "BENEFICIARY" +) + +type Destination struct { + Type string `json:"type"` + ID string `json:"id"` +} + type TransferRequest struct { - SourceAccountID string `json:"sourceAccountId"` - Destination struct { - Type string `json:"type"` - ID string `json:"id"` - } `json:"destination"` - Currency string `json:"currency"` - Amount *big.Int `json:"amount"` - Reference string `json:"reference"` - ExternalReference string `json:"externalReference"` + SourceAccountID string `json:"sourceAccountId"` + Destination Destination `json:"destination"` + Currency string `json:"currency"` + Amount *big.Float `json:"amount"` + Reference string `json:"reference"` + ExternalReference string `json:"externalReference"` + PaymentDate string `json:"paymentDate"` +} + +type getTransferResponse struct { + Content []*TransferResponse `json:"content"` } type TransferResponse struct { @@ -35,14 +49,20 @@ func (c *Client) InitiateTransfer(transferRequest *TransferRequest) (*TransferRe return nil, err } - resp, err := c.httpClient.Post(c.buildEndpoint("payments"), "application/json", bytes.NewBuffer(body)) + req, err := http.NewRequest(http.MethodPost, c.buildEndpoint("payments"), bytes.NewBuffer(body)) if err != nil { - return nil, err + return nil, fmt.Errorf("failed to create transfer request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + + resp, err := c.httpClient.Do(req) + if err != nil { + return nil, fmt.Errorf("failed to initiate transfer: %w", err) } defer resp.Body.Close() if resp.StatusCode != http.StatusCreated { - return nil, fmt.Errorf("unexpected status code: %d", resp.StatusCode) + return nil, getError(resp) } var res TransferResponse @@ -61,13 +81,17 @@ func (c *Client) GetTransfer(transferID string) (*TransferResponse, error) { defer resp.Body.Close() if resp.StatusCode != http.StatusOK { - return nil, fmt.Errorf("unexpected status code: %d", resp.StatusCode) + return nil, getError(resp) } - var res TransferResponse + var res getTransferResponse if err = json.NewDecoder(resp.Body).Decode(&res); err != nil { return nil, err } - return &res, nil + if len(res.Content) == 0 { + return nil, fmt.Errorf("transfer not found") + } + + return res.Content[0], nil } diff --git a/components/payments/internal/app/connectors/modulr/connector.go b/components/payments/internal/app/connectors/modulr/connector.go index 21bf55e564..c696224c75 100644 --- a/components/payments/internal/app/connectors/modulr/connector.go +++ b/components/payments/internal/app/connectors/modulr/connector.go @@ -9,6 +9,7 @@ import ( "github.com/formancehq/payments/internal/app/integration" "github.com/formancehq/payments/internal/app/task" + "github.com/formancehq/stack/libs/go-libs/contextutil" "github.com/formancehq/stack/libs/go-libs/logging" ) @@ -26,8 +27,31 @@ type Connector struct { } func (c *Connector) InitiatePayment(ctx task.ConnectorContext, transfer *models.TransferInitiation) error { - // TODO implement me - return errors.New("not implemented") + // Detach the context since we're launching an async task and we're mostly + // coming from a HTTP request. + detachedCtx, _ := contextutil.Detached(ctx.Context()) + taskDescriptor, err := models.EncodeTaskDescriptor(TaskDescriptor{ + Name: "Initiate payment", + Key: taskNameInitiatePayment, + TransferID: transfer.ID.String(), + }) + if err != nil { + return err + } + + err = ctx.Scheduler().Schedule(detachedCtx, taskDescriptor, models.TaskSchedulerOptions{ + // We want to polling every c.cfg.PollingPeriod.Duration seconds the users + // and their transactions. + ScheduleOption: models.OPTIONS_RUN_NOW, + // No need to restart this task, since the connector is not existing or + // was uninstalled previously, the task does not exists in the database + Restart: true, + }) + if err != nil && !errors.Is(err, task.ErrAlreadyScheduled) { + return err + } + + return nil } func (c *Connector) Install(ctx task.ConnectorContext) error { diff --git a/components/payments/internal/app/connectors/modulr/task_payments.go b/components/payments/internal/app/connectors/modulr/task_payments.go new file mode 100644 index 0000000000..8d7d1d19a8 --- /dev/null +++ b/components/payments/internal/app/connectors/modulr/task_payments.go @@ -0,0 +1,304 @@ +package modulr + +import ( + "context" + "math/big" + "regexp" + "time" + + "github.com/formancehq/payments/internal/app/connectors/currency" + "github.com/formancehq/payments/internal/app/connectors/modulr/client" + "github.com/formancehq/payments/internal/app/ingestion" + "github.com/formancehq/payments/internal/app/metrics" + "github.com/formancehq/payments/internal/app/models" + "github.com/formancehq/payments/internal/app/storage" + "github.com/formancehq/payments/internal/app/task" + "github.com/formancehq/stack/libs/go-libs/contextutil" + "github.com/formancehq/stack/libs/go-libs/logging" + "github.com/pkg/errors" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/metric" +) + +var ( + initiateTransferAttrs = metric.WithAttributes(append(connectorAttrs, attribute.String(metrics.ObjectAttributeKey, "initiate_transfer"))...) + initiatePayoutAttrs = metric.WithAttributes(append(connectorAttrs, attribute.String(metrics.ObjectAttributeKey, "initiate_payout"))...) + ReferencePatternRegexp = regexp.MustCompile("[a-zA-Z0-9 ]*") +) + +func taskInitiatePayment(logger logging.Logger, modulrClient *client.Client, transferID string) task.Task { + return func( + ctx context.Context, + ingester ingestion.Ingester, + scheduler task.Scheduler, + storageReader storage.Reader, + metricsRegistry metrics.MetricsRegistry, + ) error { + logger.Info("initiate payment for transfer-initiation %s", transferID) + + transferInitiationID := models.MustTransferInitiationIDFromString(transferID) + + attrs := metric.WithAttributes(connectorAttrs...) + var err error + defer func() { + if err != nil { + ctx, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() + metricsRegistry.ConnectorObjectsErrors().Add(ctx, 1, attrs) + if err := ingester.UpdateTransferInitiationStatus(ctx, transferInitiationID, models.TransferInitiationStatusFailed, err.Error(), time.Now()); err != nil { + logger.Error("failed to update transfer initiation status: %v", err) + } + } + }() + + err = ingester.UpdateTransferInitiationStatus(ctx, transferInitiationID, models.TransferInitiationStatusProcessing, "", time.Now()) + if err != nil { + return err + } + + var transfer *models.TransferInitiation + transfer, err = getTransfer(ctx, storageReader, transferInitiationID, true) + if err != nil { + return err + } + + attrs = initiateTransferAttrs + if transfer.Type == models.TransferInitiationTypePayout { + attrs = initiatePayoutAttrs + } + + logger.Info("initiate payment between", transfer.SourceAccountID, " and %s", transfer.DestinationAccountID) + + now := time.Now() + defer func() { + metricsRegistry.ConnectorObjectsLatency().Record(ctx, time.Since(now).Milliseconds(), attrs) + }() + + if transfer.SourceAccount.Type == models.AccountTypeExternal { + err = errors.New("payin not implemented: source account must be an internal account") + return err + } + + var curr string + curr, _, err = currency.GetCurrencyAndPrecisionFromAsset(transfer.Asset) + if err != nil { + return err + } + + amount := big.NewFloat(0).SetInt(transfer.Amount) + amount = amount.Quo(amount, big.NewFloat(100)) + + reference := "" + if len(transfer.Description) <= 18 && ReferencePatternRegexp.MatchString(transfer.Description) { + reference = transfer.Description + } + + var connectorPaymentID string + ctx, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() + switch transfer.DestinationAccount.Type { + case models.AccountTypeInternal: + // Transfer between internal accounts + var resp *client.TransferResponse + resp, err = modulrClient.InitiateTransfer(&client.TransferRequest{ + SourceAccountID: transfer.SourceAccount.Reference, + Destination: client.Destination{ + Type: string(client.DestinationTypeAccount), + ID: transfer.DestinationAccount.Reference, + }, + Currency: curr, + Amount: amount, + Reference: reference, + ExternalReference: reference, + PaymentDate: time.Now().Add(24 * time.Hour).Format("2006-01-02"), + }) + if err != nil { + return err + } + + connectorPaymentID = resp.ID + case models.AccountTypeExternal: + // Payout to an external account + var resp *client.PayoutResponse + resp, err = modulrClient.InitiatePayout(&client.PayoutRequest{ + SourceAccountID: transfer.SourceAccount.Reference, + Destination: client.Destination{ + Type: string(client.DestinationTypeBeneficiary), + ID: transfer.DestinationAccountID.Reference, + }, + Currency: curr, + Amount: amount, + Reference: reference, + ExternalReference: reference, + }) + if err != nil { + return err + } + + connectorPaymentID = resp.ID + } + metricsRegistry.ConnectorObjects().Add(ctx, 1, attrs) + + err = ingester.UpdateTransferInitiationPaymentReference(ctx, transferInitiationID, connectorPaymentID, time.Now()) + if err != nil { + return err + } + + taskDescriptor, err := models.EncodeTaskDescriptor(TaskDescriptor{ + Name: "Update transfer initiation status", + Key: taskNameUpdatePaymentStatus, + TransferID: transfer.ID.String(), + Attempt: 1, + }) + if err != nil { + return err + } + + ctx, _ = contextutil.DetachedWithTimeout(ctx, 10*time.Second) + err = scheduler.Schedule(ctx, taskDescriptor, models.TaskSchedulerOptions{ + // We want to polling every c.cfg.PollingPeriod.Duration seconds the users + // and their transactions. + ScheduleOption: models.OPTIONS_RUN_NOW, + // No need to restart this task, since the connector is not existing or + // was uninstalled previously, the task does not exists in the database + Restart: true, + }) + if err != nil && !errors.Is(err, task.ErrAlreadyScheduled) { + return err + } + + return nil + } +} + +var ( + updateTransferAttrs = metric.WithAttributes(append(connectorAttrs, attribute.String(metrics.ObjectAttributeKey, "update_transfer"))...) + updatePayoutAttrs = metric.WithAttributes(append(connectorAttrs, attribute.String(metrics.ObjectAttributeKey, "update_payout"))...) +) + +func taskUpdatePaymentStatus( + logger logging.Logger, + modulrClient *client.Client, + transferID string, + attempt int, +) task.Task { + return func( + ctx context.Context, + ingester ingestion.Ingester, + scheduler task.Scheduler, + storageReader storage.Reader, + metricsRegistry metrics.MetricsRegistry, + ) error { + transferInitiationID := models.MustTransferInitiationIDFromString(transferID) + transfer, err := getTransfer(ctx, storageReader, transferInitiationID, false) + if err != nil { + return err + } + logger.Info("attempt: ", attempt, " fetching status of ", transfer.Reference) + + attrs := updateTransferAttrs + if transfer.Type == models.TransferInitiationTypePayout { + attrs = updatePayoutAttrs + } + + now := time.Now() + defer func() { + metricsRegistry.ConnectorObjectsLatency().Record(ctx, time.Since(now).Milliseconds(), attrs) + }() + + defer func() { + if err != nil { + metricsRegistry.ConnectorObjectsErrors().Add(ctx, 1, attrs) + } + }() + + var status string + var resultMessage string + switch transfer.Type { + case models.TransferInitiationTypeTransfer: + var resp *client.TransferResponse + resp, err = modulrClient.GetTransfer(transfer.Reference) + if err != nil { + return err + } + + status = resp.Status + resultMessage = resp.Message + case models.TransferInitiationTypePayout: + var resp *client.PayoutResponse + resp, err = modulrClient.GetPayout(transfer.Reference) + if err != nil { + return err + } + + status = resp.Status + resultMessage = resp.Message + } + + switch status { + case "SUBMITTED", "PENDING_FOR_DATE", "PENDING_FOR_FUNDS", "VALIDATED", "SCREENING_REQ": + taskDescriptor, err := models.EncodeTaskDescriptor(TaskDescriptor{ + Name: "Update transfer initiation status", + Key: taskNameUpdatePaymentStatus, + TransferID: transfer.ID.String(), + Attempt: attempt + 1, + }) + if err != nil { + return err + } + + err = scheduler.Schedule(ctx, taskDescriptor, models.TaskSchedulerOptions{ + ScheduleOption: models.OPTIONS_RUN_IN_DURATION, + Duration: 2 * time.Minute, + Restart: true, + }) + if err != nil && !errors.Is(err, task.ErrAlreadyScheduled) { + return err + } + case "EXT_PROC", "PROCESSED", "RECONCILED": + err = ingester.UpdateTransferInitiationStatus(ctx, transferInitiationID, models.TransferInitiationStatusProcessed, "", time.Now()) + if err != nil { + return err + } + + return nil + default: + err = ingester.UpdateTransferInitiationStatus(ctx, transferInitiationID, models.TransferInitiationStatusFailed, resultMessage, time.Now()) + if err != nil { + return err + } + + return nil + } + + return nil + } +} + +func getTransfer( + ctx context.Context, + reader storage.Reader, + transferID models.TransferInitiationID, + expand bool, +) (*models.TransferInitiation, error) { + transfer, err := reader.ReadTransferInitiation(ctx, transferID) + if err != nil { + return nil, err + } + + if expand { + sourceAccount, err := reader.GetAccount(ctx, transfer.SourceAccountID.String()) + if err != nil { + return nil, err + } + transfer.SourceAccount = sourceAccount + + destinationAccount, err := reader.GetAccount(ctx, transfer.DestinationAccountID.String()) + if err != nil { + return nil, err + } + transfer.DestinationAccount = destinationAccount + } + + return transfer, nil +} diff --git a/components/payments/internal/app/connectors/modulr/task_resolve.go b/components/payments/internal/app/connectors/modulr/task_resolve.go index a49965bf1b..a9b3d5b65d 100644 --- a/components/payments/internal/app/connectors/modulr/task_resolve.go +++ b/components/payments/internal/app/connectors/modulr/task_resolve.go @@ -10,17 +10,21 @@ import ( ) const ( - taskNameMain = "main" - taskNameFetchTransactions = "fetch-transactions" - taskNameFetchAccounts = "fetch-accounts" - taskNameFetchBeneficiaries = "fetch-beneficiaries" + taskNameMain = "main" + taskNameFetchTransactions = "fetch-transactions" + taskNameFetchAccounts = "fetch-accounts" + taskNameFetchBeneficiaries = "fetch-beneficiaries" + taskNameInitiatePayment = "initiate-payment" + taskNameUpdatePaymentStatus = "update-payment-status" ) // TaskDescriptor is the definition of a task. type TaskDescriptor struct { - Name string `json:"name" yaml:"name" bson:"name"` - Key string `json:"key" yaml:"key" bson:"key"` - AccountID string `json:"accountID" yaml:"accountID" bson:"accountID"` + Name string `json:"name" yaml:"name" bson:"name"` + Key string `json:"key" yaml:"key" bson:"key"` + AccountID string `json:"accountID" yaml:"accountID" bson:"accountID"` + TransferID string `json:"transferID" yaml:"transferID" bson:"transferID"` + Attempt int `json:"attempt" yaml:"attempt" bson:"attempt"` } func resolveTasks(logger logging.Logger, config Config) func(taskDefinition TaskDescriptor) task.Task { @@ -41,6 +45,10 @@ func resolveTasks(logger logging.Logger, config Config) func(taskDefinition Task return taskFetchAccounts(logger, modulrClient) case taskNameFetchBeneficiaries: return taskFetchBeneficiaries(logger, modulrClient) + case taskNameInitiatePayment: + return taskInitiatePayment(logger, modulrClient, taskDefinition.TransferID) + case taskNameUpdatePaymentStatus: + return taskUpdatePaymentStatus(logger, modulrClient, taskDefinition.TransferID, taskDefinition.Attempt) case taskNameFetchTransactions: return taskFetchTransactions(logger, modulrClient, taskDefinition.AccountID) }