payments/internal/connectors/engine/workflow/utils.go

package workflow

import (
	"errors"
	"fmt"
	"math/big"

	"github.com/hanzo-fi/payments/internal/connectors/engine/activities"
	"github.com/hanzo-fi/payments/pkg/domain/models"
	"github.com/hanzo-fi/payments/internal/storage"
	"github.com/google/uuid"
	"go.temporal.io/sdk/temporal"
	"go.temporal.io/sdk/workflow"
)

func (w Workflow) SearchAttributes(ctx workflow.Context, connectorID *models.ConnectorID) map[string]interface{} {
	m := map[string]interface{}{
		SearchAttributeStack: w.stack,
	}
	if IsConnectorIDSearchAttributeEnabled(ctx) && connectorID != nil {
		m[SearchAttributeConnectorID] = connectorID.String()
	}
	return m
}

func (w Workflow) ScheduleSearchAttributes(ctx workflow.Context, connectorID *models.ConnectorID, scheduleID string) map[string]interface{} {
	m := w.SearchAttributes(ctx, connectorID)
	if scheduleID != "" {
		m[SearchAttributeScheduleID] = scheduleID
	}
	return m
}

func (w Workflow) storePIPaymentWithStatus(
	ctx workflow.Context,
	payment models.Payment,
	paymentInitiationID models.PaymentInitiationID,
	status models.PaymentInitiationAdjustmentStatus,
) error {
	// payment is available, storing it
	err := activities.StoragePaymentsStore(
		infiniteRetryContext(ctx),
		[]models.Payment{payment},
	)
	if err != nil {
		return err
	}

	err = activities.StoragePaymentInitiationsRelatedPaymentsStore(
		infiniteRetryContext(ctx),
		paymentInitiationID,
		payment.ID,
		payment.CreatedAt,
	)
	if err != nil {
		return err
	}

	// Payment events are now sent via outbox pattern in PaymentsUpsert

	err = w.addPIAdjustment(
		ctx,
		models.PaymentInitiationAdjustmentID{
			PaymentInitiationID: paymentInitiationID,
			CreatedAt:           workflow.Now(ctx),
			Status:              status,
		},
		payment.Amount,
		&payment.Asset,
		nil,
		nil,
	)
	if err != nil {
		return err
	}
	return nil
}

func (w Workflow) addPIAdjustment(
	ctx workflow.Context,
	adjustmentID models.PaymentInitiationAdjustmentID,
	amount *big.Int,
	asset *string,
	err error,
	metadata map[string]string,
) error {
	adj := models.PaymentInitiationAdjustment{
		ID:        adjustmentID,
		CreatedAt: workflow.Now(ctx),
		Status:    adjustmentID.Status,
		Amount:    amount,
		Asset:     asset,
		Error:     err,
		Metadata:  metadata,
	}

	if err := activities.StoragePaymentInitiationsAdjustmentsStore(
		infiniteRetryContext(ctx),
		adj,
	); err != nil {
		return err
	}

	// Payment initiation adjustment events are now sent via outbox pattern in PaymentInitiationAdjustmentsUpsert
	return nil
}

func (w Workflow) getPSPPI(
	ctx workflow.Context,
	pi *models.PaymentInitiation,
) (models.PSPPaymentInitiation, error) {
	var sourceAccount *models.Account
	if pi.SourceAccountID != nil {
		var err error
		sourceAccount, err = activities.StorageAccountsGet(
			infiniteRetryContext(ctx),
			*pi.SourceAccountID,
		)
		if err != nil {
			return models.PSPPaymentInitiation{}, err
		}
	}

	var destinationAccount *models.Account
	if pi.DestinationAccountID != nil {
		var err error
		destinationAccount, err = activities.StorageAccountsGet(
			infiniteRetryContext(ctx),
			*pi.DestinationAccountID,
		)
		if err != nil {
			return models.PSPPaymentInitiation{}, err
		}

		if destinationAccount.Type == models.ACCOUNT_TYPE_EXTERNAL {
			if err := fillFormanceBankAccount(ctx, destinationAccount); err != nil {
				return models.PSPPaymentInitiation{}, err
			}
		}
	}
	pspPI := models.FromPaymentInitiationToPSPPaymentInitiation(pi, models.ToPSPAccount(sourceAccount), models.ToPSPAccount(destinationAccount))
	return pspPI, nil
}

