tech/sendico
2
0
Fork 0
Code Issues 30 Pull Requests Actions Packages Projects Releases Wiki Activity

refactored payment orchestration

Browse Source
This commit is contained in:
Stephan D
2026-02-03 00:40:46 +01:00
parent 05d998e0f7
commit 5e87e2f2f9
184 changed files with 3920 additions and 2219 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
api/payments/orchestrator/go.mod
View File

@@ -15,6 +15,7 @@ replace github.com/tech/sendico/fx/oracle => ../../fx/oracle
replace github.com/tech/sendico/ledger => ../../ledger
require (
github.com/google/uuid v1.6.0
github.com/prometheus/client_golang v1.23.2
github.com/shopspring/decimal v1.4.0
github.com/tech/sendico/fx/oracle v0.0.0-00010101000000-000000000000
@@ -55,7 +56,6 @@ require (
github.com/go-logr/stdr v1.2.2 // indirect
github.com/go-ole/go-ole v1.3.0 // indirect
github.com/gogo/protobuf v1.3.2 // indirect
github.com/google/uuid v1.6.0 // indirect
github.com/klauspost/compress v1.18.3 // indirect
github.com/lufia/plan9stats v0.0.0-20250827001030-24949be3fa54 // indirect
github.com/magiconair/properties v1.8.7 // indirect
@@ -106,5 +106,5 @@ require (
golang.org/x/sync v0.19.0 // indirect
golang.org/x/sys v0.40.0 // indirect
golang.org/x/text v0.33.0 // indirect
google.golang.org/genproto/googleapis/rpc v0.0.0-20260128011058-8636f8732409 // indirect
google.golang.org/genproto/googleapis/rpc v0.0.0-20260202165425-ce8ad4cf556b // indirect
)

4
api/payments/orchestrator/go.sum
View File

@@ -266,8 +266,8 @@ gonum.org/v1/gonum v0.16.0 h1:5+ul4Swaf3ESvrOnidPp4GZbzf0mxVQpDCYUQE7OJfk=
gonum.org/v1/gonum v0.16.0/go.mod h1:fef3am4MQ93R2HHpKnLk4/Tbh/s0+wqD5nfa6Pnwy4E=
google.golang.org/genproto/googleapis/api v0.0.0-20251029180050-ab9386a59fda h1:+2XxjfsAu6vqFxwGBRcHiMaDCuZiqXGDUDVWVtrFAnE=
google.golang.org/genproto/googleapis/api v0.0.0-20251029180050-ab9386a59fda/go.mod h1:fDMmzKV90WSg1NbozdqrE64fkuTv6mlq2zxo9ad+3yo=
google.golang.org/genproto/googleapis/rpc v0.0.0-20260128011058-8636f8732409 h1:H86B94AW+VfJWDqFeEbBPhEtHzJwJfTbgE2lZa54ZAQ=
google.golang.org/genproto/googleapis/rpc v0.0.0-20260128011058-8636f8732409/go.mod h1:j9x/tPzZkyxcgEFkiKEEGxfvyumM01BEtsW8xzOahRQ=
google.golang.org/genproto/googleapis/rpc v0.0.0-20260202165425-ce8ad4cf556b h1:GZxXGdFaHX27ZSMHudWc4FokdD+xl8BC2UJm1OVIEzs=
google.golang.org/genproto/googleapis/rpc v0.0.0-20260202165425-ce8ad4cf556b/go.mod h1:j9x/tPzZkyxcgEFkiKEEGxfvyumM01BEtsW8xzOahRQ=
google.golang.org/grpc v1.78.0 h1:K1XZG/yGDJnzMdd/uZHAkVqJE+xIDOcmdSFZkBUicNc=
google.golang.org/grpc v1.78.0/go.mod h1:I47qjTo4OKbMkjA/aOOwxDIiPSBofUtQUI5EfpWvW7U=
google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE=

33
api/payments/orchestrator/internal/service/orchestrator/card_payout_funding.go
View File

@@ -152,7 +152,7 @@ func (s *Service) submitCardFundingTransfers(ctx context.Context, payment *model
if topUpMoney != nil && topUpPositive {
gasStep = ensureExecutionStep(plan, stepCodeGasTopUp)
setExecutionStepRole(gasStep, executionStepRoleSource)
setExecutionStepStatus(gasStep, executionStepStatusPlanned)
setExecutionStepStatus(gasStep, model.OperationStatePlanned)
gasStep.Description = "Top up native gas from fee wallet"
gasStep.Amount = moneyFromProto(topUpMoney)
gasStep.NetworkFee = moneyFromProto(topUpFee)
@@ -162,7 +162,7 @@ func (s *Service) submitCardFundingTransfers(ctx context.Context, payment *model
fundStep := ensureExecutionStep(plan, stepCodeFundingTransfer)
setExecutionStepRole(fundStep, executionStepRoleSource)
setExecutionStepStatus(fundStep, executionStepStatusPlanned)
setExecutionStepStatus(fundStep, model.OperationStatePlanned)
fundStep.Description = "Transfer payout amount to card funding wallet"
fundStep.Amount = cloneMoney(intentAmount)
fundStep.NetworkFee = moneyFromProto(fundingFee)
@@ -172,7 +172,7 @@ func (s *Service) submitCardFundingTransfers(ctx context.Context, payment *model
if feeRequired {
feeStep = ensureExecutionStep(plan, stepCodeFeeTransfer)
setExecutionStepRole(feeStep, executionStepRoleSource)
setExecutionStepStatus(feeStep, executionStepStatusPlanned)
setExecutionStepStatus(feeStep, model.OperationStatePlanned)
feeStep.Description = "Transfer fee to fee wallet"
feeStep.Amount = cloneMoney(feeAmount)
feeStep.NetworkFee = moneyFromProto(feeTransferFee)
@@ -182,7 +182,7 @@ func (s *Service) submitCardFundingTransfers(ctx context.Context, payment *model
cardStep := ensureExecutionStep(plan, stepCodeCardPayout)
setExecutionStepRole(cardStep, executionStepRoleConsumer)
setExecutionStepStatus(cardStep, executionStepStatusPlanned)
setExecutionStepStatus(cardStep, model.OperationStatePlanned)
cardStep.Description = "Submit card payout"
cardStep.Amount = cloneMoney(payoutAmount)
if card := intent.Destination.Card; card != nil {
@@ -202,11 +202,13 @@ func (s *Service) submitCardFundingTransfers(ctx context.Context, payment *model
ensureResp, gasErr := chainClient.EnsureGasTopUp(ctx, &chainv1.EnsureGasTopUpRequest{
IdempotencyKey: payment.IdempotencyKey + ":card:gas",
OrganizationRef: payment.OrganizationRef.Hex(),
IntentRef: strings.TrimSpace(payment.Intent.Ref),
OperationRef: strings.TrimSpace(cardStep.OperationRef),
SourceWalletRef: feeWalletRef,
TargetWalletRef: sourceWalletRef,
EstimatedTotalFee: estimatedTotalFee,
Metadata: cloneMetadata(payment.Metadata),
ClientReference: payment.PaymentRef,
PaymentRef: payment.PaymentRef,
})
if gasErr != nil {
s.logger.Warn("card gas top-up transfer failed", zap.Error(gasErr), zap.String("payment_ref", payment.PaymentRef))
@@ -241,12 +243,12 @@ func (s *Service) submitCardFundingTransfers(ctx context.Context, payment *model
return err
}
gasStep.NetworkFee = moneyFromProto(topUpFee)
setExecutionStepStatus(gasStep, executionStepStatusSubmitted)
setExecutionStepStatus(gasStep, model.OperationStateWaiting)
} else {
gasStep.Amount = nil
gasStep.NetworkFee = nil
gasStep.TransferRef = ""
setExecutionStepStatus(gasStep, executionStepStatusSkipped)
setExecutionStepStatus(gasStep, model.OperationStateSkipped)
}
}
if gasStep != nil {
@@ -255,6 +257,7 @@ func (s *Service) submitCardFundingTransfers(ctx context.Context, payment *model
updateExecutionPlanTotalNetworkFee(plan)
}
s.logger.Warn("Request", zap.Any("intent", intent))
fundResp, err := chainClient.SubmitTransfer(ctx, &chainv1.SubmitTransferRequest{
IdempotencyKey: payment.IdempotencyKey + ":card:fund",
OrganizationRef: payment.OrganizationRef.Hex(),
@@ -262,7 +265,9 @@ func (s *Service) submitCardFundingTransfers(ctx context.Context, payment *model
Destination: fundingDest,
Amount: cloneProtoMoney(intentAmountProto),
Metadata: cloneMetadata(payment.Metadata),
ClientReference: payment.PaymentRef,
PaymentRef: payment.PaymentRef,
IntentRef: strings.TrimSpace(intent.Ref),
OperationRef: strings.TrimSpace(cardStep.OperationRef),
})
if err != nil {
return err
@@ -271,20 +276,22 @@ func (s *Service) submitCardFundingTransfers(ctx context.Context, payment *model
exec.ChainTransferRef = strings.TrimSpace(fundResp.GetTransfer().GetTransferRef())
fundStep.TransferRef = exec.ChainTransferRef
}
setExecutionStepStatus(fundStep, executionStepStatusSubmitted)
setExecutionStepStatus(fundStep, model.OperationStateWaiting)
updateExecutionPlanTotalNetworkFee(plan)
if feeRequired {
feeResp, err := chainClient.SubmitTransfer(ctx, &chainv1.SubmitTransferRequest{
IntentRef: intent.Ref,
OperationRef: feeStep.OperationRef,
IdempotencyKey: payment.IdempotencyKey + ":card:fee",
OrganizationRef: payment.OrganizationRef.Hex(),
SourceWalletRef: sourceWalletRef,
Destination: &chainv1.TransferDestination{
Destination: &chainv1.TransferDestination_ManagedWalletRef{ManagedWalletRef: feeWalletRef},
},
Amount: cloneProtoMoney(feeAmountProto),
Metadata: cloneMetadata(payment.Metadata),
ClientReference: payment.PaymentRef,
Amount: cloneProtoMoney(feeAmountProto),
Metadata: cloneMetadata(payment.Metadata),
PaymentRef: payment.PaymentRef,
})
if err != nil {
return err
@@ -293,7 +300,7 @@ func (s *Service) submitCardFundingTransfers(ctx context.Context, payment *model
exec.FeeTransferRef = strings.TrimSpace(feeResp.GetTransfer().GetTransferRef())
feeStep.TransferRef = exec.FeeTransferRef
}
setExecutionStepStatus(feeStep, executionStepStatusSubmitted)
setExecutionStepStatus(feeStep, model.OperationStateWaiting)
s.logger.Info("card fee transfer submitted", zap.String("payment_ref", payment.PaymentRef), zap.String("transfer_ref", exec.FeeTransferRef))
}

69
api/payments/orchestrator/internal/service/orchestrator/card_payout_submit.go
View File

@@ -4,6 +4,7 @@ import (
"context"
"strings"
"github.com/google/uuid"
"github.com/shopspring/decimal"
"github.com/tech/sendico/payments/orchestrator/storage/model"
"github.com/tech/sendico/pkg/merrors"
@@ -87,6 +88,7 @@ func (s *Service) submitCardPayout(ctx context.Context, payment *model.Payment)
if token := strings.TrimSpace(card.Token); token != "" {
req := &mntxv1.CardTokenPayoutRequest{
PayoutId: payoutID,
IdempotencyKey: payment.IdempotencyKey,
CustomerId: customerID,
CustomerFirstName: customerFirstName,
CustomerMiddleName: customerMiddleName,
@@ -113,6 +115,7 @@ func (s *Service) submitCardPayout(ctx context.Context, payment *model.Payment)
} else if pan := strings.TrimSpace(card.Pan); pan != "" {
req := &mntxv1.CardPayoutRequest{
PayoutId: payoutID,
IdempotencyKey: payment.IdempotencyKey,
CustomerId: customerID,
CustomerFirstName: customerFirstName,
CustomerMiddleName: customerMiddleName,
@@ -166,7 +169,7 @@ func (s *Service) submitCardPayout(ctx context.Context, payment *model.Payment)
if exec.CardPayoutRef != "" {
step.TransferRef = exec.CardPayoutRef
}
setExecutionStepStatus(step, executionStepStatusSubmitted)
setExecutionStepStatus(step, model.OperationStateWaiting)
updateExecutionPlanTotalNetworkFee(plan)
}
@@ -225,21 +228,36 @@ func updateCardPayoutPlanSteps(payment *model.Payment, payout *mntxv1.CardPayout
}
execStep := plan.Steps[idx]
if execStep == nil {
execStep = &model.ExecutionStep{Code: planStepID(planStep, idx), Description: describePlanStep(planStep)}
execStep = &model.ExecutionStep{
Code: planStepID(planStep, idx),
Description: describePlanStep(planStep),
OperationRef: uuid.New().String(),
State: model.OperationStateCreated,
}
plan.Steps[idx] = execStep
}
if execStep.TransferRef == "" {
execStep.TransferRef = payoutID
}
switch payout.GetStatus() {
case mntxv1.PayoutStatus_PAYOUT_STATUS_PROCESSED:
setExecutionStepStatus(execStep, executionStepStatusConfirmed)
case mntxv1.PayoutStatus_PAYOUT_STATUS_CREATED:
setExecutionStepStatus(execStep, model.OperationStateCreated)
case mntxv1.PayoutStatus_PAYOUT_STATUS_WAITING:
setExecutionStepStatus(execStep, model.OperationStateWaiting)
case mntxv1.PayoutStatus_PAYOUT_STATUS_SUCCESS:
setExecutionStepStatus(execStep, model.OperationStateSuccess)
case mntxv1.PayoutStatus_PAYOUT_STATUS_FAILED:
setExecutionStepStatus(execStep, executionStepStatusFailed)
case mntxv1.PayoutStatus_PAYOUT_STATUS_PENDING:
setExecutionStepStatus(execStep, executionStepStatusSubmitted)
setExecutionStepStatus(execStep, model.OperationStateFailed)
case mntxv1.PayoutStatus_PAYOUT_STATUS_CANCELLED:
setExecutionStepStatus(execStep, model.OperationStateCancelled)
default:
setExecutionStepStatus(execStep, executionStepStatusPlanned)
setExecutionStepStatus(execStep, model.OperationStatePlanned)
}
updated = true
}
@@ -271,27 +289,46 @@ func applyCardPayoutUpdate(payment *model.Payment, payout *mntxv1.CardPayoutStat
}
}
switch payout.GetStatus() {
case mntxv1.PayoutStatus_PAYOUT_STATUS_PROCESSED:
setExecutionStepStatus(step, executionStepStatusConfirmed)
case mntxv1.PayoutStatus_PAYOUT_STATUS_CREATED:
setExecutionStepStatus(step, model.OperationStatePlanned)
case mntxv1.PayoutStatus_PAYOUT_STATUS_WAITING:
setExecutionStepStatus(step, model.OperationStateWaiting)
case mntxv1.PayoutStatus_PAYOUT_STATUS_SUCCESS:
setExecutionStepStatus(step, model.OperationStateSuccess)
case mntxv1.PayoutStatus_PAYOUT_STATUS_FAILED:
setExecutionStepStatus(step, executionStepStatusFailed)
case mntxv1.PayoutStatus_PAYOUT_STATUS_PENDING:
setExecutionStepStatus(step, executionStepStatusSubmitted)
setExecutionStepStatus(step, model.OperationStateFailed)
case mntxv1.PayoutStatus_PAYOUT_STATUS_CANCELLED:
setExecutionStepStatus(step, model.OperationStateCancelled)
default:
setExecutionStepStatus(step, executionStepStatusPlanned)
setExecutionStepStatus(step, model.OperationStatePlanned)
}
}
payment.State = mapMntxStatusToState(payout.GetStatus())
switch payout.GetStatus() {
case mntxv1.PayoutStatus_PAYOUT_STATUS_PROCESSED:
case mntxv1.PayoutStatus_PAYOUT_STATUS_SUCCESS:
payment.FailureCode = model.PaymentFailureCodeUnspecified
payment.FailureReason = ""
case mntxv1.PayoutStatus_PAYOUT_STATUS_FAILED:
payment.FailureCode = model.PaymentFailureCodePolicy
payment.FailureReason = strings.TrimSpace(payout.GetProviderMessage())
case mntxv1.PayoutStatus_PAYOUT_STATUS_CANCELLED:
payment.FailureCode = model.PaymentFailureCodePolicy
payment.FailureReason = "payout cancelled"
default:
// leave as-is for pending/unspecified
// CREATED / WAITING — keep as is
}
}

5
api/payments/orchestrator/internal/service/orchestrator/card_payout_test.go
View File

@@ -89,6 +89,7 @@ func TestSubmitCardFundingTransfers_PlansTopUpAndFunding(t *testing.T) {
IdempotencyKey: "pay-1",
OrganizationBoundBase: mo.OrganizationBoundBase{OrganizationRef: bson.NewObjectID()},
Intent: model.PaymentIntent{
Ref: "ref-1",
Kind: model.PaymentKindPayout,
Source: model.PaymentEndpoint{
Type: model.EndpointTypeManagedWallet,
@@ -232,7 +233,7 @@ func TestSubmitCardPayout_UsesSettlementAmount(t *testing.T) {
return &mntxv1.CardPayoutResponse{
Payout: &mntxv1.CardPayoutState{
PayoutId: "payout-1",
Status: mntxv1.PayoutStatus_PAYOUT_STATUS_PENDING,
Status: mntxv1.PayoutStatus_PAYOUT_STATUS_WAITING,
},
}, nil
},
@@ -251,6 +252,7 @@ func TestSubmitCardPayout_UsesSettlementAmount(t *testing.T) {
IdempotencyKey: "pay-2",
OrganizationBoundBase: mo.OrganizationBoundBase{OrganizationRef: bson.NewObjectID()},
Intent: model.PaymentIntent{
Ref: "ref-2",
Kind: model.PaymentKindPayout,
Source: model.PaymentEndpoint{
Type: model.EndpointTypeManagedWallet,
@@ -340,6 +342,7 @@ func TestSubmitCardFundingTransfers_RequiresFeeWalletRef(t *testing.T) {
IdempotencyKey: "pay-3",
OrganizationBoundBase: mo.OrganizationBoundBase{OrganizationRef: bson.NewObjectID()},
Intent: model.PaymentIntent{
Ref: "ref-3",
Kind: model.PaymentKindPayout,
Source: model.PaymentEndpoint{
Type: model.EndpointTypeManagedWallet,

4
api/payments/orchestrator/internal/service/orchestrator/convert.go
View File

@@ -23,6 +23,7 @@ func intentFromProto(src *orchestratorv1.PaymentIntent) model.PaymentIntent {
return model.PaymentIntent{}
}
intent := model.PaymentIntent{
Ref: src.GetRef(),
Kind: modelKindFromProto(src.GetKind()),
Source: endpointFromProto(src.GetSource()),
Destination: endpointFromProto(src.GetDestination()),
@@ -163,6 +164,7 @@ func toProtoPayment(src *model.Payment) *orchestratorv1.Payment {
func protoIntentFromModel(src model.PaymentIntent) *orchestratorv1.PaymentIntent {
intent := &orchestratorv1.PaymentIntent{
Ref: src.Ref,
Kind: protoKindFromModel(src.Kind),
Source: protoEndpointFromModel(src.Source),
Destination: protoEndpointFromModel(src.Destination),
@@ -315,6 +317,7 @@ func protoExecutionStepFromModel(src *model.ExecutionStep) *orchestratorv1.Execu
DestinationRef: src.DestinationRef,
TransferRef: src.TransferRef,
Metadata: cloneMetadata(src.Metadata),
OperationRef: src.OperationRef,
}
}
@@ -346,7 +349,6 @@ func protoPaymentStepFromModel(src *model.PaymentStep) *orchestratorv1.PaymentSt
GatewayId: strings.TrimSpace(src.GatewayID),
Action: protoRailOperationFromModel(src.Action),
Amount: protoMoney(src.Amount),
Ref: strings.TrimSpace(src.Ref),
StepId: strings.TrimSpace(src.StepID),
InstanceId: strings.TrimSpace(src.InstanceID),
DependsOn: cloneStringList(src.DependsOn),

58
api/payments/orchestrator/internal/service/orchestrator/execution_plan.go
View File

@@ -13,13 +13,6 @@ const (
executionStepRoleSource = "source"
executionStepRoleConsumer = "consumer"
executionStepStatusPlanned = "planned"
executionStepStatusSubmitted = "submitted"
executionStepStatusConfirmed = "confirmed"
executionStepStatusFailed = "failed"
executionStepStatusCancelled = "cancelled"
executionStepStatusSkipped = "skipped"
)
func setExecutionStepRole(step *model.ExecutionStep, role string) {
@@ -27,9 +20,9 @@ func setExecutionStepRole(step *model.ExecutionStep, role string) {
setExecutionStepMetadata(step, executionStepMetadataRole, role)
}
func setExecutionStepStatus(step *model.ExecutionStep, status string) {
status = strings.ToLower(strings.TrimSpace(status))
setExecutionStepMetadata(step, executionStepMetadataStatus, status)
func setExecutionStepStatus(step *model.ExecutionStep, state model.OperationState) {
step.State = state
setExecutionStepMetadata(step, executionStepMetadataStatus, string(state))
}
func executionStepRole(step *model.ExecutionStep) string {
@@ -45,17 +38,6 @@ func executionStepRole(step *model.ExecutionStep) string {
return executionStepRoleSource
}
func executionStepStatus(step *model.ExecutionStep) string {
if step == nil {
return ""
}
status := strings.TrimSpace(step.Metadata[executionStepMetadataStatus])
if status == "" {
return executionStepStatusPlanned
}
return strings.ToLower(status)
}
func isSourceExecutionStep(step *model.ExecutionStep) bool {
return executionStepRole(step) == executionStepRoleSource
}
@@ -69,12 +51,11 @@ func sourceStepsConfirmed(plan *model.ExecutionPlan) bool {
if step == nil || !isSourceExecutionStep(step) {
continue
}
status := executionStepStatus(step)
if status == executionStepStatusSkipped {
if step.State == model.OperationStateSkipped {
continue
}
hasSource = true
if status != executionStepStatusConfirmed {
if step.State != model.OperationStateSuccess {
return false
}
}
@@ -131,20 +112,29 @@ func updateExecutionStepFromTransfer(plan *model.ExecutionPlan, event *chainv1.T
return nil
}
func executionStepStatusFromTransferStatus(status chainv1.TransferStatus) string {
func executionStepStatusFromTransferStatus(status chainv1.TransferStatus) model.OperationState {
switch status {
case chainv1.TransferStatus_TRANSFER_CONFIRMED:
return executionStepStatusConfirmed
case chainv1.TransferStatus_TRANSFER_CREATED:
return model.OperationStatePlanned
case chainv1.TransferStatus_TRANSFER_PROCESSING:
return model.OperationStateProcessing
case chainv1.TransferStatus_TRANSFER_WAITING:
return model.OperationStateWaiting
case chainv1.TransferStatus_TRANSFER_SUCCESS:
return model.OperationStateSuccess
case chainv1.TransferStatus_TRANSFER_FAILED:
return executionStepStatusFailed
return model.OperationStateFailed
case chainv1.TransferStatus_TRANSFER_CANCELLED:
return executionStepStatusCancelled
case chainv1.TransferStatus_TRANSFER_SIGNING,
chainv1.TransferStatus_TRANSFER_PENDING,
chainv1.TransferStatus_TRANSFER_SUBMITTED:
return executionStepStatusSubmitted
return model.OperationStateCancelled
default:
return ""
return model.OperationStatePlanned
}
}

220
api/payments/orchestrator/internal/service/orchestrator/gateway_execution_consumer.go
View File

@@ -9,15 +9,18 @@ import (
cons "github.com/tech/sendico/pkg/messaging/consumer"
paymentgateway "github.com/tech/sendico/pkg/messaging/notifications/paymentgateway"
np "github.com/tech/sendico/pkg/messaging/notifications/processor"
"github.com/tech/sendico/pkg/mlogger"
"github.com/tech/sendico/pkg/model"
"github.com/tech/sendico/pkg/mservice"
"github.com/tech/sendico/pkg/payments/rail"
"go.uber.org/zap"
)
func (s *Service) startGatewayConsumers() {
if s == nil || s.gatewayBroker == nil {
s.logger.Warn("Missing broker. Gateway feedback consumer has NOT started")
return
}
s.logger.Info("Gateway feedback consumer started")
processor := paymentgateway.NewPaymentGatewayExecutionProcessor(s.logger, s.onGatewayExecution)
s.consumeGatewayProcessor(processor)
}
@@ -36,117 +39,198 @@ func (s *Service) consumeGatewayProcessor(processor np.EnvelopeProcessor) {
}()
}
func executionPlanSucceeded(plan *paymodel.ExecutionPlan) bool {
for _, s := range plan.Steps {
if !s.IsTerminal() {
return false
}
if s.State != paymodel.OperationStateSuccess {
return false
}
}
return true
}
func executionPlanFailed(plan *paymodel.ExecutionPlan) bool {
hasFailed := false
for _, s := range plan.Steps {
if !s.IsTerminal() {
return false
}
if s.State == paymodel.OperationStateFailed {
hasFailed = true
}
}
return hasFailed
}
func (s *Service) onGatewayExecution(ctx context.Context, exec *model.PaymentGatewayExecution) error {
if exec == nil {
return merrors.InvalidArgument("payment gateway execution is nil", "execution")
}
paymentRef := strings.TrimSpace(exec.PaymentIntentID)
paymentRef := strings.TrimSpace(exec.PaymentRef)
if paymentRef == "" {
return merrors.InvalidArgument("payment_intent_id is required", "payment_intent_id")
}
if s.storage == nil || s.storage.Payments() == nil {
return errStorageUnavailable
return merrors.InvalidArgument("payment_ref is required", "payment_ref")
}
store := s.storage.Payments()
payment, err := store.GetByPaymentRef(ctx, paymentRef)
if err != nil {
return err
}
// --- metadata
if payment.Metadata == nil {
payment.Metadata = map[string]string{}
}
if exec.RequestID != "" {
payment.Metadata["gateway_request_id"] = exec.RequestID
}
if exec.QuoteRef != "" {
payment.Metadata["gateway_quote_ref"] = exec.QuoteRef
}
if exec.ExecutedMoney != nil {
payment.Metadata["gateway_executed_amount"] = exec.ExecutedMoney.Amount
payment.Metadata["gateway_executed_currency"] = exec.ExecutedMoney.Currency
}
payment.Metadata["gateway_confirmation_status"] = string(exec.Status)
payment.Metadata["gateway_operation_result"] = string(exec.Status)
payment.Metadata["gateway_operation_ref"] = exec.OperationRef
payment.Metadata["gateway_request_idempotency"] = exec.IdempotencyKey
updatedPlan := updateExecutionStepsFromGatewayExecution(payment, exec)
switch exec.Status {
case model.ConfirmationStatusConfirmed, model.ConfirmationStatusClarified:
if payment.PaymentPlan != nil && updatedPlan && payment.ExecutionPlan != nil && !executionPlanComplete(payment.ExecutionPlan) {
return s.resumePaymentPlan(ctx, store, payment)
}
payment.State = paymodel.PaymentStateSettled
payment.FailureCode = paymodel.PaymentFailureCodeUnspecified
payment.FailureReason = ""
case model.ConfirmationStatusRejected:
payment.State = paymodel.PaymentStateFailed
payment.FailureCode = paymodel.PaymentFailureCodePolicy
payment.FailureReason = "gateway_rejected"
case model.ConfirmationStatusTimeout:
payment.State = paymodel.PaymentStateFailed
payment.FailureCode = paymodel.PaymentFailureCodePolicy
payment.FailureReason = "confirmation_timeout"
default:
s.logger.Warn("Unhandled gateway confirmation status", zap.String("status", string(exec.Status)), zap.String("payment_ref", paymentRef))
// --- update exactly ONE step
updated := updateExecutionStepsFromGatewayExecution(s.logger, payment, exec)
if !updated {
s.logger.Warn("No execution step matched gateway result",
zap.String("payment_ref", paymentRef),
zap.String("operation_ref", exec.OperationRef),
zap.String("idempotency", exec.IdempotencyKey),
)
}
if err := store.Update(ctx, payment); err != nil {
return err
}
s.logger.Info("Payment gateway execution applied", zap.String("payment_ref", paymentRef), zap.String("status", string(exec.Status)), zap.String("service", string(mservice.PaymentGateway)))
// reload unified state
payment, err = store.GetByPaymentRef(ctx, paymentRef)
if err != nil {
return err
}
// --- if plan can continue — continue
if payment.ExecutionPlan != nil && !executionPlanComplete(payment.ExecutionPlan) {
return s.resumePaymentPlan(ctx, store, payment)
}
// --- plan is terminal: decide payment fate by aggregation
if payment.ExecutionPlan != nil && executionPlanComplete(payment.ExecutionPlan) {
switch {
case executionPlanSucceeded(payment.ExecutionPlan):
payment.State = paymodel.PaymentStateSettled
case executionPlanFailed(payment.ExecutionPlan):
payment.State = paymodel.PaymentStateFailed
payment.FailureReason = "execution_plan_failed"
}
return store.Update(ctx, payment)
}
return nil
}
func updateExecutionStepsFromGatewayExecution(payment *paymodel.Payment, exec *model.PaymentGatewayExecution) bool {
if payment == nil || exec == nil || payment.PaymentPlan == nil {
func updateExecutionStepsFromGatewayExecution(
logger mlogger.Logger,
payment *paymodel.Payment,
exec *model.PaymentGatewayExecution,
) bool {
if payment == nil || payment.PaymentPlan == nil || exec == nil {
logger.Warn("updateExecutionSteps: invalid input",
zap.String("payment_ref", payment.PaymentRef),
)
return false
}
requestID := strings.TrimSpace(exec.RequestID)
if requestID == "" {
operationRef := strings.TrimSpace(exec.OperationRef)
if operationRef == "" {
logger.Warn("updateExecutionSteps: empty operation_ref from gateway",
zap.String("payment_ref", payment.PaymentRef),
)
return false
}
execPlan := ensureExecutionPlanForPlan(payment, payment.PaymentPlan)
if execPlan == nil {
logger.Error("updateExecutionSteps: execution plan missing",
zap.String("payment_ref", payment.PaymentRef),
)
return false
}
status := executionStepStatusFromGatewayStatus(exec.Status)
if status == "" {
logger.Warn("updateExecutionSteps: unknown gateway status",
zap.String("payment_ref", payment.PaymentRef),
zap.String("gateway_status", string(exec.Status)),
)
return false
}
updated := false
for idx, planStep := range payment.PaymentPlan.Steps {
if planStep == nil {
continue
}
if idx >= len(execPlan.Steps) {
continue
}
execStep := execPlan.Steps[idx]
logger.Debug("updateExecutionSteps: matching by operation_ref",
zap.String("payment_ref", payment.PaymentRef),
zap.String("operation_ref", operationRef),
zap.String("mapped_status", string(status)),
)
for idx, execStep := range execPlan.Steps {
if execStep == nil {
execStep = &paymodel.ExecutionStep{Code: planStepID(planStep, idx), Description: describePlanStep(planStep)}
execPlan.Steps[idx] = execStep
}
if strings.EqualFold(strings.TrimSpace(execStep.TransferRef), requestID) {
setExecutionStepStatus(execStep, status)
updated = true
continue
}
if execStep.TransferRef == "" && planStep.Rail == paymodel.RailProviderSettlement {
if planStep.Action == paymodel.RailOperationObserveConfirm || planStep.Action == paymodel.RailOperationSend {
execStep.TransferRef = requestID
setExecutionStepStatus(execStep, status)
updated = true
if strings.EqualFold(strings.TrimSpace(execStep.OperationRef), operationRef) {
logger.Debug("updateExecutionSteps: matched execution step",
zap.String("payment_ref", payment.PaymentRef),
zap.Int("step_index", idx),
zap.String("step_code", execStep.Code),
zap.String("prev_state", string(execStep.State)),
)
// update transfer ref if not set yet
if execStep.TransferRef == "" && exec.TransferRef != "" {
execStep.TransferRef = strings.TrimSpace(exec.TransferRef)
}
setExecutionStepStatus(execStep, status)
logger.Debug("updateExecutionSteps: step state updated",
zap.String("payment_ref", payment.PaymentRef),
zap.Int("step_index", idx),
zap.String("step_code", execStep.Code),
zap.String("new_state", string(execStep.State)),
)
return true
}
}
return updated
logger.Error("updateExecutionSteps: no execution step found for operation_ref",
zap.String("payment_ref", payment.PaymentRef),
zap.String("operation_ref", operationRef),
)
return false
}
func executionStepStatusFromGatewayStatus(status model.ConfirmationStatus) string {
func executionStepStatusFromGatewayStatus(status rail.OperationResult) paymodel.OperationState {
switch status {
case model.ConfirmationStatusConfirmed, model.ConfirmationStatusClarified:
return executionStepStatusConfirmed
case model.ConfirmationStatusRejected, model.ConfirmationStatusTimeout:
return executionStepStatusFailed
case rail.OperationResultSuccess:
return paymodel.OperationStateSuccess
case rail.OperationResultFailed:
return paymodel.OperationStateFailed
case rail.OperationResultCancelled:
return paymodel.OperationStateCancelled
default:
return ""
return paymodel.OperationStateFailed
}
}

60
api/payments/orchestrator/internal/service/orchestrator/gateway_execution_consumer_test.go
View File

@@ -7,63 +7,89 @@ import (
paymodel "github.com/tech/sendico/payments/orchestrator/storage/model"
mloggerfactory "github.com/tech/sendico/pkg/mlogger/factory"
"github.com/tech/sendico/pkg/model"
"github.com/tech/sendico/pkg/payments/rail"
)
func TestGatewayExecutionConfirmedUpdatesPayment(t *testing.T) {
func TestGatewayExecutionSuccessUpdatesMetadataOnly(t *testing.T) {
logger := mloggerfactory.NewLogger(false)
store := newHelperPaymentStore()
payment := &paymodel.Payment{PaymentRef: "pi-1", State: paymodel.PaymentStateSubmitted}
payment := &paymodel.Payment{
PaymentRef: "pi-1",
State: paymodel.PaymentStateSubmitted,
}
if err := store.Create(context.Background(), payment); err != nil {
t.Fatalf("failed to seed payment: %v", err)
}
svc := &Service{
logger: logger,
storage: stubRepo{payments: store},
}
exec := &model.PaymentGatewayExecution{
PaymentIntentID: "pi-1",
Status: model.ConfirmationStatusConfirmed,
RequestID: "req-1",
QuoteRef: "quote-1",
PaymentRef: "pi-1",
Status: rail.OperationResultSuccess,
IdempotencyKey: "idem-1",
OperationRef: "oper-1",
}
if err := svc.onGatewayExecution(context.Background(), exec); err != nil {
t.Fatalf("onGatewayExecution error: %v", err)
}
updated, _ := store.GetByPaymentRef(context.Background(), "pi-1")
if updated.State != paymodel.PaymentStateSettled {
t.Fatalf("expected payment settled, got %s", updated.State)
// Should not be Settled without execution plan
if updated.State != paymodel.PaymentStateSubmitted {
t.Fatalf("expected payment to remain submitted, got %s", updated.State)
}
if updated.Metadata["gateway_request_id"] != "req-1" {
t.Fatalf("expected gateway_request_id metadata")
if updated.Metadata["gateway_request_idempotency"] != "idem-1" {
t.Fatalf("expected gateway_request_idempotency metadata")
}
if updated.Metadata["gateway_confirmation_status"] != string(model.ConfirmationStatusConfirmed) {
t.Fatalf("expected gateway_confirmation_status metadata")
if updated.Metadata["gateway_operation_result"] != string(rail.OperationResultSuccess) {
t.Fatalf("expected gateway_operation_result metadata")
}
}
func TestGatewayExecutionRejectedFailsPayment(t *testing.T) {
logger := mloggerfactory.NewLogger(false)
store := newHelperPaymentStore()
payment := &paymodel.Payment{PaymentRef: "pi-2", State: paymodel.PaymentStateSubmitted}
payment := &paymodel.Payment{
PaymentRef: "pi-2", State: paymodel.PaymentStateSubmitted, IdempotencyKey: "idem-1",
ExecutionPlan: &paymodel.ExecutionPlan{
Steps: []*paymodel.ExecutionStep{
{OperationRef: "s1", State: paymodel.OperationStatePlanned, TransferRef: "trn-1"}}}}
if err := store.Create(context.Background(), payment); err != nil {
t.Fatalf("failed to seed payment: %v", err)
}
svc := &Service{
logger: logger,
storage: stubRepo{payments: store},
}
exec := &model.PaymentGatewayExecution{
PaymentIntentID: "pi-2",
Status: model.ConfirmationStatusRejected,
PaymentRef: "pi-2",
TransferRef: "trn-1",
Status: rail.OperationResultFailed,
}
if err := svc.onGatewayExecution(context.Background(), exec); err != nil {
t.Fatalf("onGatewayExecution error: %v", err)
}
updated, _ := store.GetByPaymentRef(context.Background(), "pi-2")
if updated.State != paymodel.PaymentStateFailed {
t.Fatalf("expected payment failed, got %s", updated.State)
}
if updated.FailureReason != "gateway_rejected" {
t.Fatalf("expected failure reason gateway_rejected, got %q", updated.FailureReason)
if updated.FailureReason != "execution_plan_failed" {
t.Fatalf("expected failure reason execution_plan_failed, got %q", updated.FailureReason)
}
}

2
api/payments/orchestrator/internal/service/orchestrator/handlers_commands.go
View File

@@ -9,6 +9,7 @@ import (
"strings"
"time"
"github.com/google/uuid"
"github.com/tech/sendico/payments/orchestrator/storage"
"github.com/tech/sendico/payments/orchestrator/storage/model"
"github.com/tech/sendico/pkg/api/routers/gsresponse"
@@ -881,6 +882,7 @@ func (h *initiateConversionCommand) Execute(ctx context.Context, req *orchestrat
}
intentProto := &orchestratorv1.PaymentIntent{
Ref: uuid.New().String(),
Kind: orchestratorv1.PaymentKind_PAYMENT_KIND_FX_CONVERSION,
Source: req.GetSource(),
Destination: req.GetDestination(),

68
api/payments/orchestrator/internal/service/orchestrator/handlers_events.go
View File

@@ -85,7 +85,7 @@ func (h *paymentEventHandler) processTransferUpdate(ctx context.Context, req *or
return gsresponse.Auto[orchestratorv1.ProcessTransferUpdateResponse](h.logger, mservice.PaymentOrchestrator, err)
}
return gsresponse.Success(&orchestratorv1.ProcessTransferUpdateResponse{Payment: toProtoPayment(payment)})
case chainv1.TransferStatus_TRANSFER_CONFIRMED:
case chainv1.TransferStatus_TRANSFER_SUCCESS:
if h.resumePlan != nil {
if err := h.resumePlan(ctx, store, payment); err != nil {
return gsresponse.Auto[orchestratorv1.ProcessTransferUpdateResponse](h.logger, mservice.PaymentOrchestrator, err)
@@ -97,9 +97,7 @@ func (h *paymentEventHandler) processTransferUpdate(ctx context.Context, req *or
}
}
return gsresponse.Success(&orchestratorv1.ProcessTransferUpdateResponse{Payment: toProtoPayment(payment)})
case chainv1.TransferStatus_TRANSFER_SIGNING,
chainv1.TransferStatus_TRANSFER_PENDING,
chainv1.TransferStatus_TRANSFER_SUBMITTED:
case chainv1.TransferStatus_TRANSFER_WAITING:
if payment.State != model.PaymentStateFailed && payment.State != model.PaymentStateCancelled && payment.State != model.PaymentStateSettled {
payment.State = model.PaymentStateSubmitted
}
@@ -133,7 +131,7 @@ func (h *paymentEventHandler) processTransferUpdate(ctx context.Context, req *or
payment.State = model.PaymentStateCancelled
payment.FailureCode = model.PaymentFailureCodePolicy
payment.FailureReason = reason
case chainv1.TransferStatus_TRANSFER_CONFIRMED:
case chainv1.TransferStatus_TRANSFER_SUCCESS:
if payment.State != model.PaymentStateFailed && payment.State != model.PaymentStateCancelled && payment.State != model.PaymentStateSettled {
if cardPayoutDependenciesConfirmed(payment.PaymentPlan, payment.ExecutionPlan) {
if payment.Execution.CardPayoutRef != "" {
@@ -155,9 +153,7 @@ func (h *paymentEventHandler) processTransferUpdate(ctx context.Context, req *or
payment.State = model.PaymentStateSubmitted
}
}
case chainv1.TransferStatus_TRANSFER_SIGNING,
chainv1.TransferStatus_TRANSFER_PENDING,
chainv1.TransferStatus_TRANSFER_SUBMITTED:
case chainv1.TransferStatus_TRANSFER_WAITING:
if payment.State != model.PaymentStateFailed && payment.State != model.PaymentStateCancelled && payment.State != model.PaymentStateSettled {
payment.State = model.PaymentStateSubmitted
}
@@ -251,18 +247,72 @@ func (h *paymentEventHandler) processCardPayoutUpdate(ctx context.Context, req *
}
applyCardPayoutUpdate(payment, payout)
switch payout.GetStatus() {
case mntxv1.PayoutStatus_PAYOUT_STATUS_PROCESSED:
case mntxv1.PayoutStatus_PAYOUT_STATUS_SUCCESS:
h.logger.Info("card payout success received",
zap.String("payment_ref", payment.PaymentRef),
zap.String("payout_ref", payout.GetPayoutId()),
zap.String("payment_state_before", string(payment.State)),
zap.Bool("has_plan", payment.PaymentPlan != nil && len(payment.PaymentPlan.Steps) > 0),
zap.Bool("resume_plan_present", h.resumePlan != nil),
)
if h.resumePlan != nil && payment.PaymentPlan != nil && len(payment.PaymentPlan.Steps) > 0 {
if err := h.resumePlan(ctx, store, payment); err != nil {
h.logger.Error("resumePlan failed after payout success",
zap.String("payment_ref", payment.PaymentRef),
zap.String("payout_ref", payout.GetPayoutId()),
zap.Error(err),
)
return gsresponse.Auto[orchestratorv1.ProcessCardPayoutUpdateResponse](h.logger, mservice.PaymentOrchestrator, err)
}
h.logger.Info("resumePlan executed after payout success",
zap.String("payment_ref", payment.PaymentRef),
zap.String("payout_ref", payout.GetPayoutId()),
)
} else {
h.logger.Warn("payout success but plan cannot be resumed",
zap.String("payment_ref", payment.PaymentRef),
zap.String("payout_ref", payout.GetPayoutId()),
zap.Bool("resume_plan_present", h.resumePlan != nil),
zap.Bool("has_plan", payment.PaymentPlan != nil && len(payment.PaymentPlan.Steps) > 0),
)
}
case mntxv1.PayoutStatus_PAYOUT_STATUS_FAILED:
h.logger.Warn("card payout failed",
zap.String("payment_ref", payment.PaymentRef),
zap.String("payout_ref", payout.GetPayoutId()),
zap.String("provider_message", payout.GetProviderMessage()),
)
payment.State = model.PaymentStateFailed
payment.FailureCode = model.PaymentFailureCodePolicy
payment.FailureReason = strings.TrimSpace(payout.GetProviderMessage())
if h.releaseHold != nil && payment.PaymentPlan != nil && len(payment.PaymentPlan.Steps) > 0 {
h.logger.Info("releasing hold after payout failure",
zap.String("payment_ref", payment.PaymentRef),
zap.String("payout_ref", payout.GetPayoutId()),
)
if err := h.releaseHold(ctx, store, payment); err != nil {
h.logger.Error("releaseHold failed after payout failure",
zap.String("payment_ref", payment.PaymentRef),
zap.String("payout_ref", payout.GetPayoutId()),
zap.Error(err),
)
return gsresponse.Auto[orchestratorv1.ProcessCardPayoutUpdateResponse](h.logger, mservice.PaymentOrchestrator, err)
}
} else {
h.logger.Warn("payout failed but hold cannot be released",
zap.String("payment_ref", payment.PaymentRef),
zap.String("payout_ref", payout.GetPayoutId()),
zap.Bool("release_hold_present", h.releaseHold != nil),
zap.Bool("has_plan", payment.PaymentPlan != nil && len(payment.PaymentPlan.Steps) > 0),
)
}
}

45
api/payments/orchestrator/internal/service/orchestrator/helpers.go
View File

@@ -7,7 +7,6 @@ import (
"github.com/shopspring/decimal"
oracleclient "github.com/tech/sendico/fx/oracle/client"
"github.com/tech/sendico/pkg/merrors"
"github.com/tech/sendico/pkg/payments/rail"
chainv1 "github.com/tech/sendico/pkg/proto/gateway/chain/v1"
ledgerv1 "github.com/tech/sendico/pkg/proto/ledger/v1"
oraclev1 "github.com/tech/sendico/pkg/proto/oracle/v1"
@@ -416,50 +415,6 @@ func quoteExpiry(now time.Time, feeQuote *feesv1.PrecomputeFeesResponse, fxQuote
return expiry
}
func feeBreakdownFromQuote(quote *orchestratorv1.PaymentQuote) []rail.FeeBreakdown {
if quote == nil {
return nil
}
lines := quote.GetFeeLines()
breakdown := make([]rail.FeeBreakdown, 0, len(lines)+1)
for _, line := range lines {
if line == nil {
continue
}
amount := moneyFromProto(line.GetMoney())
if amount == nil {
continue
}
code := strings.TrimSpace(line.GetMeta()["fee_code"])
if code == "" {
code = strings.TrimSpace(line.GetMeta()["fee_rule_id"])
}
if code == "" {
code = line.GetLineType().String()
}
desc := strings.TrimSpace(line.GetMeta()["description"])
breakdown = append(breakdown, rail.FeeBreakdown{
FeeCode: code,
Amount: amount,
Description: desc,
})
}
if quote.GetNetworkFee() != nil && quote.GetNetworkFee().GetNetworkFee() != nil {
networkAmount := moneyFromProto(quote.GetNetworkFee().GetNetworkFee())
if networkAmount != nil {
breakdown = append(breakdown, rail.FeeBreakdown{
FeeCode: "network_fee",
Amount: networkAmount,
Description: strings.TrimSpace(quote.GetNetworkFee().GetEstimationContext()),
})
}
}
if len(breakdown) == 0 {
return nil
}
return breakdown
}
func assignLedgerAccounts(lines []*feesv1.DerivedPostingLine, account string) []*feesv1.DerivedPostingLine {
if account == "" || len(lines) == 0 {
return lines

16
api/payments/orchestrator/internal/service/orchestrator/internal_helpers.go
View File

@@ -110,12 +110,22 @@ func fxIntentForQuote(intent *orchestratorv1.PaymentIntent) *orchestratorv1.FXIn
func mapMntxStatusToState(status mntxv1.PayoutStatus) model.PaymentState {
switch status {
case mntxv1.PayoutStatus_PAYOUT_STATUS_PROCESSED:
case mntxv1.PayoutStatus_PAYOUT_STATUS_CREATED:
return model.PaymentStateFundsReserved
case mntxv1.PayoutStatus_PAYOUT_STATUS_WAITING:
return model.PaymentStateSubmitted
case mntxv1.PayoutStatus_PAYOUT_STATUS_SUCCESS:
return model.PaymentStateSettled
case mntxv1.PayoutStatus_PAYOUT_STATUS_FAILED:
return model.PaymentStateFailed
case mntxv1.PayoutStatus_PAYOUT_STATUS_PENDING:
return model.PaymentStateSubmitted
case mntxv1.PayoutStatus_PAYOUT_STATUS_CANCELLED:
return model.PaymentStateCancelled
default:
return model.PaymentStateUnspecified
}

6
api/payments/orchestrator/internal/service/orchestrator/internal_helpers_test.go
View File

@@ -10,6 +10,7 @@ import (
func TestShouldEstimateNetworkFeeSkipsCard(t *testing.T) {
intent := &orchestratorv1.PaymentIntent{
Ref: "ref-1",
Kind: orchestratorv1.PaymentKind_PAYMENT_KIND_PAYOUT,
Destination: &orchestratorv1.PaymentEndpoint{
Endpoint: &orchestratorv1.PaymentEndpoint_Card{
@@ -24,6 +25,7 @@ func TestShouldEstimateNetworkFeeSkipsCard(t *testing.T) {
func TestShouldEstimateNetworkFeeManagedWallet(t *testing.T) {
intent := &orchestratorv1.PaymentIntent{
Ref: "ref-1",
Destination: &orchestratorv1.PaymentEndpoint{
Endpoint: &orchestratorv1.PaymentEndpoint_ManagedWallet{
ManagedWallet: &orchestratorv1.ManagedWalletEndpoint{ManagedWalletRef: "mw"},
@@ -36,13 +38,13 @@ func TestShouldEstimateNetworkFeeManagedWallet(t *testing.T) {
}
func TestMapMntxStatusToState(t *testing.T) {
if mapMntxStatusToState(mntxv1.PayoutStatus_PAYOUT_STATUS_PROCESSED) != model.PaymentStateSettled {
if mapMntxStatusToState(mntxv1.PayoutStatus_PAYOUT_STATUS_SUCCESS) != model.PaymentStateSettled {
t.Fatalf("processed should map to settled")
}
if mapMntxStatusToState(mntxv1.PayoutStatus_PAYOUT_STATUS_FAILED) != model.PaymentStateFailed {
t.Fatalf("failed should map to failed")
}
if mapMntxStatusToState(mntxv1.PayoutStatus_PAYOUT_STATUS_PENDING) != model.PaymentStateSubmitted {
if mapMntxStatusToState(mntxv1.PayoutStatus_PAYOUT_STATUS_WAITING) != model.PaymentStateSubmitted {
t.Fatalf("pending should map to submitted")
}
if mapMntxStatusToState(mntxv1.PayoutStatus_PAYOUT_STATUS_UNSPECIFIED) != model.PaymentStateUnspecified {

16
api/payments/orchestrator/internal/service/orchestrator/payment_executor.go
View File

@@ -207,23 +207,31 @@ func applyTransferStatus(event *chainv1.TransferStatusChangedEvent, payment *mod
reason = strings.TrimSpace(transfer.GetFailureReason())
}
switch transfer.GetStatus() {
case chainv1.TransferStatus_TRANSFER_CONFIRMED:
case chainv1.TransferStatus_TRANSFER_SUCCESS:
payment.State = model.PaymentStateSettled
payment.FailureCode = model.PaymentFailureCodeUnspecified
payment.FailureReason = ""
case chainv1.TransferStatus_TRANSFER_FAILED:
payment.State = model.PaymentStateFailed
payment.FailureCode = model.PaymentFailureCodeChain
payment.FailureReason = reason
case chainv1.TransferStatus_TRANSFER_CANCELLED:
payment.State = model.PaymentStateCancelled
payment.FailureCode = model.PaymentFailureCodePolicy
payment.FailureReason = reason
case chainv1.TransferStatus_TRANSFER_SIGNING,
chainv1.TransferStatus_TRANSFER_PENDING,
chainv1.TransferStatus_TRANSFER_SUBMITTED:
case chainv1.TransferStatus_TRANSFER_WAITING:
payment.State = model.PaymentStateSubmitted
case chainv1.TransferStatus_TRANSFER_CREATED,
chainv1.TransferStatus_TRANSFER_PROCESSING:
// do nothing, retain previous state
default:
// retain previous state
}
}

13
api/payments/orchestrator/internal/service/orchestrator/payment_plan_card.go
View File

@@ -7,12 +7,12 @@ import (
"github.com/shopspring/decimal"
"github.com/tech/sendico/payments/orchestrator/storage/model"
"github.com/tech/sendico/pkg/merrors"
pmodel "github.com/tech/sendico/pkg/model"
"github.com/tech/sendico/pkg/model/account_role"
moneyv1 "github.com/tech/sendico/pkg/proto/common/money/v1"
mntxv1 "github.com/tech/sendico/pkg/proto/gateway/mntx/v1"
)
func (p *paymentExecutor) submitCardPayoutPlan(ctx context.Context, payment *model.Payment, amount *moneyv1.Money, fromRole, toRole *pmodel.AccountRole) (string, error) {
func (p *paymentExecutor) submitCardPayoutPlan(ctx context.Context, payment *model.Payment, operationRef string, amount *moneyv1.Money, fromRole, toRole *account_role.AccountRole) (string, error) {
if payment == nil {
return "", merrors.InvalidArgument("payment is required")
}
@@ -42,13 +42,13 @@ func (p *paymentExecutor) submitCardPayoutPlan(ctx context.Context, payment *mod
if meta == nil {
meta = map[string]string{}
}
meta[pmodel.MetadataKeyFromRole] = strings.TrimSpace(string(mergeAccountRole(fromRole)))
meta[account_role.MetadataKeyFromRole] = strings.TrimSpace(string(mergeAccountRole(fromRole)))
}
if strings.TrimSpace(string(mergeAccountRole(toRole))) != "" {
if meta == nil {
meta = map[string]string{}
}
meta[pmodel.MetadataKeyToRole] = strings.TrimSpace(string(mergeAccountRole(toRole)))
meta[account_role.MetadataKeyToRole] = strings.TrimSpace(string(mergeAccountRole(toRole)))
}
customer := intent.Customer
customerID := ""
@@ -112,6 +112,7 @@ func (p *paymentExecutor) submitCardPayoutPlan(ctx context.Context, payment *mod
CardHolder: holder,
MaskedPan: strings.TrimSpace(card.MaskedPan),
Metadata: meta,
OperationRef: operationRef,
}
resp, err := p.deps.mntx.client.CreateCardTokenPayout(ctx, req)
if err != nil {
@@ -159,11 +160,11 @@ func (p *paymentExecutor) submitCardPayoutPlan(ctx context.Context, payment *mod
return exec.CardPayoutRef, nil
}
func mergeAccountRole(role *pmodel.AccountRole) pmodel.AccountRole {
func mergeAccountRole(role *account_role.AccountRole) account_role.AccountRole {
if role == nil {
return ""
}
return pmodel.AccountRole(strings.TrimSpace(string(*role)))
return account_role.AccountRole(strings.TrimSpace(string(*role)))
}
func (p *paymentExecutor) resolveCardRoute(intent model.PaymentIntent) (CardGatewayRoute, error) {

12
api/payments/orchestrator/internal/service/orchestrator/payment_plan_chain.go
View File

@@ -5,13 +5,13 @@ import (
"github.com/tech/sendico/payments/orchestrator/storage/model"
"github.com/tech/sendico/pkg/merrors"
pmodel "github.com/tech/sendico/pkg/model"
"github.com/tech/sendico/pkg/model/account_role"
"github.com/tech/sendico/pkg/payments/rail"
paymenttypes "github.com/tech/sendico/pkg/payments/types"
orchestratorv1 "github.com/tech/sendico/pkg/proto/payments/orchestrator/v1"
)
func (p *paymentExecutor) buildCryptoTransferRequest(payment *model.Payment, amount *paymenttypes.Money, action model.RailOperation, idempotencyKey string, quote *orchestratorv1.PaymentQuote, fromRole, toRole *pmodel.AccountRole) (rail.TransferRequest, error) {
func (p *paymentExecutor) buildCryptoTransferRequest(payment *model.Payment, amount *paymenttypes.Money, action model.RailOperation, idempotencyKey, operationRef string, quote *orchestratorv1.PaymentQuote, fromRole, toRole *account_role.AccountRole) (rail.TransferRequest, error) {
if payment == nil {
return rail.TransferRequest{}, merrors.InvalidArgument("chain: payment is required")
}
@@ -26,8 +26,15 @@ func (p *paymentExecutor) buildCryptoTransferRequest(payment *model.Payment, amo
if err != nil {
return rail.TransferRequest{}, err
}
paymentRef := strings.TrimSpace(payment.PaymentRef)
if paymentRef == "" {
return rail.TransferRequest{}, merrors.InvalidArgument("chain: payment reference is required")
}
req := rail.TransferRequest{
IntentRef: strings.TrimSpace(payment.Intent.Ref),
OperationRef: strings.TrimSpace(operationRef),
OrganizationRef: payment.OrganizationRef.Hex(),
PaymentRef: strings.TrimSpace(payment.PaymentRef),
FromAccountID: strings.TrimSpace(source.ManagedWalletRef),
ToAccountID: strings.TrimSpace(destRef),
Currency: strings.TrimSpace(amount.GetCurrency()),
@@ -35,7 +42,6 @@ func (p *paymentExecutor) buildCryptoTransferRequest(payment *model.Payment, amo
Amount: strings.TrimSpace(amount.GetAmount()),
IdempotencyKey: strings.TrimSpace(idempotencyKey),
Metadata: cloneMetadata(payment.Metadata),
ClientReference: payment.PaymentRef,
DestinationMemo: memo,
}
if fromRole != nil {

341
api/payments/orchestrator/internal/service/orchestrator/payment_plan_executor.go
View File

@@ -2,107 +2,300 @@ package orchestrator
import (
"context"
"strings"
"errors"
"github.com/tech/sendico/payments/orchestrator/storage"
"github.com/tech/sendico/payments/orchestrator/storage/model"
"github.com/tech/sendico/pkg/merrors"
orchestratorv1 "github.com/tech/sendico/pkg/proto/payments/orchestrator/v1"
"go.uber.org/zap"
)
func (p *paymentExecutor) executePaymentPlan(ctx context.Context, store storage.PaymentsStore, payment *model.Payment, quote *orchestratorv1.PaymentQuote) error {
if store == nil {
return errStorageUnavailable
}
if payment == nil {
return merrors.InvalidArgument("payment plan: payment is required")
}
plan := payment.PaymentPlan
func analyzeExecutionPlan(plan *model.ExecutionPlan) (done bool, failed bool, rootErr error) {
if plan == nil || len(plan.Steps) == 0 {
return merrors.InvalidArgument("payment plan: steps are required")
return true, false, nil
}
done = true
for _, s := range plan.Steps {
if s == nil {
continue
}
if s.State == model.OperationStateFailed {
failed = true
if rootErr == nil && s.Error != "" {
rootErr = errors.New(s.Error)
}
}
if !isStepFinal(s) { // created/waiting/processing
done = false
}
}
return done, failed, rootErr
}
func buildStepIndex(plan *model.PaymentPlan) map[string]int {
m := make(map[string]int, len(plan.Steps))
for i, s := range plan.Steps {
if s == nil {
continue
}
m[s.StepID] = i
}
return m
}
func isPlanComplete(payment *model.Payment) bool {
if (payment.State == model.PaymentStateCancelled) ||
(payment.State == model.PaymentStateSettled) ||
(payment.State == model.PaymentStateFailed) {
return true
}
return false
}
func isStepFinal(step *model.ExecutionStep) bool {
if (step.State == model.OperationStateFailed) || (step.State == model.OperationStateSuccess) || (step.State == model.OperationStateCancelled) {
return true
}
return false
}
func stepCodeIsDependent(code string, previousSteps []string) bool {
for _, ps := range previousSteps {
if ps == code {
return true
}
}
return false
}
func stepIsIndependent(
step *model.ExecutionStep,
plan *model.PaymentPlan,
execSteps map[string]*model.ExecutionStep,
) bool {
for _, s := range plan.Steps {
if s.StepID != step.Code {
continue
}
// Do not process step if it is already in a final state
if isStepFinal(step) {
return false
}
// If the step has no dependencies, it is independent
if len(s.DependsOn) == 0 {
return true
}
// All dependent steps must be successfully completed
for _, dep := range s.DependsOn {
depStep := execSteps[dep]
if depStep == nil || depStep.State != model.OperationStateSuccess {
return false
}
}
return true
}
return false
}
func planStep(execStep *model.ExecutionStep, plan *model.PaymentPlan) *model.PaymentStep {
if (execStep == nil) || (plan == nil) {
return nil
}
for _, step := range plan.Steps {
if step != nil {
if step.StepID == execStep.Code {
return step
}
}
}
return nil
}
func (p *paymentExecutor) pickIndependentSteps(
ctx context.Context,
l *zap.Logger,
store storage.PaymentsStore,
waiting []*model.ExecutionStep,
payment *model.Payment,
quote *orchestratorv1.PaymentQuote,
) error {
logger := l.With(zap.Int("waiting_steps", len(waiting)))
logger.Debug("Selecting independent steps for execution")
execSteps := executionStepsByCode(payment.ExecutionPlan)
planSteps := planStepsByID(payment.PaymentPlan)
execQuote := executionQuote(payment, quote)
charges := ledgerChargesFromFeeLines(execQuote.GetFeeLines())
stepIdx := buildStepIndex(payment.PaymentPlan)
order, _, err := planExecutionOrder(plan)
if err != nil {
return p.failPayment(ctx, store, payment, model.PaymentFailureCodePolicy, strings.TrimSpace(err.Error()), err)
}
execPlan := ensureExecutionPlanForPlan(payment, plan)
execSteps := executionStepsByCode(execPlan)
planSteps := planStepsByID(plan)
asyncSubmitted := false
for _, idx := range order {
step := plan.Steps[idx]
if step == nil {
return p.failPayment(ctx, store, payment, model.PaymentFailureCodePolicy, "payment plan: step is required", merrors.InvalidArgument("payment plan: step is required"))
}
stepID := planStepID(step, idx)
execStep := execSteps[stepID]
for _, execStep := range waiting {
if execStep == nil {
execStep = &model.ExecutionStep{Code: stepID}
execSteps[stepID] = execStep
}
if step.Action == model.RailOperationRelease {
setExecutionStepStatus(execStep, executionStepStatusSkipped)
continue
}
status := executionStepStatus(execStep)
switch status {
case executionStepStatusConfirmed, executionStepStatusSkipped:
continue
case executionStepStatusFailed:
payment.State = model.PaymentStateFailed
payment.FailureCode = failureCodeForStep(step)
return p.persistPayment(ctx, store, payment)
case executionStepStatusCancelled:
payment.State = model.PaymentStateCancelled
payment.FailureCode = model.PaymentFailureCodePolicy
return p.persistPayment(ctx, store, payment)
case executionStepStatusSubmitted:
asyncSubmitted = true
continue
}
ready, blocked, err := stepDependenciesReady(step, execSteps, planSteps, false)
lg := logger.With(
zap.String("step_code", execStep.Code),
zap.String("step_state", string(execStep.State)),
)
planStep := planSteps[execStep.Code]
if planStep == nil {
lg.Warn("Plan step not found")
continue
}
ready, waitingDep, blocked, err :=
stepDependenciesReady(planStep, execSteps, planSteps, true)
if err != nil {
return p.failPayment(ctx, store, payment, model.PaymentFailureCodePolicy, strings.TrimSpace(err.Error()), err)
lg.Warn("Dependency evaluation failed", zap.Error(err))
continue
}
if blocked {
if step.CommitPolicy == model.CommitPolicyAfterFailure && commitAfterDependenciesSucceeded(step, execSteps) {
setExecutionStepStatus(execStep, executionStepStatusSkipped)
continue
}
payment.State = model.PaymentStateFailed
payment.FailureCode = failureCodeForStep(step)
return p.persistPayment(ctx, store, payment)
lg.Debug("Step permanently blocked by dependency failure")
continue
}
if waitingDep {
lg.Debug("Step waiting for dependencies")
continue
}
if !ready {
continue
}
async, err := p.executePlanStep(ctx, payment, step, execStep, execQuote, charges, idx)
lg.Debug("Executing independent step")
idx := stepIdx[execStep.Code]
async, err := p.executePlanStep(
ctx,
payment,
planStep,
execStep,
quote,
charges,
idx,
)
if err != nil {
return p.failPayment(ctx, store, payment, failureCodeForStep(step), strings.TrimSpace(err.Error()), err)
}
if async {
asyncSubmitted = true
lg.Warn("Step execution failed", zap.Error(err), zap.Bool("async", async))
return err
}
}
if asyncSubmitted && !executionPlanComplete(execPlan) {
if blockStepConfirmed(plan, execPlan) {
payment.State = model.PaymentStateFundsReserved
} else {
payment.State = model.PaymentStateSubmitted
}
return p.persistPayment(ctx, store, payment)
}
payment.State = model.PaymentStateSettled
payment.FailureCode = model.PaymentFailureCodeUnspecified
payment.FailureReason = ""
return p.persistPayment(ctx, store, payment)
return nil
}
func (p *paymentExecutor) pickWaitingSteps(
ctx context.Context,
l *zap.Logger,
store storage.PaymentsStore,
payment *model.Payment,
quote *orchestratorv1.PaymentQuote,
) error {
if payment == nil || payment.ExecutionPlan == nil {
l.Debug("No execution plan")
return nil
}
logger := l.With(zap.Int("total_steps", len(payment.ExecutionPlan.Steps)))
logger.Debug("Collecting waiting steps")
waitingSteps := make([]*model.ExecutionStep, 0, len(payment.ExecutionPlan.Steps))
for _, step := range payment.ExecutionPlan.Steps {
if step == nil {
continue
}
if step.State != model.OperationStatePlanned {
continue
}
waitingSteps = append(waitingSteps, step)
}
if len(waitingSteps) == 0 {
logger.Debug("No waiting steps to process")
return nil
}
return p.pickIndependentSteps(ctx, logger, store, waitingSteps, payment, quote)
}
func (p *paymentExecutor) executePaymentPlan(
ctx context.Context,
store storage.PaymentsStore,
payment *model.Payment,
quote *orchestratorv1.PaymentQuote,
) error {
if payment == nil {
return merrors.InvalidArgument("plan must be provided")
}
logger := p.logger.With(zap.String("payment_ref", payment.PaymentRef))
logger.Debug("Starting plan execution")
if isPlanComplete(payment) {
logger.Debug("Plan already completed")
return nil
}
if payment.State != model.PaymentStateSubmitted &&
payment.State != model.PaymentStateFundsReserved {
payment.State = model.PaymentStateSubmitted
if err := store.Update(ctx, payment); err != nil {
return err
}
}
if payment.ExecutionPlan == nil {
logger.Debug("Initializing execution plan from payment plan")
payment.ExecutionPlan = ensureExecutionPlanForPlan(payment, payment.PaymentPlan)
if err := store.Update(ctx, payment); err != nil {
return err
}
}
// Execute steps
if err := p.pickWaitingSteps(ctx, logger, store, payment, quote); err != nil {
logger.Warn("Step execution returned infrastructure error", zap.Error(err))
return err
}
if err := store.Update(ctx, payment); err != nil {
return err
}
done, failed, rootErr := analyzeExecutionPlan(payment.ExecutionPlan)
if !done {
return nil
}
if failed {
payment.State = model.PaymentStateFailed
} else {
payment.State = model.PaymentStateSettled
}
if err := store.Update(ctx, payment); err != nil {
logger.Warn("Failed to update final payment state", zap.Error(err))
return err
}
if failed && rootErr != nil {
return rootErr
}
return nil
}

90
api/payments/orchestrator/internal/service/orchestrator/payment_plan_executor_test.go
View File

@@ -9,7 +9,7 @@ import (
ledgerclient "github.com/tech/sendico/ledger/client"
"github.com/tech/sendico/payments/orchestrator/storage/model"
mo "github.com/tech/sendico/pkg/model"
pmodel "github.com/tech/sendico/pkg/model"
"github.com/tech/sendico/pkg/model/account_role"
"github.com/tech/sendico/pkg/payments/rail"
paymenttypes "github.com/tech/sendico/pkg/payments/types"
connectorv1 "github.com/tech/sendico/pkg/proto/connector/v1"
@@ -34,7 +34,7 @@ func TestExecutePaymentPlan_SourceBeforeDestination(t *testing.T) {
sendFn: func(ctx context.Context, req rail.TransferRequest) (rail.RailResult, error) {
ref := transferRefs[sendCalls]
sendCalls++
return rail.RailResult{ReferenceID: ref, Status: rail.TransferStatusPending}, nil
return rail.RailResult{ReferenceID: ref, Status: rail.TransferStatusWaiting}, nil
},
}
@@ -105,6 +105,7 @@ func TestExecutePaymentPlan_SourceBeforeDestination(t *testing.T) {
OrganizationRef: bson.NewObjectID(),
},
Intent: model.PaymentIntent{
Ref: "ref-1",
Kind: model.PaymentKindPayout,
Source: model.PaymentEndpoint{
Type: model.EndpointTypeManagedWallet,
@@ -141,9 +142,9 @@ func TestExecutePaymentPlan_SourceBeforeDestination(t *testing.T) {
{StepID: "crypto_send", Rail: model.RailCrypto, Action: model.RailOperationSend, Amount: &paymenttypes.Money{Currency: "USDT", Amount: "100"}},
{StepID: "crypto_fee", Rail: model.RailCrypto, Action: model.RailOperationFee, DependsOn: []string{"crypto_send"}, Amount: &paymenttypes.Money{Currency: "USDT", Amount: "5"}},
{StepID: "crypto_observe", Rail: model.RailProviderSettlement, Action: model.RailOperationObserveConfirm, DependsOn: []string{"crypto_send"}},
{StepID: "ledger_credit", Rail: model.RailLedger, Action: model.RailOperationMove, DependsOn: []string{"crypto_observe"}, Amount: &paymenttypes.Money{Currency: "USDT", Amount: "95"}, FromRole: rolePtr(pmodel.AccountRolePending), ToRole: rolePtr(pmodel.AccountRoleOperating)},
{StepID: "ledger_credit", Rail: model.RailLedger, Action: model.RailOperationMove, DependsOn: []string{"crypto_observe"}, Amount: &paymenttypes.Money{Currency: "USDT", Amount: "95"}, FromRole: rolePtr(account_role.AccountRolePending), ToRole: rolePtr(account_role.AccountRoleOperating)},
{StepID: "card_payout", Rail: model.RailCardPayout, Action: model.RailOperationSend, DependsOn: []string{"ledger_credit"}, Amount: &paymenttypes.Money{Currency: "USDT", Amount: "95"}},
{StepID: "ledger_debit", Rail: model.RailLedger, Action: model.RailOperationMove, DependsOn: []string{"card_payout"}, CommitPolicy: model.CommitPolicyAfterSuccess, CommitAfter: []string{"card_payout"}, Amount: &paymenttypes.Money{Currency: "USDT", Amount: "95"}, FromRole: rolePtr(pmodel.AccountRoleOperating), ToRole: rolePtr(pmodel.AccountRoleTransit)},
{StepID: "ledger_debit", Rail: model.RailLedger, Action: model.RailOperationMove, DependsOn: []string{"card_payout"}, CommitPolicy: model.CommitPolicyAfterSuccess, CommitAfter: []string{"card_payout"}, Amount: &paymenttypes.Money{Currency: "USDT", Amount: "95"}, FromRole: rolePtr(account_role.AccountRoleOperating), ToRole: rolePtr(account_role.AccountRoleTransit)},
},
},
}
@@ -154,64 +155,66 @@ func TestExecutePaymentPlan_SourceBeforeDestination(t *testing.T) {
t.Fatalf("executePaymentPlan error: %v", err)
}
if sendCalls != 2 {
t.Fatalf("expected 2 rail sends, got %d", sendCalls)
if payment.Execution == nil || payment.Execution.ChainTransferRef == "" {
t.Fatalf("expected chain transfer ref set")
}
if moveCalls != 0 {
t.Fatalf("unexpected ledger move calls: %d", moveCalls)
}
if payoutCalls != 0 {
t.Fatalf("expected no payout before source confirmation, got %d", payoutCalls)
}
if payment.State != model.PaymentStateSubmitted {
t.Fatalf("expected submitted state, got %s", payment.State)
}
if payment.Execution == nil || payment.Execution.ChainTransferRef == "" || payment.Execution.FeeTransferRef == "" {
t.Fatalf("expected chain and fee transfer refs set")
}
if payment.ExecutionPlan == nil || len(payment.ExecutionPlan.Steps) != 6 {
t.Fatalf("expected execution plan with 6 steps")
}
if executionStepStatus(payment.ExecutionPlan.Steps[0]) != executionStepStatusSubmitted {
t.Fatalf("expected send step submitted")
}
if executionStepStatus(payment.ExecutionPlan.Steps[1]) != executionStepStatusSubmitted {
t.Fatalf("expected fee step submitted")
}
if executionStepStatus(payment.ExecutionPlan.Steps[2]) != executionStepStatusSubmitted {
t.Fatalf("expected observe step submitted")
if payment.Execution.FeeTransferRef != "" {
t.Fatalf("fee must NOT be executed before send success")
}
setExecutionStepStatus(payment.ExecutionPlan.Steps[0], executionStepStatusConfirmed)
setExecutionStepStatus(payment.ExecutionPlan.Steps[1], executionStepStatusConfirmed)
setExecutionStepStatus(payment.ExecutionPlan.Steps[2], executionStepStatusConfirmed)
steps := executionStepsByCode(payment.ExecutionPlan)
if steps["crypto_send"].State != model.OperationStateWaiting {
t.Fatalf("send must be waiting")
}
if steps["crypto_fee"].State != model.OperationStatePlanned {
t.Fatalf("fee must NOT start before send success")
}
if steps["crypto_observe"].State != model.OperationStatePlanned {
t.Fatalf("observe must NOT start before send success")
}
// ---- имитируем подтверждение сети по crypto_send ----
setExecutionStepStatus(steps["crypto_send"], model.OperationStateSuccess)
if err := executor.executePaymentPlan(ctx, store, payment, &orchestratorv1.PaymentQuote{}); err != nil {
t.Fatalf("executePaymentPlan resume error: %v", err)
}
// Теперь должны стартовать fee и observe
if steps["crypto_fee"].State != model.OperationStateWaiting {
t.Fatalf("fee must start after send success")
}
if steps["crypto_observe"].State != model.OperationStateWaiting {
t.Fatalf("observe must start after send success")
}
// Имитируем подтверждение observe (это unlock ledger_credit)
setExecutionStepStatus(steps["crypto_observe"], model.OperationStateSuccess)
if err := executor.executePaymentPlan(ctx, store, payment, &orchestratorv1.PaymentQuote{}); err != nil {
t.Fatalf("executePaymentPlan resume after observe error: %v", err)
}
if moveCalls != 1 {
t.Fatalf("expected one ledger move after source confirmation, got %d", moveCalls)
t.Fatalf("expected one ledger move after observe confirmation, got %d", moveCalls)
}
if payoutCalls != 1 {
t.Fatalf("expected card payout submitted, got %d", payoutCalls)
}
if payment.Execution == nil || payment.Execution.CardPayoutRef == "" {
t.Fatalf("expected card payout ref set")
}
steps := executionStepsByCode(payment.ExecutionPlan)
cardStep := steps["card_payout"]
if cardStep == nil {
t.Fatalf("expected card payout step in execution plan")
}
setExecutionStepStatus(cardStep, executionStepStatusConfirmed)
// Mock card payout confirmation
cardStep := executionStepsByCode(payment.ExecutionPlan)["card_payout"]
setExecutionStepStatus(cardStep, model.OperationStateSuccess)
if err := executor.executePaymentPlan(ctx, store, payment, &orchestratorv1.PaymentQuote{}); err != nil {
t.Fatalf("executePaymentPlan finalize error: %v", err)
}
if moveCalls != 2 {
t.Fatalf("expected two ledger moves after payout confirmation, got %d", moveCalls)
}
}
func TestExecutePaymentPlan_RejectsLegacyLedgerOperations(t *testing.T) {
@@ -242,6 +245,7 @@ func TestExecutePaymentPlan_RejectsLegacyLedgerOperations(t *testing.T) {
OrganizationRef: bson.NewObjectID(),
},
Intent: model.PaymentIntent{
Ref: "ref-legacy-1",
Kind: model.PaymentKindPayout,
Source: model.PaymentEndpoint{
Type: model.EndpointTypeManagedWallet,
@@ -269,7 +273,7 @@ func TestExecutePaymentPlan_RejectsLegacyLedgerOperations(t *testing.T) {
if err == nil {
t.Fatal("expected legacy ledger operation error")
}
if !strings.Contains(err.Error(), "unsupported legacy ledger operation") {
if !strings.Contains(err.Error(), "unsupported action") {
t.Fatalf("unexpected error: %v", err)
}
}

62
api/payments/orchestrator/internal/service/orchestrator/payment_plan_helpers.go
View File

@@ -4,8 +4,9 @@ import (
"fmt"
"strings"
"github.com/google/uuid"
"github.com/tech/sendico/payments/orchestrator/storage/model"
pmodel "github.com/tech/sendico/pkg/model"
"github.com/tech/sendico/pkg/model/account_role"
orchestratorv1 "github.com/tech/sendico/pkg/proto/payments/orchestrator/v1"
)
@@ -26,40 +27,32 @@ func executionQuote(payment *model.Payment, quote *orchestratorv1.PaymentQuote)
return &orchestratorv1.PaymentQuote{}
}
func ensureExecutionPlanForPlan(payment *model.Payment, plan *model.PaymentPlan) *model.ExecutionPlan {
if payment == nil || plan == nil {
return nil
func ensureExecutionPlanForPlan(
payment *model.Payment,
plan *model.PaymentPlan,
) *model.ExecutionPlan {
if payment.ExecutionPlan != nil {
return payment.ExecutionPlan
}
execPlan := payment.ExecutionPlan
if execPlan == nil {
execPlan = &model.ExecutionPlan{}
payment.ExecutionPlan = execPlan
exec := &model.ExecutionPlan{
Steps: make([]*model.ExecutionStep, 0, len(plan.Steps)),
}
existing := map[string]*model.ExecutionStep{}
for _, step := range execPlan.Steps {
if step == nil || strings.TrimSpace(step.Code) == "" {
for _, step := range plan.Steps {
if step == nil {
continue
}
existing[strings.TrimSpace(step.Code)] = step
exec.Steps = append(exec.Steps, &model.ExecutionStep{
Code: step.StepID,
State: model.OperationStatePlanned,
OperationRef: uuid.New().String(),
})
}
steps := make([]*model.ExecutionStep, len(plan.Steps))
for idx, planStep := range plan.Steps {
code := planStepID(planStep, idx)
step := existing[code]
if step == nil {
step = &model.ExecutionStep{Code: code}
}
if step.Description == "" {
step.Description = describePlanStep(planStep)
}
step.Amount = cloneMoney(planStep.Amount)
if step.Metadata == nil || strings.TrimSpace(step.Metadata[executionStepMetadataStatus]) == "" {
setExecutionStepStatus(step, executionStepStatusPlanned)
}
steps[idx] = step
}
execPlan.Steps = steps
return execPlan
return exec
}
func executionPlanComplete(plan *model.ExecutionPlan) bool {
@@ -70,11 +63,10 @@ func executionPlanComplete(plan *model.ExecutionPlan) bool {
if step == nil {
continue
}
status := executionStepStatus(step)
if status == executionStepStatusSkipped {
if step.State == model.OperationStateSkipped {
continue
}
if status != executionStepStatusConfirmed {
if step.State != model.OperationStateSuccess {
return false
}
}
@@ -94,14 +86,14 @@ func blockStepConfirmed(plan *model.PaymentPlan, execPlan *model.ExecutionPlan)
if execStep == nil {
continue
}
if executionStepStatus(execStep) == executionStepStatusConfirmed {
if execStep.State == model.OperationStateSuccess {
return true
}
}
return false
}
func roleHintsForStep(plan *model.PaymentPlan, idx int) (*pmodel.AccountRole, *pmodel.AccountRole) {
func roleHintsForStep(plan *model.PaymentPlan, idx int) (*account_role.AccountRole, *account_role.AccountRole) {
if plan == nil || idx <= 0 {
return nil, nil
}

178
api/payments/orchestrator/internal/service/orchestrator/payment_plan_ledger.go
View File

@@ -8,7 +8,7 @@ import (
"github.com/tech/sendico/payments/orchestrator/storage/model"
"github.com/tech/sendico/pkg/ledgerconv"
"github.com/tech/sendico/pkg/merrors"
pmodel "github.com/tech/sendico/pkg/model"
"github.com/tech/sendico/pkg/model/account_role"
"github.com/tech/sendico/pkg/payments/rail"
moneyv1 "github.com/tech/sendico/pkg/proto/common/money/v1"
connectorv1 "github.com/tech/sendico/pkg/proto/connector/v1"
@@ -144,122 +144,6 @@ func (p *paymentExecutor) postLedgerMove(ctx context.Context, payment *model.Pay
return entryRef, nil
}
func (p *paymentExecutor) postLedgerBlock(ctx context.Context, payment *model.Payment, amount *moneyv1.Money, idempotencyKey string, idx int) (string, error) {
paymentRef := ""
if payment != nil {
paymentRef = strings.TrimSpace(payment.PaymentRef)
}
if p.deps.ledger.internal == nil {
p.logger.Error("Ledger client unavailable", zap.String("action", "block"), zap.String("payment_ref", paymentRef))
return "", merrors.Internal("ledger_client_unavailable")
}
if payment == nil {
return "", merrors.InvalidArgument("ledger: payment is required")
}
if payment.OrganizationRef == bson.NilObjectID {
return "", merrors.InvalidArgument("ledger: organization_ref is required")
}
if amount == nil || strings.TrimSpace(amount.GetAmount()) == "" || strings.TrimSpace(amount.GetCurrency()) == "" {
return "", merrors.InvalidArgument("ledger: amount is required")
}
sourceAccount, err := ledgerDebitAccountRef(payment)
if err != nil {
return "", err
}
blockAccount, err := p.resolveLedgerBlockAccount(ctx, payment, amount)
if err != nil {
return "", err
}
resp, err := p.deps.ledger.internal.TransferInternal(ctx, &ledgerv1.TransferRequest{
IdempotencyKey: strings.TrimSpace(idempotencyKey),
OrganizationRef: payment.OrganizationRef.Hex(),
FromLedgerAccountRef: strings.TrimSpace(sourceAccount),
ToLedgerAccountRef: strings.TrimSpace(blockAccount),
Money: cloneProtoMoney(amount),
Description: paymentDescription(payment),
Metadata: cloneMetadata(payment.Metadata),
})
if err != nil {
p.logger.Warn("Ledger block failed",
zap.String("payment_ref", paymentRef),
zap.Int("step_index", idx),
zap.String("from_account", strings.TrimSpace(sourceAccount)),
zap.String("to_account", strings.TrimSpace(blockAccount)),
zap.String("amount", strings.TrimSpace(amount.GetAmount())),
zap.String("currency", strings.TrimSpace(amount.GetCurrency())),
zap.Error(err))
return "", err
}
entryRef := strings.TrimSpace(resp.GetJournalEntryRef())
p.logger.Info("Ledger block posted",
zap.String("payment_ref", paymentRef),
zap.Int("step_index", idx),
zap.String("entry_ref", entryRef),
zap.String("from_account", strings.TrimSpace(sourceAccount)),
zap.String("to_account", strings.TrimSpace(blockAccount)),
zap.String("amount", strings.TrimSpace(amount.GetAmount())),
zap.String("currency", strings.TrimSpace(amount.GetCurrency())))
return entryRef, nil
}
func (p *paymentExecutor) postLedgerRelease(ctx context.Context, payment *model.Payment, amount *moneyv1.Money, idempotencyKey string, idx int) (string, error) {
paymentRef := ""
if payment != nil {
paymentRef = strings.TrimSpace(payment.PaymentRef)
}
if p.deps.ledger.internal == nil {
p.logger.Error("Ledger client unavailable", zap.String("action", "release"), zap.String("payment_ref", paymentRef))
return "", merrors.Internal("ledger_client_unavailable")
}
if payment == nil {
return "", merrors.InvalidArgument("ledger: payment is required")
}
if payment.OrganizationRef == bson.NilObjectID {
return "", merrors.InvalidArgument("ledger: organization_ref is required")
}
if amount == nil || strings.TrimSpace(amount.GetAmount()) == "" || strings.TrimSpace(amount.GetCurrency()) == "" {
return "", merrors.InvalidArgument("ledger: amount is required")
}
sourceAccount, err := ledgerDebitAccountRef(payment)
if err != nil {
return "", err
}
blockAccount, err := p.resolveLedgerBlockAccount(ctx, payment, amount)
if err != nil {
return "", err
}
resp, err := p.deps.ledger.internal.TransferInternal(ctx, &ledgerv1.TransferRequest{
IdempotencyKey: strings.TrimSpace(idempotencyKey),
OrganizationRef: payment.OrganizationRef.Hex(),
FromLedgerAccountRef: strings.TrimSpace(blockAccount),
ToLedgerAccountRef: strings.TrimSpace(sourceAccount),
Money: cloneProtoMoney(amount),
Description: paymentDescription(payment),
Metadata: cloneMetadata(payment.Metadata),
})
if err != nil {
p.logger.Warn("Ledger release failed",
zap.String("payment_ref", paymentRef),
zap.Int("step_index", idx),
zap.String("from_account", strings.TrimSpace(blockAccount)),
zap.String("to_account", strings.TrimSpace(sourceAccount)),
zap.String("amount", strings.TrimSpace(amount.GetAmount())),
zap.String("currency", strings.TrimSpace(amount.GetCurrency())),
zap.Error(err))
return "", err
}
entryRef := strings.TrimSpace(resp.GetJournalEntryRef())
p.logger.Info("Ledger release posted",
zap.String("payment_ref", paymentRef),
zap.Int("step_index", idx),
zap.String("entry_ref", entryRef),
zap.String("from_account", strings.TrimSpace(blockAccount)),
zap.String("to_account", strings.TrimSpace(sourceAccount)),
zap.String("amount", strings.TrimSpace(amount.GetAmount())),
zap.String("currency", strings.TrimSpace(amount.GetCurrency())))
return entryRef, nil
}
func (p *paymentExecutor) ledgerTxForAction(ctx context.Context, payment *model.Payment, amount *moneyv1.Money, charges []*ledgerv1.PostingLine, idempotencyKey string, idx int, action model.RailOperation, quote *orchestratorv1.PaymentQuote) (rail.LedgerTx, error) {
if payment == nil {
return rail.LedgerTx{}, merrors.InvalidArgument("ledger: payment is required")
@@ -426,7 +310,7 @@ func ledgerExternalReference(plan *model.ExecutionPlan, idx int) string {
return ""
}
func ledgerMoveRoles(step *model.PaymentStep) (pmodel.AccountRole, pmodel.AccountRole, error) {
func ledgerMoveRoles(step *model.PaymentStep) (account_role.AccountRole, account_role.AccountRole, error) {
if step == nil {
return "", "", merrors.InvalidArgument("ledger: step is required")
}
@@ -441,10 +325,10 @@ func ledgerMoveRoles(step *model.PaymentStep) (pmodel.AccountRole, pmodel.Accoun
if from == "" || to == "" || strings.EqualFold(from, to) {
return "", "", merrors.InvalidArgument("ledger: from_role and to_role must differ")
}
return pmodel.AccountRole(from), pmodel.AccountRole(to), nil
return account_role.AccountRole(from), account_role.AccountRole(to), nil
}
func ledgerRoleFromAccountRole(role pmodel.AccountRole) ledgerv1.AccountRole {
func ledgerRoleFromAccountRole(role account_role.AccountRole) ledgerv1.AccountRole {
if strings.TrimSpace(string(role)) == "" {
return ledgerv1.AccountRole_ACCOUNT_ROLE_UNSPECIFIED
}
@@ -454,7 +338,7 @@ func ledgerRoleFromAccountRole(role pmodel.AccountRole) ledgerv1.AccountRole {
return ledgerv1.AccountRole_ACCOUNT_ROLE_UNSPECIFIED
}
func (p *paymentExecutor) resolveAccount(ctx context.Context, orgRef bson.ObjectID, asset string, rail model.Rail, role pmodel.AccountRole) (string, error) {
func (p *paymentExecutor) resolveAccount(ctx context.Context, orgRef bson.ObjectID, asset string, rail model.Rail, role account_role.AccountRole) (string, error) {
switch rail {
case model.RailLedger:
return p.resolveLedgerAccountByRole(ctx, orgRef, asset, role)
@@ -463,7 +347,7 @@ func (p *paymentExecutor) resolveAccount(ctx context.Context, orgRef bson.Object
}
}
func (p *paymentExecutor) resolveLedgerAccountByRole(ctx context.Context, orgRef bson.ObjectID, asset string, role pmodel.AccountRole) (string, error) {
func (p *paymentExecutor) resolveLedgerAccountByRole(ctx context.Context, orgRef bson.ObjectID, asset string, role account_role.AccountRole) (string, error) {
if p == nil || p.deps == nil || p.deps.ledger.client == nil {
return "", merrors.Internal("ledger_client_unavailable")
}
@@ -590,27 +474,27 @@ func (p *paymentExecutor) resolveOrgOwnedLedgerAccount(ctx context.Context, paym
}
func connectorAccountIsSettlement(account *connectorv1.Account) bool {
return connectorAccountRole(account) == pmodel.AccountRoleSettlement
return connectorAccountRole(account) == account_role.AccountRoleSettlement
}
func connectorAccountRole(account *connectorv1.Account) pmodel.AccountRole {
func connectorAccountRole(account *connectorv1.Account) account_role.AccountRole {
if account == nil || account.GetProviderDetails() == nil {
return ""
}
details := account.GetProviderDetails().AsMap()
if value := strings.TrimSpace(fmt.Sprint(details["role"])); value != "" {
if role, ok := pmodel.Parse(value); ok {
if role, ok := account_role.Parse(value); ok {
return role
}
}
switch v := details["is_settlement"].(type) {
case bool:
if v {
return pmodel.AccountRoleSettlement
return account_role.AccountRoleSettlement
}
case string:
if strings.EqualFold(strings.TrimSpace(v), "true") {
return pmodel.AccountRoleSettlement
return account_role.AccountRoleSettlement
}
}
return ""
@@ -631,23 +515,6 @@ func setLedgerAccountAttributes(payment *model.Payment, accountRef string) {
}
}
func setLedgerBlockAccountAttributes(payment *model.Payment, accountRef string) {
if payment == nil || strings.TrimSpace(accountRef) == "" {
return
}
if payment.Intent.Attributes == nil {
payment.Intent.Attributes = map[string]string{}
}
if attributeLookup(payment.Intent.Attributes,
"ledger_block_account_ref",
"ledgerBlockAccountRef",
"ledger_hold_account_ref",
"ledgerHoldAccountRef",
) == "" {
payment.Intent.Attributes["ledger_block_account_ref"] = accountRef
}
}
func ledgerDebitAccount(payment *model.Payment) (string, string, error) {
if payment == nil {
return "", "", merrors.InvalidArgument("ledger: payment is required")
@@ -662,11 +529,6 @@ func ledgerDebitAccount(payment *model.Payment) (string, string, error) {
return "", "", merrors.InvalidArgument("ledger: source account is required")
}
func ledgerDebitAccountRef(payment *model.Payment) (string, error) {
account, _, err := ledgerDebitAccount(payment)
return account, err
}
func ledgerBlockAccount(payment *model.Payment) (string, error) {
if payment == nil {
return "", merrors.InvalidArgument("ledger: payment is required")
@@ -690,24 +552,6 @@ func ledgerBlockAccount(payment *model.Payment) (string, error) {
return "", merrors.InvalidArgument("ledger: block account is required")
}
func (p *paymentExecutor) resolveLedgerBlockAccount(ctx context.Context, payment *model.Payment, amount *moneyv1.Money) (string, error) {
if payment == nil {
return "", merrors.InvalidArgument("ledger: payment is required")
}
if amount == nil || strings.TrimSpace(amount.GetCurrency()) == "" {
return "", merrors.InvalidArgument("ledger: amount is required")
}
if ref, err := ledgerBlockAccount(payment); err == nil && strings.TrimSpace(ref) != "" {
return ref, nil
}
account, err := p.resolveOrgOwnedLedgerAccount(ctx, payment, amount)
if err != nil {
return "", err
}
setLedgerBlockAccountAttributes(payment, account)
return account, nil
}
func ledgerBlockAccountIfConfirmed(payment *model.Payment) string {
if payment == nil {
return ""

5
api/payments/orchestrator/internal/service/orchestrator/payment_plan_ledger_test.go
View File

@@ -6,7 +6,7 @@ import (
ledgerclient "github.com/tech/sendico/ledger/client"
"github.com/tech/sendico/payments/orchestrator/storage/model"
pmodel "github.com/tech/sendico/pkg/model"
"github.com/tech/sendico/pkg/model/account_role"
paymenttypes "github.com/tech/sendico/pkg/payments/types"
connectorv1 "github.com/tech/sendico/pkg/proto/connector/v1"
ledgerv1 "github.com/tech/sendico/pkg/proto/ledger/v1"
@@ -80,13 +80,14 @@ func TestLedgerAccountResolution_UsesRoleAccounts(t *testing.T) {
PaymentRef: "pay-1",
IdempotencyKey: "pay-1",
Intent: model.PaymentIntent{
Ref: "ref-1",
Kind: model.PaymentKindPayout,
},
PaymentPlan: &model.PaymentPlan{
ID: "pay-1",
IdempotencyKey: "pay-1",
Steps: []*model.PaymentStep{
{StepID: "ledger_move", Rail: model.RailLedger, Action: model.RailOperationMove, Amount: cloneMoney(amount), FromRole: rolePtr(pmodel.AccountRoleOperating), ToRole: rolePtr(pmodel.AccountRoleTransit)},
{StepID: "ledger_move", Rail: model.RailLedger, Action: model.RailOperationMove, Amount: cloneMoney(amount), FromRole: rolePtr(account_role.AccountRoleOperating), ToRole: rolePtr(account_role.AccountRoleTransit)},
},
},
}

172
api/payments/orchestrator/internal/service/orchestrator/payment_plan_order.go
View File

@@ -103,136 +103,143 @@ func planStepsByID(plan *model.PaymentPlan) map[string]*model.PaymentStep {
return result
}
func stepDependenciesReady(step *model.PaymentStep, execSteps map[string]*model.ExecutionStep, planSteps map[string]*model.PaymentStep, requireConfirmed bool) (bool, bool, error) {
func stepDependenciesReady(
step *model.PaymentStep,
execSteps map[string]*model.ExecutionStep,
planSteps map[string]*model.PaymentStep,
requireSuccess bool,
) (ready bool, waiting bool, blocked bool, err error) {
if step == nil {
return false, false, merrors.InvalidArgument("payment plan: step is required")
return false, false, false,
merrors.InvalidArgument("payment plan: step is required")
}
// ------------------------------------------------------------
// DependsOn — это ПРОСТО готовность, не успех
// ------------------------------------------------------------
for _, dep := range step.DependsOn {
key := strings.TrimSpace(dep)
if key == "" {
continue
}
execStep := execSteps[key]
if execStep == nil {
return false, false, merrors.InvalidArgument("payment plan: dependency missing")
// шага ещё не было → ждём
return false, true, false, nil
}
depStep := planSteps[key]
needsConfirm := requireConfirmed
if depStep != nil && depStep.Action == model.RailOperationObserveConfirm {
needsConfirm = true
if execStep.State == model.OperationStateFailed ||
execStep.State == model.OperationStateCancelled {
// зависимость умерла → этот шаг уже невозможен
return false, false, true, nil
}
status := executionStepStatus(execStep)
switch status {
case executionStepStatusFailed, executionStepStatusCancelled:
return false, true, nil
case executionStepStatusConfirmed, executionStepStatusSkipped:
continue
case executionStepStatusSubmitted:
if needsConfirm {
return false, false, nil
}
continue
default:
return false, false, nil
if !execStep.ReadyForNext() {
// шаг ещё в процессе → ждём
return false, true, false, nil
}
}
// Handle commit policies
// ------------------------------------------------------------
// Commit policies
// ------------------------------------------------------------
switch step.CommitPolicy {
case model.CommitPolicyImmediate, model.CommitPolicyUnspecified:
// Execute immediately once dependencies are satisfied
return true, false, nil
return true, false, false, nil
case model.CommitPolicyAfterSuccess:
// Wait for commitAfter dependencies to succeed (confirmed/skipped)
commitAfter := step.CommitAfter
if len(commitAfter) == 0 {
commitAfter = step.DependsOn
}
for _, dep := range commitAfter {
key := strings.TrimSpace(dep)
if key == "" {
continue
}
execStep := execSteps[key]
if execStep == nil {
return false, false, merrors.InvalidArgument("payment plan: commit dependency missing")
return false, true, false,
merrors.InvalidArgument("commit dependency missing")
}
status := executionStepStatus(execStep)
switch status {
case executionStepStatusFailed, executionStepStatusCancelled:
return false, true, nil
case executionStepStatusConfirmed, executionStepStatusSkipped:
continue
default:
return false, false, nil
if execStep.State == model.OperationStateFailed ||
execStep.State == model.OperationStateCancelled {
return false, false, true, nil
}
if !execStep.IsSuccess() {
return false, true, false, nil
}
}
return true, false, nil
return true, false, false, nil
case model.CommitPolicyAfterFailure:
// Wait for commitAfter dependencies to fail
commitAfter := step.CommitAfter
if len(commitAfter) == 0 {
commitAfter = step.DependsOn
}
for _, dep := range commitAfter {
key := strings.TrimSpace(dep)
if key == "" {
continue
}
execStep := execSteps[key]
if execStep == nil {
return false, false, merrors.InvalidArgument("payment plan: commit dependency missing")
return false, true, false,
merrors.InvalidArgument("commit dependency missing")
}
status := executionStepStatus(execStep)
switch status {
case executionStepStatusFailed:
// Dependency failed - this is what we're waiting for
if execStep.State == model.OperationStateFailed {
continue
case executionStepStatusCancelled:
// If cancelled, also block this step
return false, true, nil
case executionStepStatusConfirmed, executionStepStatusSkipped:
// Dependency succeeded - can't proceed with AFTER_FAILURE
return false, true, nil
default:
// Still waiting for failure
return false, false, nil
}
if execStep.IsTerminal() {
// завершился не фейлом → блокируем
return false, false, true, nil
}
// ещё выполняется → ждём
return false, true, false, nil
}
return true, false, nil
return true, false, false, nil
case model.CommitPolicyAfterCanceled:
// Wait for commitAfter dependencies to reach any terminal state (confirmed, failed, cancelled, skipped)
commitAfter := step.CommitAfter
if len(commitAfter) == 0 {
commitAfter = step.DependsOn
}
for _, dep := range commitAfter {
key := strings.TrimSpace(dep)
if key == "" {
continue
}
execStep := execSteps[key]
if execStep == nil {
return false, false, merrors.InvalidArgument("payment plan: commit dependency missing")
return false, true, false,
merrors.InvalidArgument("commit dependency missing")
}
status := executionStepStatus(execStep)
switch status {
case executionStepStatusConfirmed, executionStepStatusFailed, executionStepStatusCancelled, executionStepStatusSkipped:
// Dependency reached terminal state
continue
default:
// Still waiting for terminal state
return false, false, nil
if !execStep.IsTerminal() {
return false, true, false, nil
}
}
return true, false, nil
return true, false, false, nil
default:
// Unknown policy - treat as immediate
return true, false, nil
return true, false, false, nil
}
}
@@ -240,6 +247,7 @@ func commitAfterDependenciesSucceeded(step *model.PaymentStep, execSteps map[str
if step == nil {
return false
}
commitAfter := step.CommitAfter
if len(commitAfter) == 0 {
commitAfter = step.DependsOn
@@ -247,47 +255,73 @@ func commitAfterDependenciesSucceeded(step *model.PaymentStep, execSteps map[str
if len(commitAfter) == 0 {
return false
}
for _, dep := range commitAfter {
key := strings.TrimSpace(dep)
if key == "" {
continue
}
execStep := execSteps[key]
if execStep == nil {
return false
}
status := executionStepStatus(execStep)
switch status {
case executionStepStatusConfirmed, executionStepStatusSkipped:
switch execStep.State {
case model.OperationStateSuccess,
model.OperationStateSkipped:
continue
default:
return false
}
}
return true
}
func cardPayoutDependenciesConfirmed(plan *model.PaymentPlan, execPlan *model.ExecutionPlan) bool {
func cardPayoutDependenciesConfirmed(
plan *model.PaymentPlan,
execPlan *model.ExecutionPlan,
) bool {
if execPlan == nil {
return false
}
if plan == nil || len(plan.Steps) == 0 {
return sourceStepsConfirmed(execPlan)
}
execSteps := executionStepsByCode(execPlan)
planSteps := planStepsByID(plan)
for _, step := range plan.Steps {
if step == nil {
continue
}
if step.Rail != model.RailCardPayout || step.Action != model.RailOperationSend {
if step.Rail != model.RailCardPayout ||
step.Action != model.RailOperationSend {
continue
}
ready, blocked, err := stepDependenciesReady(step, execSteps, planSteps, true)
ready, waiting, blocked, err :=
stepDependenciesReady(step, execSteps, planSteps, true)
if err != nil || blocked {
// payout definitely cannot run
return false
}
if waiting {
// dependencies exist but are not finished yet
// payout must NOT run
return false
}
// only true when dependencies are REALLY satisfied
return ready
}
return false
}

3
api/payments/orchestrator/internal/service/orchestrator/payment_plan_release.go
View File

@@ -35,8 +35,7 @@ func (p *paymentExecutor) releasePaymentHold(ctx context.Context, store storage.
execPlan.Steps[idx] = execStep
}
}
status := executionStepStatus(execStep)
if status == executionStepStatusConfirmed {
if execStep.State == model.OperationStateSuccess {
p.logger.Debug("Payment step already confirmed, skipping", zap.String("step_id", stepID), zap.String("quutation", execQuote.QuoteRef))
continue
}

5
api/payments/orchestrator/internal/service/orchestrator/payment_plan_release_test.go
View File

@@ -41,6 +41,7 @@ func TestReleasePaymentHold_RejectsLegacyLedgerRelease(t *testing.T) {
OrganizationRef: bson.NewObjectID(),
},
Intent: model.PaymentIntent{
Ref: "ref-release-1",
Kind: model.PaymentKindPayout,
Source: model.PaymentEndpoint{
Type: model.EndpointTypeManagedWallet,
@@ -71,13 +72,13 @@ func TestReleasePaymentHold_RejectsLegacyLedgerRelease(t *testing.T) {
if blockStep == nil {
t.Fatalf("expected block step in execution plan")
}
setExecutionStepStatus(blockStep, executionStepStatusConfirmed)
setExecutionStepStatus(blockStep, model.OperationStateSuccess)
err := executor.releasePaymentHold(ctx, store, payment)
if err == nil {
t.Fatal("expected legacy ledger operation error")
}
if !strings.Contains(err.Error(), "unsupported legacy ledger operation") {
if !strings.Contains(err.Error(), "unsupported action") {
t.Fatalf("unexpected error: %v", err)
}
}

304
api/payments/orchestrator/internal/service/orchestrator/payment_plan_steps.go
View File

@@ -11,216 +11,404 @@ import (
"go.uber.org/zap"
)
func (p *paymentExecutor) executePlanStep(ctx context.Context, payment *model.Payment, step *model.PaymentStep, execStep *model.ExecutionStep, quote *orchestratorv1.PaymentQuote, charges []*ledgerv1.PostingLine, idx int) (bool, error) {
func (p *paymentExecutor) executePlanStep(
ctx context.Context,
payment *model.Payment,
step *model.PaymentStep,
execStep *model.ExecutionStep,
quote *orchestratorv1.PaymentQuote,
charges []*ledgerv1.PostingLine,
idx int,
) (bool, error) {
if payment == nil || step == nil || execStep == nil {
return false, merrors.InvalidArgument("payment plan: step is required")
}
if step.Rail == model.RailLedger {
switch step.Action {
case model.RailOperationBlock, model.RailOperationRelease:
p.logger.Warn("Legacy operation detected", zap.String("action", string(step.Action)))
return false, merrors.InvalidArgument("unsupported legacy ledger operation, use ledger.move with roles")
}
stepID := execStep.Code
logger := p.logger.With(
zap.String("payment_ref", payment.PaymentRef),
zap.String("step_id", stepID),
zap.String("rail", string(step.Rail)),
zap.String("action", string(step.Action)),
zap.Int("idx", idx),
)
logger.Debug("Executing payment plan step")
if isStepFinal(execStep) {
logger.Debug("Step already in final state, skipping execution",
zap.String("state", string(execStep.State)),
)
return false, nil
}
switch step.Action {
case model.RailOperationMove:
logger.Debug("Posting ledger move")
amount, err := requireMoney(cloneMoney(step.Amount), "ledger move amount")
if err != nil {
logger.Warn("Ledger move amount invalid", zap.Error(err))
return false, err
}
ref, err := p.postLedgerMove(ctx, payment, step, protoMoney(amount), planStepIdempotencyKey(payment, idx, step), idx)
if err != nil {
logger.Warn("Ledger move failed", zap.Error(err))
return false, err
}
execStep.TransferRef = strings.TrimSpace(ref)
setExecutionStepStatus(execStep, executionStepStatusConfirmed)
setExecutionStepStatus(execStep, model.OperationStateSuccess)
logger.Info("Ledger move completed", zap.String("journal_ref", ref))
return false, nil
case model.RailOperationDebit, model.RailOperationExternalDebit:
logger.Debug("Posting ledger debit")
amount, err := requireMoney(cloneMoney(step.Amount), "ledger debit amount")
if err != nil {
logger.Warn("Ledger debit amount invalid", zap.Error(err))
return false, err
}
ref, err := p.postLedgerDebit(ctx, payment, protoMoney(amount), charges, planStepIdempotencyKey(payment, idx, step), idx, step.Action, quote)
if err != nil {
logger.Warn("Ledger debit failed", zap.Error(err))
return false, err
}
ensureExecutionRefs(payment).DebitEntryRef = ref
setExecutionStepStatus(execStep, executionStepStatusConfirmed)
setExecutionStepStatus(execStep, model.OperationStateSuccess)
logger.Info("Ledger debit completed", zap.String("journal_ref", ref))
return false, nil
case model.RailOperationCredit, model.RailOperationExternalCredit:
logger.Debug("Posting ledger credit")
amount, err := requireMoney(cloneMoney(step.Amount), "ledger credit amount")
if err != nil {
logger.Warn("Ledger credit amount invalid", zap.Error(err))
return false, err
}
ref, err := p.postLedgerCredit(ctx, payment, protoMoney(amount), planStepIdempotencyKey(payment, idx, step), idx, step.Action, quote)
if err != nil {
logger.Warn("Ledger credit failed", zap.Error(err))
return false, err
}
ensureExecutionRefs(payment).CreditEntryRef = ref
setExecutionStepStatus(execStep, executionStepStatusConfirmed)
return false, nil
case model.RailOperationBlock:
if step.Rail != model.RailLedger {
return false, merrors.InvalidArgument("payment plan: block requires ledger rail")
}
amount, err := requireMoney(cloneMoney(step.Amount), "ledger block amount")
if err != nil {
return false, err
}
ref, err := p.postLedgerBlock(ctx, payment, protoMoney(amount), planStepIdempotencyKey(payment, idx, step), idx)
if err != nil {
return false, err
}
execStep.TransferRef = strings.TrimSpace(ref)
setExecutionStepStatus(execStep, executionStepStatusConfirmed)
return false, nil
case model.RailOperationRelease:
if step.Rail != model.RailLedger {
return false, merrors.InvalidArgument("payment plan: release requires ledger rail")
}
amount, err := requireMoney(cloneMoney(step.Amount), "ledger release amount")
if err != nil {
return false, err
}
ref, err := p.postLedgerRelease(ctx, payment, protoMoney(amount), planStepIdempotencyKey(payment, idx, step), idx)
if err != nil {
return false, err
}
execStep.TransferRef = strings.TrimSpace(ref)
setExecutionStepStatus(execStep, executionStepStatusConfirmed)
setExecutionStepStatus(execStep, model.OperationStateSuccess)
logger.Info("Ledger credit completed", zap.String("journal_ref", ref))
return false, nil
case model.RailOperationFXConvert:
logger.Debug("Applying FX conversion")
if err := p.applyFX(ctx, payment, quote, charges, paymentDescription(payment), cloneMetadata(payment.Metadata), ensureExecutionRefs(payment)); err != nil {
logger.Warn("FX conversion failed", zap.Error(err))
return false, err
}
setExecutionStepStatus(execStep, executionStepStatusConfirmed)
setExecutionStepStatus(execStep, model.OperationStateSuccess)
logger.Info("FX conversion completed")
return false, nil
case model.RailOperationObserveConfirm:
setExecutionStepStatus(execStep, executionStepStatusSubmitted)
setExecutionStepStatus(execStep, model.OperationStateWaiting)
logger.Info("ObserveConfirm step set to waiting for external confirmation")
return true, nil
case model.RailOperationSend:
return p.executeSendStep(ctx, payment, step, execStep, quote, idx)
logger.Debug("Executing send step")
async, err := p.executeSendStep(ctx, payment, step, execStep, quote, idx)
if err != nil {
setExecutionStepStatus(execStep, model.OperationStateFailed)
execStep.Error = err.Error()
logger.Warn("Send step failed", zap.Error(err))
return false, err
}
return async, nil
case model.RailOperationFee:
return p.executeFeeStep(ctx, payment, step, execStep, idx)
logger.Debug("Executing fee step")
async, err := p.executeFeeStep(ctx, payment, step, execStep, idx)
if err != nil {
logger.Warn("Fee step failed", zap.Error(err))
return false, err
}
logger.Info("Fee step submitted")
return async, nil
default:
logger.Warn("Unsupported payment plan action")
return false, merrors.InvalidArgument("payment plan: unsupported action")
}
}
func (p *paymentExecutor) executeSendStep(ctx context.Context, payment *model.Payment, step *model.PaymentStep, execStep *model.ExecutionStep, quote *orchestratorv1.PaymentQuote, idx int) (bool, error) {
func (p *paymentExecutor) executeSendStep(
ctx context.Context,
payment *model.Payment,
step *model.PaymentStep,
execStep *model.ExecutionStep,
quote *orchestratorv1.PaymentQuote,
idx int,
) (bool, error) {
stepID := execStep.Code
logger := p.logger.With(
zap.String("payment_ref", payment.PaymentRef),
zap.String("step_id", stepID),
zap.String("rail", string(step.Rail)),
zap.String("action", string(step.Action)),
zap.Int("idx", idx),
)
logger.Debug("Executing send step")
switch step.Rail {
case model.RailCrypto:
logger.Debug("Preparing crypto transfer")
amount, err := requireMoney(cloneMoney(step.Amount), "crypto send amount")
if err != nil {
logger.Warn("Invalid crypto amount", zap.Error(err))
return false, err
}
if !p.deps.railGateways.available() {
logger.Warn("Rail gateway unavailable")
return false, merrors.Internal("rail gateway unavailable")
}
fromRole, toRole := roleHintsForStep(payment.PaymentPlan, idx)
req, err := p.buildCryptoTransferRequest(payment, amount, model.RailOperationSend, planStepIdempotencyKey(payment, idx, step), quote, fromRole, toRole)
req, err := p.buildCryptoTransferRequest(
payment,
amount,
model.RailOperationSend,
planStepIdempotencyKey(payment, idx, step),
execStep.OperationRef,
quote,
fromRole, toRole,
)
if err != nil {
logger.Warn("Failed to build crypto transfer request", zap.Error(err))
return false, err
}
gw, err := p.deps.railGateways.resolve(ctx, step)
if err != nil {
logger.Warn("Failed to resolve rail gateway", zap.Error(err))
return false, err
}
logger.Debug("Sending crypto transfer",
zap.String("idempotency", req.IdempotencyKey), zap.String("intent_ref", req.IntentRef),
zap.String("operation_ref", req.OperationRef),
)
result, err := gw.Send(ctx, req)
if err != nil {
return false, err
execStep.Error = strings.TrimSpace(err.Error())
setExecutionStepStatus(execStep, model.OperationStateFailed)
payment.State = model.PaymentStateFailed
payment.FailureCode = model.PaymentFailureCodeChain
logger.Warn("Send failed; step marked as failed", zap.Error(err))
return false, nil
}
stepID := planStepID(step, idx)
execStep.TransferRef = strings.TrimSpace(result.ReferenceID)
logger.Info("Crypto transfer submitted",
zap.String("transfer_ref", execStep.TransferRef),
)
exec := ensureExecutionRefs(payment)
if exec.ChainTransferRef == "" && execStep.TransferRef != "" {
exec.ChainTransferRef = execStep.TransferRef
}
if execStep.TransferRef != "" {
linkRailObservation(payment, step.Rail, execStep.TransferRef, stepID)
}
setExecutionStepStatus(execStep, executionStepStatusSubmitted)
setExecutionStepStatus(execStep, model.OperationStateWaiting)
return true, nil
case model.RailCardPayout:
logger.Debug("Submitting card payout")
amount, err := requireMoney(cloneMoney(step.Amount), "card payout amount")
if err != nil {
logger.Warn("Invalid card payout amount", zap.Error(err))
return false, err
}
fromRole, toRole := roleHintsForStep(payment.PaymentPlan, idx)
ref, err := p.submitCardPayoutPlan(ctx, payment, protoMoney(amount), fromRole, toRole)
ref, err := p.submitCardPayoutPlan(
ctx,
payment,
execStep.OperationRef,
protoMoney(amount),
fromRole, toRole,
)
if err != nil {
logger.Warn("Card payout submission failed", zap.Error(err))
return false, err
}
execStep.TransferRef = ref
ensureExecutionRefs(payment).CardPayoutRef = ref
setExecutionStepStatus(execStep, executionStepStatusSubmitted)
logger.Info("Card payout submitted", zap.String("payout_ref", ref))
setExecutionStepStatus(execStep, model.OperationStateWaiting)
return true, nil
case model.RailProviderSettlement:
logger.Debug("Preparing provider settlement transfer")
amount, err := requireMoney(cloneMoney(step.Amount), "provider settlement amount")
if err != nil {
logger.Warn("Invalid provider settlement amount", zap.Error(err))
return false, err
}
if !p.deps.railGateways.available() {
logger.Warn("Rail gateway unavailable")
return false, merrors.Internal("rail gateway unavailable")
}
fromRole, toRole := roleHintsForStep(payment.PaymentPlan, idx)
req, err := p.buildProviderSettlementTransferRequest(payment, step, amount, quote, idx, fromRole, toRole)
req, err := p.buildProviderSettlementTransferRequest(
payment,
step,
execStep.OperationRef,
amount,
quote,
idx,
fromRole, toRole)
if err != nil {
logger.Warn("Failed to build provider settlement request", zap.Error(err))
return false, err
}
gw, err := p.deps.railGateways.resolve(ctx, step)
if err != nil {
logger.Warn("Failed to resolve rail gateway", zap.Error(err))
return false, err
}
logger.Info("Sending provider settlement transfer",
zap.String("idempotency", req.IdempotencyKey),
)
result, err := gw.Send(ctx, req)
if err != nil {
return false, err
execStep.Error = strings.TrimSpace(err.Error())
setExecutionStepStatus(execStep, model.OperationStateFailed)
payment.State = model.PaymentStateFailed
payment.FailureCode = model.PaymentFailureCodeSettlement
logger.Warn("Send failed; step marked as failed", zap.Error(err))
return false, nil
}
execStep.TransferRef = strings.TrimSpace(result.ReferenceID)
if execStep.TransferRef == "" {
execStep.TransferRef = strings.TrimSpace(req.IdempotencyKey)
}
logger.Info("Provider settlement submitted",
zap.String("transfer_ref", execStep.TransferRef),
)
linkProviderSettlementObservation(payment, execStep.TransferRef)
setExecutionStepStatus(execStep, executionStepStatusSubmitted)
setExecutionStepStatus(execStep, model.OperationStateWaiting)
return true, nil
case model.RailFiatOnRamp:
logger.Warn("Fiat on-ramp not implemented")
return false, merrors.InvalidArgument("payment plan: fiat on-ramp execution not implemented")
default:
logger.Warn("Unsupported send rail")
return false, merrors.InvalidArgument("payment plan: unsupported send rail")
}
}
func (p *paymentExecutor) executeFeeStep(ctx context.Context, payment *model.Payment, step *model.PaymentStep, execStep *model.ExecutionStep, idx int) (bool, error) {
func (p *paymentExecutor) executeFeeStep(
ctx context.Context,
payment *model.Payment,
step *model.PaymentStep,
execStep *model.ExecutionStep,
idx int,
) (bool, error) {
if payment == nil || step == nil || execStep == nil {
return false, merrors.InvalidArgument("payment plan: fee step is required")
}
switch step.Rail {
case model.RailCrypto:
amount, err := requireMoney(cloneMoney(step.Amount), "crypto fee amount")
if err != nil {
return false, err
}
if !p.deps.railGateways.available() {
return false, merrors.Internal("rail gateway unavailable")
}
fromRole, toRole := roleHintsForStep(payment.PaymentPlan, idx)
req, err := p.buildCryptoTransferRequest(payment, amount, model.RailOperationFee, planStepIdempotencyKey(payment, idx, step), nil, fromRole, toRole)
req, err := p.buildCryptoTransferRequest(
payment,
amount,
model.RailOperationFee,
planStepIdempotencyKey(payment, idx, step),
execStep.OperationRef,
nil,
fromRole,
toRole,
)
if err != nil {
return false, err
}
gw, err := p.deps.railGateways.resolve(ctx, step)
if err != nil {
return false, err
}
p.logger.Debug("Executing crypto fee transfer",
zap.String("payment_ref", payment.PaymentRef),
zap.String("step_id", planStepID(step, idx)),
zap.String("amount", amount.GetAmount()),
zap.String("currency", amount.GetCurrency()),
)
result, err := gw.Send(ctx, req)
if err != nil {
return false, err
p.logger.Warn("Crypto fee transfer failed to submit", zap.Error(err),
zap.String("payment_ref", payment.PaymentRef),
)
return false, nil
}
execStep.TransferRef = strings.TrimSpace(result.ReferenceID)
if execStep.TransferRef != "" {
ensureExecutionRefs(payment).FeeTransferRef = execStep.TransferRef
}
setExecutionStepStatus(execStep, executionStepStatusSubmitted)
// ВАЖНО: больше не Submitted
setExecutionStepStatus(execStep, model.OperationStateWaiting)
p.logger.Info("Crypto fee transfer submitted, waiting confirmation",
zap.String("payment_ref", payment.PaymentRef),
zap.String("transfer_ref", execStep.TransferRef),
)
return true, nil
default:
return false, merrors.InvalidArgument("payment plan: unsupported fee rail")
}

14
api/payments/orchestrator/internal/service/orchestrator/plan_builder_default_test.go
View File

@@ -7,7 +7,7 @@ import (
"github.com/tech/sendico/payments/orchestrator/storage/model"
mloggerfactory "github.com/tech/sendico/pkg/mlogger/factory"
pmodel "github.com/tech/sendico/pkg/model"
"github.com/tech/sendico/pkg/model/account_role"
paymenttypes "github.com/tech/sendico/pkg/payments/types"
moneyv1 "github.com/tech/sendico/pkg/proto/common/money/v1"
orchestratorv1 "github.com/tech/sendico/pkg/proto/payments/orchestrator/v1"
@@ -21,6 +21,7 @@ func TestDefaultPlanBuilder_BuildsPlanFromRoutes_CryptoToCard(t *testing.T) {
PaymentRef: "pay-1",
IdempotencyKey: "idem-1",
Intent: model.PaymentIntent{
Ref: "ref-1",
Kind: model.PaymentKindPayout,
Source: model.PaymentEndpoint{
Type: model.EndpointTypeManagedWallet,
@@ -67,9 +68,9 @@ func TestDefaultPlanBuilder_BuildsPlanFromRoutes_CryptoToCard(t *testing.T) {
{StepID: "crypto_send", Rail: model.RailCrypto, Operation: "payout.crypto"},
{StepID: "crypto_fee", Rail: model.RailCrypto, Operation: "fee.send", DependsOn: []string{"crypto_send"}},
{StepID: "crypto_observe", Rail: model.RailCrypto, Operation: "observe.confirm", DependsOn: []string{"crypto_send"}},
{StepID: "ledger_credit", Rail: model.RailLedger, Operation: "ledger.move", DependsOn: []string{"crypto_observe"}, FromRole: rolePtr(pmodel.AccountRolePending), ToRole: rolePtr(pmodel.AccountRoleOperating)},
{StepID: "ledger_credit", Rail: model.RailLedger, Operation: "ledger.move", DependsOn: []string{"crypto_observe"}, FromRole: rolePtr(account_role.AccountRolePending), ToRole: rolePtr(account_role.AccountRoleOperating)},
{StepID: "card_payout", Rail: model.RailCardPayout, Operation: "payout.card", DependsOn: []string{"ledger_credit"}},
{StepID: "ledger_debit", Rail: model.RailLedger, Operation: "ledger.move", DependsOn: []string{"card_payout"}, CommitPolicy: model.CommitPolicyAfterSuccess, CommitAfter: []string{"card_payout"}, FromRole: rolePtr(pmodel.AccountRoleOperating), ToRole: rolePtr(pmodel.AccountRoleTransit)},
{StepID: "ledger_debit", Rail: model.RailLedger, Operation: "ledger.move", DependsOn: []string{"card_payout"}, CommitPolicy: model.CommitPolicyAfterSuccess, CommitAfter: []string{"card_payout"}, FromRole: rolePtr(account_role.AccountRoleOperating), ToRole: rolePtr(account_role.AccountRoleTransit)},
},
},
},
@@ -143,6 +144,7 @@ func TestDefaultPlanBuilder_ErrorsWhenRouteMissing(t *testing.T) {
PaymentRef: "pay-1",
IdempotencyKey: "idem-1",
Intent: model.PaymentIntent{
Ref: "ref-1",
Kind: model.PaymentKindPayout,
Source: model.PaymentEndpoint{
Type: model.EndpointTypeManagedWallet,
@@ -177,6 +179,7 @@ func TestBuildPlanFromTemplate_ProviderSettlementUsesNetAmountWhenFixReceived(t
PaymentRef: "pay-settle-1",
IdempotencyKey: "idem-settle-1",
Intent: model.PaymentIntent{
Ref: "ref-1",
Kind: model.PaymentKindPayout,
SettlementMode: model.SettlementModeFixReceived,
Amount: &paymenttypes.Money{Currency: "USDT", Amount: "100"},
@@ -240,6 +243,7 @@ func TestDefaultPlanBuilder_UsesSourceCurrencyForCryptoSendWithFX(t *testing.T)
PaymentRef: "pay-2",
IdempotencyKey: "idem-2",
Intent: model.PaymentIntent{
Ref: "ref-1",
Kind: model.PaymentKindPayout,
RequiresFX: true,
Source: model.PaymentEndpoint{
@@ -290,9 +294,9 @@ func TestDefaultPlanBuilder_UsesSourceCurrencyForCryptoSendWithFX(t *testing.T)
{StepID: "crypto_send", Rail: model.RailCrypto, Operation: "payout.crypto"},
{StepID: "crypto_fee", Rail: model.RailCrypto, Operation: "fee.send", DependsOn: []string{"crypto_send"}},
{StepID: "crypto_observe", Rail: model.RailCrypto, Operation: "observe.confirm", DependsOn: []string{"crypto_send"}},
{StepID: "ledger_credit", Rail: model.RailLedger, Operation: "ledger.move", DependsOn: []string{"crypto_observe"}, FromRole: rolePtr(pmodel.AccountRolePending), ToRole: rolePtr(pmodel.AccountRoleOperating)},
{StepID: "ledger_credit", Rail: model.RailLedger, Operation: "ledger.move", DependsOn: []string{"crypto_observe"}, FromRole: rolePtr(account_role.AccountRolePending), ToRole: rolePtr(account_role.AccountRoleOperating)},
{StepID: "card_payout", Rail: model.RailCardPayout, Operation: "payout.card", DependsOn: []string{"ledger_credit"}},
{StepID: "ledger_debit", Rail: model.RailLedger, Operation: "ledger.move", DependsOn: []string{"card_payout"}, CommitPolicy: model.CommitPolicyAfterSuccess, CommitAfter: []string{"card_payout"}, FromRole: rolePtr(pmodel.AccountRoleOperating), ToRole: rolePtr(pmodel.AccountRoleTransit)},
{StepID: "ledger_debit", Rail: model.RailLedger, Operation: "ledger.move", DependsOn: []string{"card_payout"}, CommitPolicy: model.CommitPolicyAfterSuccess, CommitAfter: []string{"card_payout"}, FromRole: rolePtr(account_role.AccountRoleOperating), ToRole: rolePtr(account_role.AccountRoleTransit)},
},
},
},

10
api/payments/orchestrator/internal/service/orchestrator/plan_builder_steps.go
View File

@@ -6,7 +6,7 @@ import (
"github.com/tech/sendico/payments/orchestrator/storage/model"
"github.com/tech/sendico/pkg/merrors"
pmodel "github.com/tech/sendico/pkg/model"
"github.com/tech/sendico/pkg/model/account_role"
"github.com/tech/sendico/pkg/mutil/mzap"
paymenttypes "github.com/tech/sendico/pkg/payments/types"
oraclev1 "github.com/tech/sendico/pkg/proto/oracle/v1"
@@ -116,7 +116,11 @@ func (b *defaultPlanBuilder) buildPlanFromTemplate(ctx context.Context, payment
if action == model.RailOperationSend && tpl.Rail == model.RailProviderSettlement {
amount = cloneMoney(providerSettlementAmount)
}
if amount == nil && action != model.RailOperationObserveConfirm {
if amount == nil &&
action != model.RailOperationObserveConfirm &&
action != model.RailOperationFee {
logger.Warn("Plan template step has no amount for action, skipping",
zap.String("step_id", stepID), zap.String("action", string(action)))
continue
}
@@ -349,7 +353,7 @@ func observeAmountForRail(rail model.Rail, source, settlement, payout *paymentty
return source
}
func cloneAccountRole(role *pmodel.AccountRole) *pmodel.AccountRole {
func cloneAccountRole(role *account_role.AccountRole) *account_role.AccountRole {
if role == nil {
return nil
}

10
api/payments/orchestrator/internal/service/orchestrator/provider_settlement.go
View File

@@ -5,14 +5,14 @@ import (
"github.com/tech/sendico/payments/orchestrator/storage/model"
"github.com/tech/sendico/pkg/merrors"
pmodel "github.com/tech/sendico/pkg/model"
"github.com/tech/sendico/pkg/model/account_role"
"github.com/tech/sendico/pkg/payments/rail"
paymenttypes "github.com/tech/sendico/pkg/payments/types"
orchestratorv1 "github.com/tech/sendico/pkg/proto/payments/orchestrator/v1"
)
const (
providerSettlementMetaPaymentIntentID = "payment_intent_id"
providerSettlementMetaPaymentIntentID = "payment_ref"
providerSettlementMetaQuoteRef = "quote_ref"
providerSettlementMetaTargetChatID = "target_chat_id"
providerSettlementMetaOutgoingLeg = "outgoing_leg"
@@ -20,7 +20,7 @@ const (
providerSettlementMetaSourceCurrency = "source_currency"
)
func (p *paymentExecutor) buildProviderSettlementTransferRequest(payment *model.Payment, step *model.PaymentStep, amount *paymenttypes.Money, quote *orchestratorv1.PaymentQuote, idx int, fromRole, toRole *pmodel.AccountRole) (rail.TransferRequest, error) {
func (p *paymentExecutor) buildProviderSettlementTransferRequest(payment *model.Payment, step *model.PaymentStep, operationRef string, amount *paymenttypes.Money, quote *orchestratorv1.PaymentQuote, idx int, fromRole, toRole *account_role.AccountRole) (rail.TransferRequest, error) {
if payment == nil || step == nil {
return rail.TransferRequest{}, merrors.InvalidArgument("provider settlement: payment and step are required")
}
@@ -86,7 +86,9 @@ func (p *paymentExecutor) buildProviderSettlementTransferRequest(payment *model.
IdempotencyKey: requestID,
DestinationMemo: paymentRef,
Metadata: metadata,
ClientReference: paymentRef,
PaymentRef: paymentRef,
OperationRef: operationRef,
IntentRef: payment.Intent.Ref,
}
if fromRole != nil {
req.FromRole = *fromRole

22
api/payments/orchestrator/internal/service/orchestrator/provider_settlement_gateway.go
View File

@@ -61,7 +61,7 @@ func (g *providerSettlementGateway) Send(ctx context.Context, req rail.TransferR
metadata = map[string]string{}
}
if strings.TrimSpace(metadata[providerSettlementMetaPaymentIntentID]) == "" {
if ref := strings.TrimSpace(req.ClientReference); ref != "" {
if ref := strings.TrimSpace(req.PaymentRef); ref != "" {
metadata[providerSettlementMetaPaymentIntentID] = ref
}
}
@@ -79,8 +79,10 @@ func (g *providerSettlementGateway) Send(ctx context.Context, req rail.TransferR
Currency: currency,
Amount: amount,
},
Metadata: metadata,
ClientReference: strings.TrimSpace(req.ClientReference),
Metadata: metadata,
PaymentRef: strings.TrimSpace(req.PaymentRef),
IntentRef: req.IntentRef,
OperationRef: req.OperationRef,
}
if dest := buildProviderSettlementDestination(req); dest != nil {
submitReq.Destination = dest
@@ -143,16 +145,22 @@ func buildProviderSettlementDestination(req rail.TransferRequest) *chainv1.Trans
}
}
func providerSettlementStatusFromTransfer(status chainv1.TransferStatus) string {
func providerSettlementStatusFromTransfer(status chainv1.TransferStatus) rail.TransferStatus {
switch status {
case chainv1.TransferStatus_TRANSFER_CONFIRMED:
case chainv1.TransferStatus_TRANSFER_SUCCESS:
return rail.TransferStatusSuccess
case chainv1.TransferStatus_TRANSFER_FAILED:
return rail.TransferStatusFailed
case chainv1.TransferStatus_TRANSFER_CANCELLED:
return rail.TransferStatusRejected
// our cancellation, not from provider
return rail.TransferStatusFailed
default:
return rail.TransferStatusPending
// CREATED, PROCESSING, WAITING
return rail.TransferStatusWaiting
}
}

3
api/payments/orchestrator/internal/service/orchestrator/quote_engine_test.go
View File

@@ -60,6 +60,7 @@ func TestBuildPaymentQuote_RequestsFXWhenSettlementDiffers(t *testing.T) {
req := &orchestratorv1.QuotePaymentRequest{
Meta: &orchestratorv1.RequestMeta{OrganizationRef: "org"},
Intent: &orchestratorv1.PaymentIntent{
Ref: "ref-1",
Source: &orchestratorv1.PaymentEndpoint{
Endpoint: &orchestratorv1.PaymentEndpoint_Ledger{Ledger: &orchestratorv1.LedgerEndpoint{LedgerAccountRef: "ledger:source"}},
},
@@ -94,6 +95,7 @@ func TestBuildPaymentQuote_FeesRequestedForExternalRails(t *testing.T) {
req := &orchestratorv1.QuotePaymentRequest{
Meta: &orchestratorv1.RequestMeta{OrganizationRef: "org"},
Intent: &orchestratorv1.PaymentIntent{
Ref: "ref-1",
Source: &orchestratorv1.PaymentEndpoint{
Endpoint: &orchestratorv1.PaymentEndpoint_Ledger{Ledger: &orchestratorv1.LedgerEndpoint{LedgerAccountRef: "ledger:source"}},
},
@@ -128,6 +130,7 @@ func TestBuildPaymentQuote_FeesSkippedForLedgerTransfer(t *testing.T) {
req := &orchestratorv1.QuotePaymentRequest{
Meta: &orchestratorv1.RequestMeta{OrganizationRef: "org"},
Intent: &orchestratorv1.PaymentIntent{
Ref: "ref-1",
Source: &orchestratorv1.PaymentEndpoint{
Endpoint: &orchestratorv1.PaymentEndpoint_Ledger{Ledger: &orchestratorv1.LedgerEndpoint{LedgerAccountRef: "ledger:source"}},
},

2
api/payments/orchestrator/internal/service/orchestrator/quote_payment_idempotency_integration_test.go
View File

@@ -83,7 +83,9 @@ func TestQuotePayment_IdempotencyReuseAfterExpiry(t *testing.T) {
req := &orchestratorv1.QuotePaymentRequest{
Meta: &orchestratorv1.RequestMeta{OrganizationRef: orgID.Hex()},
IdempotencyKey: "idem-expired-quote",
Ref: "ref-expired",
Intent: &orchestratorv1.PaymentIntent{
Ref: "intent-1",
Source: &orchestratorv1.PaymentEndpoint{
Endpoint: &orchestratorv1.PaymentEndpoint_Ledger{
Ledger: &orchestratorv1.LedgerEndpoint{LedgerAccountRef: "ledger:source"},

3
api/payments/orchestrator/internal/service/orchestrator/quote_request_test.go
View File

@@ -45,6 +45,7 @@ func TestRequestFXQuoteUsesQuoteAmountWhenCurrencyMatchesQuote(t *testing.T) {
req := &orchestratorv1.QuotePaymentRequest{
Meta: &orchestratorv1.RequestMeta{OrganizationRef: "org"},
Intent: &orchestratorv1.PaymentIntent{
Ref: "ref-1",
Amount: &moneyv1.Money{Currency: "USD", Amount: "100"},
SettlementCurrency: "USD",
Fx: &orchestratorv1.FXIntent{
@@ -83,6 +84,7 @@ func TestRequestFXQuoteFailsWhenRequiredAndOracleUnavailable(t *testing.T) {
req := &orchestratorv1.QuotePaymentRequest{
Meta: &orchestratorv1.RequestMeta{OrganizationRef: "org"},
Intent: &orchestratorv1.PaymentIntent{
Ref: "ref-1",
RequiresFx: true,
Amount: &moneyv1.Money{Currency: "USDT", Amount: "1"},
SettlementCurrency: "RUB",
@@ -125,6 +127,7 @@ func TestRequestFXQuoteFailsWhenRequiredAndQuoteMissing(t *testing.T) {
req := &orchestratorv1.QuotePaymentRequest{
Meta: &orchestratorv1.RequestMeta{OrganizationRef: "org"},
Intent: &orchestratorv1.PaymentIntent{
Ref: "ref-1",
RequiresFx: true,
Amount: &moneyv1.Money{Currency: "USDT", Amount: "1"},
SettlementCurrency: "RUB",

8
api/payments/orchestrator/internal/service/orchestrator/rail_gateway_fake_test.go
View File

@@ -32,26 +32,26 @@ func (f *fakeRailGateway) Send(ctx context.Context, req rail.TransferRequest) (r
if f.sendFn != nil {
return f.sendFn(ctx, req)
}
return rail.RailResult{ReferenceID: "transfer-1", Status: rail.TransferStatusPending}, nil
return rail.RailResult{ReferenceID: "transfer-1", Status: rail.TransferStatusWaiting}, nil
}
func (f *fakeRailGateway) Observe(ctx context.Context, referenceID string) (rail.ObserveResult, error) {
if f.observeFn != nil {
return f.observeFn(ctx, referenceID)
}
return rail.ObserveResult{ReferenceID: referenceID, Status: rail.TransferStatusPending}, nil
return rail.ObserveResult{ReferenceID: referenceID, Status: rail.TransferStatusWaiting}, nil
}
func (f *fakeRailGateway) Block(ctx context.Context, req rail.BlockRequest) (rail.RailResult, error) {
if f.blockFn != nil {
return f.blockFn(ctx, req)
}
return rail.RailResult{ReferenceID: req.IdempotencyKey, Status: rail.TransferStatusPending}, nil
return rail.RailResult{ReferenceID: req.IdempotencyKey, Status: rail.TransferStatusWaiting}, nil
}
func (f *fakeRailGateway) Release(ctx context.Context, req rail.ReleaseRequest) (rail.RailResult, error) {
if f.releaseFn != nil {
return f.releaseFn(ctx, req)
}
return rail.RailResult{ReferenceID: req.ReferenceID, Status: rail.TransferStatusPending}, nil
return rail.RailResult{ReferenceID: req.ReferenceID, Status: rail.TransferStatusWaiting}, nil
}

30
api/payments/orchestrator/internal/service/orchestrator/service_helpers_test.go
View File

@@ -11,7 +11,7 @@ import (
"github.com/tech/sendico/payments/orchestrator/storage/model"
clockpkg "github.com/tech/sendico/pkg/clock"
mloggerfactory "github.com/tech/sendico/pkg/mlogger/factory"
pmodel "github.com/tech/sendico/pkg/model"
"github.com/tech/sendico/pkg/model/account_role"
moneyv1 "github.com/tech/sendico/pkg/proto/common/money/v1"
connectorv1 "github.com/tech/sendico/pkg/proto/connector/v1"
ledgerv1 "github.com/tech/sendico/pkg/proto/ledger/v1"
@@ -54,6 +54,7 @@ func TestRequireIdempotencyKey(t *testing.T) {
func TestNewPayment(t *testing.T) {
org := bson.NewObjectID()
intent := &orchestratorv1.PaymentIntent{
Ref: "ref-1",
Amount: &moneyv1.Money{Currency: "USD", Amount: "10"},
SettlementMode: orchestratorv1.SettlementMode_SETTLEMENT_FIX_RECEIVED,
SettlementCurrency: "USD",
@@ -81,10 +82,12 @@ func TestResolvePaymentQuote_NotFound(t *testing.T) {
clock: clockpkg.NewSystem(),
}
_, _, err := svc.resolvePaymentQuote(context.Background(), quoteResolutionInput{
OrgRef: org.Hex(),
OrgID: org,
Meta: &orchestratorv1.RequestMeta{OrganizationRef: org.Hex()},
Intent: &orchestratorv1.PaymentIntent{Amount: &moneyv1.Money{Currency: "USD", Amount: "1"}, SettlementCurrency: "USD"},
OrgRef: org.Hex(),
OrgID: org,
Meta: &orchestratorv1.RequestMeta{OrganizationRef: org.Hex()},
Intent: &orchestratorv1.PaymentIntent{
Ref: "ref-1",
Amount: &moneyv1.Money{Currency: "USD", Amount: "1"}, SettlementCurrency: "USD"},
QuoteRef: "missing",
})
if qerr, ok := err.(quoteResolutionError); !ok || qerr.code != "quote_not_found" {
@@ -94,7 +97,9 @@ func TestResolvePaymentQuote_NotFound(t *testing.T) {
func TestResolvePaymentQuote_Expired(t *testing.T) {
org := bson.NewObjectID()
intent := &orchestratorv1.PaymentIntent{Amount: &moneyv1.Money{Currency: "USD", Amount: "1"}, SettlementCurrency: "USD"}
intent := &orchestratorv1.PaymentIntent{
Ref: "ref-1",
Amount: &moneyv1.Money{Currency: "USD", Amount: "1"}, SettlementCurrency: "USD"}
record := &model.PaymentQuoteRecord{
QuoteRef: "q1",
Intent: intentFromProto(intent),
@@ -119,7 +124,9 @@ func TestResolvePaymentQuote_Expired(t *testing.T) {
func TestResolvePaymentQuote_QuoteRefUsesStoredIntent(t *testing.T) {
org := bson.NewObjectID()
intent := &orchestratorv1.PaymentIntent{Amount: &moneyv1.Money{Currency: "USD", Amount: "1"}, SettlementCurrency: "USD"}
intent := &orchestratorv1.PaymentIntent{
Ref: "ref-1",
Amount: &moneyv1.Money{Currency: "USD", Amount: "1"}, SettlementCurrency: "USD"}
record := &model.PaymentQuoteRecord{
QuoteRef: "q1",
Intent: intentFromProto(intent),
@@ -149,6 +156,7 @@ func TestResolvePaymentQuote_QuoteRefUsesStoredIntent(t *testing.T) {
func TestResolvePaymentQuote_QuoteRefSkipsQuoteRecompute(t *testing.T) {
org := bson.NewObjectID()
intent := &orchestratorv1.PaymentIntent{
Ref: "ref-1",
Amount: &moneyv1.Money{Currency: "USD", Amount: "1"},
SettlementCurrency: "USD",
}
@@ -222,7 +230,7 @@ func TestInitiatePaymentIdempotency(t *testing.T) {
ToRail: model.RailLedger,
IsEnabled: true,
Steps: []model.OrchestrationStep{
{StepID: "ledger_move", Rail: model.RailLedger, Operation: "ledger.move", FromRole: rolePtr(pmodel.AccountRoleOperating), ToRole: rolePtr(pmodel.AccountRoleTransit)},
{StepID: "ledger_move", Rail: model.RailLedger, Operation: "ledger.move", FromRole: rolePtr(account_role.AccountRoleOperating), ToRole: rolePtr(account_role.AccountRoleTransit)},
},
},
},
@@ -235,6 +243,7 @@ func TestInitiatePaymentIdempotency(t *testing.T) {
svc.ensureHandlers()
intent := &orchestratorv1.PaymentIntent{
Ref: "ref-1",
Source: &orchestratorv1.PaymentEndpoint{
Endpoint: &orchestratorv1.PaymentEndpoint_Ledger{Ledger: &orchestratorv1.LedgerEndpoint{LedgerAccountRef: "ledger:source"}},
},
@@ -267,6 +276,7 @@ func TestInitiatePaymentByQuoteRef(t *testing.T) {
org := bson.NewObjectID()
store := newHelperPaymentStore()
intent := &orchestratorv1.PaymentIntent{
Ref: "ref-1",
Source: &orchestratorv1.PaymentEndpoint{
Endpoint: &orchestratorv1.PaymentEndpoint_Ledger{Ledger: &orchestratorv1.LedgerEndpoint{LedgerAccountRef: "ledger:source"}},
},
@@ -308,7 +318,7 @@ func TestInitiatePaymentByQuoteRef(t *testing.T) {
ToRail: model.RailLedger,
IsEnabled: true,
Steps: []model.OrchestrationStep{
{StepID: "ledger_move", Rail: model.RailLedger, Operation: "ledger.move", FromRole: rolePtr(pmodel.AccountRoleOperating), ToRole: rolePtr(pmodel.AccountRoleTransit)},
{StepID: "ledger_move", Rail: model.RailLedger, Operation: "ledger.move", FromRole: rolePtr(account_role.AccountRoleOperating), ToRole: rolePtr(account_role.AccountRoleTransit)},
},
},
},
@@ -460,6 +470,6 @@ func (s *helperQuotesStore) GetByIdempotencyKey(_ context.Context, orgRef bson.O
return nil, storage.ErrQuoteNotFound
}
func rolePtr(role pmodel.AccountRole) *pmodel.AccountRole {
func rolePtr(role account_role.AccountRole) *account_role.AccountRole {
return &role
}

31
api/payments/orchestrator/internal/service/orchestrator/service_test.go
View File

@@ -13,6 +13,7 @@ import (
"github.com/tech/sendico/pkg/api/routers/gsresponse"
"github.com/tech/sendico/pkg/merrors"
mo "github.com/tech/sendico/pkg/model"
"github.com/tech/sendico/pkg/model/account_role"
"github.com/tech/sendico/pkg/payments/rail"
paymenttypes "github.com/tech/sendico/pkg/payments/types"
moneyv1 "github.com/tech/sendico/pkg/proto/common/money/v1"
@@ -53,6 +54,7 @@ func TestExecutePayment_FXConversionSettled(t *testing.T) {
IdempotencyKey: "fx-1",
OrganizationBoundBase: mo.OrganizationBoundBase{OrganizationRef: bson.NewObjectID()},
Intent: model.PaymentIntent{
Ref: "ref-1",
Kind: model.PaymentKindFXConversion,
Source: model.PaymentEndpoint{
Type: model.EndpointTypeLedger,
@@ -75,7 +77,6 @@ func TestExecutePayment_FXConversionSettled(t *testing.T) {
Price: &moneyv1.Decimal{Value: "0.9"},
},
}
if err := svc.executePayment(ctx, store, payment, quote); err != nil {
t.Fatalf("executePayment returned error: %v", err)
}
@@ -83,6 +84,7 @@ func TestExecutePayment_FXConversionSettled(t *testing.T) {
if payment.State != model.PaymentStateSettled {
t.Fatalf("expected payment settled, got %s", payment.State)
}
if payment.Execution == nil || payment.Execution.FXEntryRef == "" {
t.Fatal("expected FX entry ref set on payment execution")
}
@@ -107,7 +109,7 @@ func TestExecutePayment_ChainFailure(t *testing.T) {
Steps: []model.OrchestrationStep{
{StepID: "crypto_send", Rail: model.RailCrypto, Operation: "payout.crypto"},
{StepID: "crypto_observe", Rail: model.RailCrypto, Operation: "observe.confirm", DependsOn: []string{"crypto_send"}},
{StepID: "ledger_credit", Rail: model.RailLedger, Operation: "ledger.move", DependsOn: []string{"crypto_observe"}, FromRole: rolePtr(mo.AccountRolePending), ToRole: rolePtr(mo.AccountRoleOperating)},
{StepID: "ledger_credit", Rail: model.RailLedger, Operation: "ledger.move", DependsOn: []string{"crypto_observe"}, FromRole: rolePtr(account_role.AccountRolePending), ToRole: rolePtr(account_role.AccountRoleOperating)},
},
},
},
@@ -156,6 +158,7 @@ func TestExecutePayment_ChainFailure(t *testing.T) {
IdempotencyKey: "chain-1",
OrganizationBoundBase: mo.OrganizationBoundBase{OrganizationRef: bson.NewObjectID()},
Intent: model.PaymentIntent{
Ref: "ref-1",
Kind: model.PaymentKindPayout,
Source: model.PaymentEndpoint{
Type: model.EndpointTypeManagedWallet,
@@ -210,7 +213,7 @@ func TestProcessTransferUpdateHandler_Settled(t *testing.T) {
Event: &chainv1.TransferStatusChangedEvent{
Transfer: &chainv1.Transfer{
TransferRef: "transfer-1",
Status: chainv1.TransferStatus_TRANSFER_CONFIRMED,
Status: chainv1.TransferStatus_TRANSFER_SUCCESS,
},
},
}
@@ -231,6 +234,7 @@ func TestProcessTransferUpdateHandler_CardFundingWaitsForSources(t *testing.T) {
PaymentRef: "pay-card",
State: model.PaymentStateSubmitted,
Intent: model.PaymentIntent{
Ref: "ref-1",
Destination: model.PaymentEndpoint{
Type: model.EndpointTypeCard,
Card: &model.CardEndpoint{MaskedPan: "4111"},
@@ -243,16 +247,16 @@ func TestProcessTransferUpdateHandler_CardFundingWaitsForSources(t *testing.T) {
fundStep := ensureExecutionStep(plan, stepCodeFundingTransfer)
fundStep.TransferRef = "fund-1"
setExecutionStepRole(fundStep, executionStepRoleSource)
setExecutionStepStatus(fundStep, executionStepStatusSubmitted)
setExecutionStepStatus(fundStep, model.OperationStateWaiting)
feeStep := ensureExecutionStep(plan, stepCodeFeeTransfer)
feeStep.TransferRef = "fee-1"
setExecutionStepRole(feeStep, executionStepRoleSource)
setExecutionStepStatus(feeStep, executionStepStatusSubmitted)
setExecutionStepStatus(feeStep, model.OperationStateWaiting)
cardStep := ensureExecutionStep(plan, stepCodeCardPayout)
setExecutionStepRole(cardStep, executionStepRoleConsumer)
setExecutionStepStatus(cardStep, executionStepStatusPlanned)
setExecutionStepStatus(cardStep, model.OperationStatePlanned)
store := newStubPaymentsStore()
store.payments[payment.PaymentRef] = payment
@@ -275,7 +279,7 @@ func TestProcessTransferUpdateHandler_CardFundingWaitsForSources(t *testing.T) {
step := ensureExecutionStep(plan, stepCodeCardPayout)
setExecutionStepRole(step, executionStepRoleConsumer)
step.TransferRef = "payout-1"
setExecutionStepStatus(step, executionStepStatusSubmitted)
setExecutionStepStatus(step, model.OperationStateWaiting)
return nil
}
svc.h.events = newPaymentEventHandler(svc.storage, svc.ensureRepository, svc.logger, submit, nil, nil)
@@ -284,7 +288,7 @@ func TestProcessTransferUpdateHandler_CardFundingWaitsForSources(t *testing.T) {
Event: &chainv1.TransferStatusChangedEvent{
Transfer: &chainv1.Transfer{
TransferRef: "fund-1",
Status: chainv1.TransferStatus_TRANSFER_CONFIRMED,
Status: chainv1.TransferStatus_TRANSFER_SUCCESS,
},
},
}
@@ -295,8 +299,8 @@ func TestProcessTransferUpdateHandler_CardFundingWaitsForSources(t *testing.T) {
if payoutCalls != 0 {
t.Fatalf("expected no payout on first confirmation, got %d", payoutCalls)
}
if executionStepStatus(fundStep) != executionStepStatusConfirmed {
t.Fatalf("expected funding step confirmed, got %s", executionStepStatus(fundStep))
if fundStep.State != model.OperationStateSuccess {
t.Fatalf("expected funding step confirmed, got %s", feeStep.State)
}
if resp.GetPayment().GetState() != orchestratorv1.PaymentState_PAYMENT_STATE_SUBMITTED {
t.Fatalf("expected submitted state, got %s", resp.GetPayment().GetState())
@@ -306,7 +310,7 @@ func TestProcessTransferUpdateHandler_CardFundingWaitsForSources(t *testing.T) {
Event: &chainv1.TransferStatusChangedEvent{
Transfer: &chainv1.Transfer{
TransferRef: "fee-1",
Status: chainv1.TransferStatus_TRANSFER_CONFIRMED,
Status: chainv1.TransferStatus_TRANSFER_SUCCESS,
},
},
}
@@ -317,8 +321,8 @@ func TestProcessTransferUpdateHandler_CardFundingWaitsForSources(t *testing.T) {
if payoutCalls != 1 {
t.Fatalf("expected payout after all sources confirmed, got %d", payoutCalls)
}
if executionStepStatus(feeStep) != executionStepStatusConfirmed {
t.Fatalf("expected fee step confirmed, got %s", executionStepStatus(feeStep))
if feeStep.State != model.OperationStateSuccess {
t.Fatalf("expected fee step confirmed, got %s", string(model.OperationStateSuccess))
}
if resp.GetPayment().GetExecution().GetCardPayoutRef() != "payout-1" {
t.Fatalf("expected card payout ref set, got %s", resp.GetPayment().GetExecution().GetCardPayoutRef())
@@ -331,6 +335,7 @@ func TestProcessDepositObservedHandler_MatchesPayment(t *testing.T) {
PaymentRef: "pay-2",
State: model.PaymentStateSubmitted,
Intent: model.PaymentIntent{
Ref: "ref-2",
Destination: model.PaymentEndpoint{
Type: model.EndpointTypeManagedWallet,
ManagedWallet: &model.ManagedWalletEndpoint{

15
api/payments/orchestrator/storage/model/operation.go Normal file
View File

@@ -0,0 +1,15 @@
package model
type OperationState string
const (
OperationStateCreated OperationState = "created" // record exists, not started
OperationStateProcessing OperationState = "processing" // we are working on it
OperationStatePlanned OperationState = "planned" // waiting for execution
OperationStateWaiting OperationState = "waiting" // waiting external world
OperationStateSuccess OperationState = "success" // final success
OperationStateFailed OperationState = "failed" // final failure
OperationStateCancelled OperationState = "cancelled" // final cancelled
OperationStateSkipped OperationState = "skipped" // final skipped
)

56
api/payments/orchestrator/storage/model/payment.go
View File

@@ -6,7 +6,7 @@ import (
"github.com/tech/sendico/pkg/db/storable"
"github.com/tech/sendico/pkg/model"
pmodel "github.com/tech/sendico/pkg/model"
"github.com/tech/sendico/pkg/model/account_role"
"github.com/tech/sendico/pkg/mservice"
paymenttypes "github.com/tech/sendico/pkg/payments/types"
)
@@ -65,6 +65,7 @@ const (
PaymentFailureCodeChain PaymentFailureCode = "chain"
PaymentFailureCodeFees PaymentFailureCode = "fees"
PaymentFailureCodePolicy PaymentFailureCode = "policy"
PaymentFailureCodeSettlement PaymentFailureCode = "settlement"
)
// Rail identifies a payment rail for orchestration.
@@ -220,6 +221,7 @@ type FXIntent struct {
// PaymentIntent models the requested payment operation.
type PaymentIntent struct {
Ref string `bson:"ref" json:"ref"`
Kind PaymentKind `bson:"kind" json:"kind"`
Source PaymentEndpoint `bson:"source" json:"source"`
Destination PaymentEndpoint `bson:"destination" json:"destination"`
@@ -271,18 +273,17 @@ type ExecutionRefs struct {
// PaymentStep is an explicit action within a payment plan.
type PaymentStep struct {
StepID string `bson:"stepId,omitempty" json:"stepId,omitempty"`
Rail Rail `bson:"rail" json:"rail"`
GatewayID string `bson:"gatewayId,omitempty" json:"gatewayId,omitempty"`
InstanceID string `bson:"instanceId,omitempty" json:"instanceId,omitempty"`
Action RailOperation `bson:"action" json:"action"`
DependsOn []string `bson:"dependsOn,omitempty" json:"dependsOn,omitempty"`
CommitPolicy CommitPolicy `bson:"commitPolicy,omitempty" json:"commitPolicy,omitempty"`
CommitAfter []string `bson:"commitAfter,omitempty" json:"commitAfter,omitempty"`
Amount *paymenttypes.Money `bson:"amount,omitempty" json:"amount,omitempty"`
Ref string `bson:"ref,omitempty" json:"ref,omitempty"`
FromRole *pmodel.AccountRole `bson:"fromRole,omitempty" json:"fromRole,omitempty"`
ToRole *pmodel.AccountRole `bson:"toRole,omitempty" json:"toRole,omitempty"`
StepID string `bson:"stepId,omitempty" json:"stepId,omitempty"`
Rail Rail `bson:"rail" json:"rail"`
GatewayID string `bson:"gatewayId,omitempty" json:"gatewayId,omitempty"`
InstanceID string `bson:"instanceId,omitempty" json:"instanceId,omitempty"`
Action RailOperation `bson:"action" json:"action"`
DependsOn []string `bson:"dependsOn,omitempty" json:"dependsOn,omitempty"`
CommitPolicy CommitPolicy `bson:"commitPolicy,omitempty" json:"commitPolicy,omitempty"`
CommitAfter []string `bson:"commitAfter,omitempty" json:"commitAfter,omitempty"`
Amount *paymenttypes.Money `bson:"amount,omitempty" json:"amount,omitempty"`
FromRole *account_role.AccountRole `bson:"fromRole,omitempty" json:"fromRole,omitempty"`
ToRole *account_role.AccountRole `bson:"toRole,omitempty" json:"toRole,omitempty"`
}
// PaymentPlan captures the ordered list of steps to execute a payment.
@@ -304,9 +305,37 @@ type ExecutionStep struct {
SourceWalletRef string `bson:"sourceWalletRef,omitempty" json:"sourceWalletRef,omitempty"`
DestinationRef string `bson:"destinationRef,omitempty" json:"destinationRef,omitempty"`
TransferRef string `bson:"transferRef,omitempty" json:"transferRef,omitempty"`
OperationRef string `bson:"operationRef,omitempty" json:"operationRef,omitempty"`
Error string `bson:"error,omitempty" json:"error,omitempty"`
State OperationState `bson:"state,omitempty" json:"state,omitempty"`
Metadata map[string]string `bson:"metadata,omitempty" json:"metadata,omitempty"`
}
func (s *ExecutionStep) IsTerminal() bool {
if s.State == OperationStateSuccess ||
s.State == OperationStateFailed ||
s.State == OperationStateCancelled ||
s.State == OperationStateSkipped {
return true
}
return false
}
func (s *ExecutionStep) IsSuccess() bool {
return s.State == OperationStateSuccess
}
func (s *ExecutionStep) ReadyForNext() bool {
switch s.State {
case OperationStateSuccess,
OperationStateSkipped:
return true
default:
return false
}
}
// ExecutionPlan captures the ordered list of steps to execute a payment.
type ExecutionPlan struct {
Steps []*ExecutionStep `bson:"steps,omitempty" json:"steps,omitempty"`
@@ -420,7 +449,6 @@ func (p *Payment) Normalize() {
step.CommitPolicy = normalizeCommitPolicy(step.CommitPolicy)
step.DependsOn = normalizeStringList(step.DependsOn)
step.CommitAfter = normalizeStringList(step.CommitAfter)
step.Ref = strings.TrimSpace(step.Ref)
}
}
}

24
api/payments/orchestrator/storage/model/plan_template.go
View File

@@ -4,20 +4,20 @@ import (
"strings"
"github.com/tech/sendico/pkg/db/storable"
pmodel "github.com/tech/sendico/pkg/model"
"github.com/tech/sendico/pkg/model/account_role"
"github.com/tech/sendico/pkg/mservice"
)
// OrchestrationStep defines a template step for execution planning.
type OrchestrationStep struct {
StepID string `bson:"stepId" json:"stepId"`
Rail Rail `bson:"rail" json:"rail"`
Operation string `bson:"operation" json:"operation"`
DependsOn []string `bson:"dependsOn,omitempty" json:"dependsOn,omitempty"`
CommitPolicy CommitPolicy `bson:"commitPolicy,omitempty" json:"commitPolicy,omitempty"`
CommitAfter []string `bson:"commitAfter,omitempty" json:"commitAfter,omitempty"`
FromRole *pmodel.AccountRole `bson:"fromRole,omitempty" json:"fromRole,omitempty"`
ToRole *pmodel.AccountRole `bson:"toRole,omitempty" json:"toRole,omitempty"`
StepID string `bson:"stepId" json:"stepId"`
Rail Rail `bson:"rail" json:"rail"`
Operation string `bson:"operation" json:"operation"`
DependsOn []string `bson:"dependsOn,omitempty" json:"dependsOn,omitempty"`
CommitPolicy CommitPolicy `bson:"commitPolicy,omitempty" json:"commitPolicy,omitempty"`
CommitAfter []string `bson:"commitAfter,omitempty" json:"commitAfter,omitempty"`
FromRole *account_role.AccountRole `bson:"fromRole,omitempty" json:"fromRole,omitempty"`
ToRole *account_role.AccountRole `bson:"toRole,omitempty" json:"toRole,omitempty"`
}
// PaymentPlanTemplate stores reusable orchestration templates.
@@ -60,7 +60,7 @@ func (t *PaymentPlanTemplate) Normalize() {
}
}
func normalizeAccountRole(role *pmodel.AccountRole) *pmodel.AccountRole {
func normalizeAccountRole(role *account_role.AccountRole) *account_role.AccountRole {
if role == nil {
return nil
}
@@ -68,14 +68,14 @@ func normalizeAccountRole(role *pmodel.AccountRole) *pmodel.AccountRole {
if trimmed == "" {
return nil
}
if parsed, ok := pmodel.Parse(trimmed); ok {
if parsed, ok := account_role.Parse(trimmed); ok {
if parsed == "" {
return nil
}
normalized := parsed
return &normalized
}
normalized := pmodel.AccountRole(strings.ToLower(trimmed))
normalized := account_role.AccountRole(strings.ToLower(trimmed))
return &normalized
}