func fillFormanceBankAccount(
	ctx workflow.Context,
	account *models.Account,
) error {
	bankAccountUUID, err := uuid.Parse(account.ID.Reference)
	if err != nil {
		// Not an uuid, so cannot be a formance bank account
		return nil
	}

	bankAccount, err := activities.StorageBankAccountsGet(
		infiniteRetryContext(ctx),
		bankAccountUUID,
		true,
	)
	if err != nil {
		if isStorageNotFoundError(err) {
			return nil
		}
		return err
	}

	models.FillBankAccountDetailsToAccountMetadata(account, bankAccount)

	return nil
}

// isStorageNotFoundError checks if an error is a storage "not found" error.
// This handles both direct errors and temporal application errors (which lose
// their original error type when serialized across the activity boundary).
func isStorageNotFoundError(err error) bool {
	if errors.Is(err, storage.ErrNotFound) {
		return true
	}
	var appErr *temporal.ApplicationError
	if errors.As(err, &appErr) && appErr.Type() == activities.ErrTypeStorage {
		return errors.Is(appErr.Unwrap(), storage.ErrNotFound) || appErr.Message() == storage.ErrNotFound.Error()
	}
	return false
}

func getPIStatusFromPayment(status models.PaymentStatus) models.PaymentInitiationAdjustmentStatus {
	switch status {
	case models.PAYMENT_STATUS_SUCCEEDED,
		models.PAYMENT_STATUS_CAPTURE,
		models.PAYMENT_STATUS_REFUND_REVERSED:
		return models.PAYMENT_INITIATION_ADJUSTMENT_STATUS_PROCESSED
	case models.PAYMENT_STATUS_CANCELLED,
		models.PAYMENT_STATUS_CAPTURE_FAILED,
		models.PAYMENT_STATUS_FAILED,
		models.PAYMENT_STATUS_EXPIRED:
		return models.PAYMENT_INITIATION_ADJUSTMENT_STATUS_FAILED
	case models.PAYMENT_STATUS_PENDING,
		models.PAYMENT_STATUS_AUTHORISATION:
		return models.PAYMENT_INITIATION_ADJUSTMENT_STATUS_PROCESSING
	case models.PAYMENT_STATUS_REFUNDED:
		return models.PAYMENT_INITIATION_ADJUSTMENT_STATUS_REVERSED
	case models.PAYMENT_STATUS_REFUNDED_FAILURE:
		return models.PAYMENT_INITIATION_ADJUSTMENT_STATUS_REVERSE_FAILED
	default:
		return models.PAYMENT_INITIATION_ADJUSTMENT_STATUS_UNKNOWN
	}
}

func craftUpdatedConnection(
	ctx workflow.Context,
	connectionID string,
	connectorID models.ConnectorID,
	connection *models.OpenBankingConnection,
	updatedStatus models.ConnectionStatus,
	updatedError *string,
) models.OpenBankingConnection {
	now := workflow.Now(ctx)
	updatedConnection := models.OpenBankingConnection{
		ConnectionID: connectionID,
		ConnectorID:  connectorID,
		CreatedAt:    now,
		UpdatedAt:    now,
		Status:       updatedStatus,
		Error:        updatedError,
	}

	if connection != nil {
		updatedConnection.CreatedAt = connection.CreatedAt
		updatedConnection.AccessToken = connection.AccessToken
		updatedConnection.Metadata = connection.Metadata
		updatedConnection.DataUpdatedAt = connection.DataUpdatedAt
	}

	return updatedConnection
}

func (w Workflow) getDefaultTaskQueue() string {
	return fmt.Sprintf("%s-default", w.stack)
}